<commit_before>package io.mio.transport.netty4;

import io.mio.core.transport.Codec;
import io.mio.core.transport.MioServer;
import io.mio.core.commons.*;
import io.mio.core.extension.Extension;
import io.mio.core.extension.ExtensionLoader;
import io.mio.core.extension.TypeReference;
import io.mio.core.transport.ServerConfig;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * NettyMioServer
 *
 * @author lry
 */
@Slf4j
@Extension("netty")
public class NettyMioServer implements MioServer {

    private ServerConfig serverConfig;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;

    private Channel serverChannel;
    private NettyMioServerHandler serverHandler;
    private Codec<ChannelPipeline> codec;

    @Override
    public void initialize(final ServerConfig serverConfig, final MioCallback<MioMessage> mioCallback) {
        this.serverConfig = serverConfig;

        ThreadFactory bossThreadFactory = MioConstants.newThreadFactory("mio-server-boss", true);
        ThreadFactory workerThreadFactory = MioConstants.newThreadFactory("mio-server-worker", true);

        // create handler
        this.serverHandler = new NettyMioServerHandler(serverConfig.getMaxConnections(), mioCallback);

        // create socket channel type and handler
        Class<? extends ServerChannel> channelClass;
        if (serverConfig.isUseLinuxNativeEpoll()) {
            // select socket channel type
            channelClass = EpollServerSocketChannel.class;
            this.bossGroup = new EpollEventLoopGroup(serverConfig.getBossThread(), bossThreadFactory);
            this.workerGroup = new EpollEventLoopGroup(serverConfig.getWorkerThread(), workerThreadFactory);
        } else {
            channelClass = NioServerSocketChannel.class;
            this.bossGroup = new NioEventLoopGroup(serverConfig.getBossThread(), bossThreadFactory);
            this.workerGroup = new NioEventLoopGroup(serverConfig.getWorkerThread(), workerThreadFactory);
        }

        // create codec
        this.codec = ExtensionLoader.getLoader(new TypeReference<Codec<ChannelPipeline>>() {
        }).getExtension(serverConfig.getCodec());

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(channelClass)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    // Set TCP buffer size
                    .option(ChannelOption.SO_BACKLOG, serverConfig.getBacklog())
                    .childOption(ChannelOption.SO_KEEPALIVE, serverConfig.isTcpKeepalive())
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childOption(ChannelOption.SO_REUSEADDR, true)
                    // Using object pools, reusing buffers
                    .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .childHandler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel ch) throws Exception {
                            // server codec
                            codec.server(serverConfig.getMaxContentLength(), ch.pipeline());
                            // heartbeat detection
                            if (serverConfig.getHeartbeat() > 0) {
                                ch.pipeline().addLast(new IdleStateHandler(0, 0,
                                        serverConfig.getHeartbeat(), TimeUnit.MILLISECONDS));
                            }
                            // process network IO
                            ch.pipeline().addLast(serverHandler);
                        }
                    });

            SocketAddress socketAddress = MioConstants.buildSocketAddress(serverConfig.getHostname(), serverConfig.getPort());
            ChannelFuture channelFuture = serverBootstrap.bind(socketAddress);
            channelFuture.syncUninterruptibly();
            serverChannel = channelFuture.channel();

            log.info("The server started success:{}", serverConfig);
            Runtime.getRuntime().addShutdownHook(new Thread(NettyMioServer.this::destroy));
        } catch (Exception e) {
            log.error("The server initialize exception", e);
        }
    }

    @Override
    public void send(MioMessage mioMessage) throws Throwable {<commit_msg>useLinuxNativeEpoll<commit_after>package io.mio.transport.netty4;

import io.mio.core.transport.Codec;
import io.mio.core.transport.MioServer;
import io.mio.core.commons.*;
import io.mio.core.extension.Extension;
import io.mio.core.extension.ExtensionLoader;
import io.mio.core.extension.TypeReference;
import io.mio.core.transport.ServerConfig;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * NettyMioServer
 *
 * @author lry
 */
@Slf4j
@Extension("netty")
public class NettyMioServer implements MioServer {

    private ServerConfig serverConfig;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;

    private Channel serverChannel;
    private NettyMioServerHandler serverHandler;
    private Codec<ChannelPipeline> codec;

    @Override
    public void initialize(final ServerConfig serverConfig, final MioCallback<MioMessage> mioCallback) {
        this.serverConfig = serverConfig;

        ThreadFactory bossThreadFactory = MioConstants.newThreadFactory("mio-server-boss", true);
        ThreadFactory workerThreadFactory = MioConstants.newThreadFactory("mio-server-worker", true);

        // create handler
        this.serverHandler = new NettyMioServerHandler(serverConfig.getMaxConnections(), mioCallback);

        // create socket channel type and handler
        Class<? extends ServerChannel> channelClass;
        if (serverConfig.isUseLinuxNativeEpoll()) {
            // select socket channel type
            channelClass = EpollServerSocketChannel.class;
            this.bossGroup = new EpollEventLoopGroup(serverConfig.getBossThread(), bossThreadFactory);
            this.workerGroup = new EpollEventLoopGroup(serverConfig.getWorkerThread(), workerThreadFactory);
        } else {
            channelClass = NioServerSocketChannel.class;
            this.bossGroup = new NioEventLoopGroup(serverConfig.getBossThread(), bossThreadFactory);
            this.workerGroup = new NioEventLoopGroup(serverConfig.getWorkerThread(), workerThreadFactory);
        }

        // create codec
        this.codec = ExtensionLoader.getLoader(new TypeReference<Codec<ChannelPipeline>>() {
        }).getExtension(serverConfig.getCodec());

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(channelClass)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    // Set TCP buffer size
                    .option(ChannelOption.SO_BACKLOG, serverConfig.getBacklog())
                    .childOption(ChannelOption.SO_KEEPALIVE, serverConfig.isTcpKeepalive())
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childOption(ChannelOption.SO_REUSEADDR, true)
                    // Using object pools, reusing buffers
                    .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .childHandler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel ch) throws Exception {
                            // server codec
                            codec.server(serverConfig.getMaxContentLength(), ch.pipeline());
                            // heartbeat detection
                            if (serverConfig.getHeartbeat() > 0) {
                                ch.pipeline().addLast(new IdleStateHandler(0, 0,
                                        serverConfig.getHeartbeat(), TimeUnit.MILLISECONDS));
                            }
                            // process network IO
                            ch.pipeline().addLast(serverHandler);
                        }
                    });

            SocketAddress socketAddress = MioConstants.buildSocketAddress(serverConfig.getHostname(), serverConfig.getPort());
            ChannelFuture channelFuture = serverBootstrap.bind(socketAddress);
            channelFuture.syncUninterruptibly();
            serverChannel = channelFuture.channel();

            log.info("The server started success:{}", serverConfig);
            Runtime.getRuntime().addShutdownHook(new Thread(NettyMioServer.this::destroy));
        } catch (Exception e) {
            log.error("The server initialize exception", e);
        }
    }

    @Override
    public void send(MioMessage mioMessage) throws Throwable {
{'fraction_non_alphanumeric': 0.08816021990967995, 'fraction_numerical': 0.0013744354997054782, 'mean_word_length': 2.6447227191413236, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4267135', 'n_tokens_mistral': 2642, 'n_tokens_neox': 2574, 'n_words': 468}