<commit_before>package com.firefly.net.tcp.aio;

import com.codahale.metrics.ScheduledReporter;
import com.firefly.net.Config;
import com.firefly.net.event.DefaultNetEvent;
import com.firefly.net.exception.NetException;
import com.firefly.utils.lang.AbstractLifeCycle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.channels.AsynchronousChannelGroup;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinWorkerThread;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author <NAME>
 */
abstract public class AbstractTcpLifeCycle extends AbstractLifeCycle {

    protected static Logger log = LoggerFactory.getLogger("firefly-system");

    protected Config config;
    protected AtomicInteger sessionId = new AtomicInteger();
    protected AsynchronousChannelGroup group;
    protected ExecutorService netExecutorService;
    protected AsynchronousTcpWorker worker;
    protected ScheduledReporter reporter;

    abstract protected String getThreadName();

    public ExecutorService getNetExecutorService() {
        return netExecutorService;
    }

    @Override
    protected void init() {
        if (config == null)
            throw new NetException("server configuration is null");

        try {
            netExecutorService = new ForkJoinPool
                    (config.getAsynchronousCorePoolSize(), pool -> {
                        ForkJoinWorkerThread worker = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
                        worker.setName(getThreadName() + worker.getPoolIndex());
                        return worker;
                    }, null, true);
            group = AsynchronousChannelGroup.withThreadPool(netExecutorService);
            log.info(config.toString());
            worker = new AsynchronousTcpWorker(config, new DefaultNetEvent(config));
            if (config.isMonitorEnable()) {
                reporter = config.getMetricReporterFactory().getScheduledReporter();
                try {
                    reporter.start(10, TimeUnit.SECONDS);
                } catch (Exception e) {
                    log.error("start metric reporter exception -> {}", e.getMessage());
                }
            }
        } catch (IOException e) {
            log.error("initialization server channel group error", e);
        }
    }

    @Override
    protected void destroy() {
        if (group != null) {
            try {
                group.shutdown();
            } catch (Exception e) {
                log.error("aio tcp thread group shutdown exception -> {}", e.getMessage());
            }
        }
        if (config.isMonitorEnable()) {
            try {
                reporter.stop();<commit_msg>[feat]: select thread pool by thread number<commit_after>package com.firefly.net.tcp.aio;

import com.codahale.metrics.ScheduledReporter;
import com.firefly.net.Config;
import com.firefly.net.event.DefaultNetEvent;
import com.firefly.net.exception.NetException;
import com.firefly.utils.lang.AbstractLifeCycle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.channels.AsynchronousChannelGroup;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author <NAME>
 */
abstract public class AbstractTcpLifeCycle extends AbstractLifeCycle {

    protected static Logger log = LoggerFactory.getLogger("firefly-system");

    protected Config config;
    protected AtomicInteger sessionId = new AtomicInteger();
    protected AsynchronousChannelGroup group;
    protected ExecutorService netExecutorService;
    protected AsynchronousTcpWorker worker;
    protected ScheduledReporter reporter;

    abstract protected String getThreadName();

    public ExecutorService getNetExecutorService() {
        return netExecutorService;
    }

    @Override
    protected void init() {
        if (config == null)
            throw new NetException("server configuration is null");

        try {
            if (config.getAsynchronousCorePoolSize() <= 0) {
                ThreadFactory threadFactory = new ThreadFactory() {

                    private AtomicInteger threadId = new AtomicInteger();

                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, getThreadName() + threadId.getAndIncrement());
                    }
                };
                netExecutorService = Executors.newCachedThreadPool(threadFactory);
                log.info("init the cached thread pool");
            } else if (config.getAsynchronousCorePoolSize() <= Config.defaultPoolSize * 2) {
                netExecutorService = new ForkJoinPool
                        (config.getAsynchronousCorePoolSize(), pool -> {
                            ForkJoinWorkerThread worker = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
                            worker.setName(getThreadName() + worker.getPoolIndex());
                            return worker;
                        }, null, true);
                log.info("init the fork join pool");
            } else {
                ThreadFactory threadFactory = new ThreadFactory() {

                    private AtomicInteger threadId = new AtomicInteger();

                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, getThreadName() + threadId.getAndIncrement());
                    }
                };
                netExecutorService = new ThreadPoolExecutor(Config.defaultPoolSize, config.getAsynchronousCorePoolSize(),
                        15L, TimeUnit.SECONDS,
                        new LinkedBlockingQueue<>(),
                        threadFactory);
                log.info("init the fixed size thread pool");
            }
            group = AsynchronousChannelGroup.withThreadPool(netExecutorService);
            log.info(config.toString());
            worker = new AsynchronousTcpWorker(config, new DefaultNetEvent(config));
            if (config.isMonitorEnable()) {
                reporter = config.getMetricReporterFactory().getScheduledReporter();
                try {
                    reporter.start(10, TimeUnit.SECONDS);
                } catch (Exception e) {
                    log.error("start metric reporter exception -> {}", e.getMessage());
                }
            }
        } catch (IOException e) {
            log.error("initialization server channel group error", e);
        }
    }

    @Override
    protected void destroy() {
        if (group != null) {
            try {
                group.shutdown();
            } catch (Exception e) {
                log.error("aio tcp thread group shutdown exception -> {}", e.getMessage());
            }
        }
        if (config.isMonitorEnable()) {
            try {
                reporter.stop();
{'fraction_non_alphanumeric': 0.08959250319466136, 'fraction_numerical': 0.0017038193951441148, 'mean_word_length': 2.1488600804649085, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 12, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2280020', 'n_tokens_mistral': 1748, 'n_tokens_neox': 1677, 'n_words': 383}