Netty Event loop benchmark

NettyGroupBenchmark.java

package dev.kofemann.playground;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
import org.openjdk.jmh.annotations.Measurement;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.TearDown;
import org.openjdk.jmh.annotations.Threads;
import org.openjdk.jmh.annotations.Warmup;
import org.openjdk.jmh.runner.Runner;
import org.openjdk.jmh.runner.RunnerException;
import org.openjdk.jmh.runner.options.Options;
import org.openjdk.jmh.runner.options.OptionsBuilder;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.concurrent.ThreadLocalRandom;

@State(Scope.Benchmark)
@BenchmarkMode(Mode.Throughput)
public class NettyGroupBenchmark {

    @Param({"epoll", "nio"})
    String type;

    private static class EchoHandler extends ChannelInboundHandlerAdapter {

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf buf = (ByteBuf) msg;

            ctx.write(buf);
            ctx.flush();
        }
    }

    private static class EchoServiceInitializer extends ChannelInitializer<SocketChannel> {
        @Override
        public void initChannel(SocketChannel channel) throws Exception {
            channel.pipeline().addLast(new EchoHandler());
        }
    }

    private ServerBootstrap server;

    private EventLoopGroup masterGroup;
    private EventLoopGroup workerGroup;
    private Class<? extends ServerSocketChannel> channelType;

    private final byte[] data = new byte[64*1024];

    private InetSocketAddress sa;
    private ThreadLocal<ByteBuffer> bufs = ThreadLocal.withInitial(() -> ByteBuffer.allocate(data.length));

    private ThreadLocal<java.nio.channels.SocketChannel> sockets = ThreadLocal.withInitial(
            () -> {
                try {
                    var socket = java.nio.channels.SocketChannel.open();
                    socket.connect(new InetSocketAddress("localhost", sa.getPort()));
                    return socket;
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
    );


    @Setup
    public void setUp() throws InterruptedException, IOException {

        switch (type) {
            case "epoll":
                masterGroup = new EpollEventLoopGroup();
                workerGroup = new EpollEventLoopGroup();
                channelType = EpollServerSocketChannel.class;
                break;
            case "nio":
                masterGroup = new NioEventLoopGroup();
                workerGroup = new NioEventLoopGroup();
                channelType = NioServerSocketChannel.class;
                break;
            default:
                throw new RuntimeException();
        }

        ThreadLocalRandom.current().nextBytes(data);

        server = new ServerBootstrap();
        server.group(masterGroup, workerGroup).channel(channelType).childHandler(new EchoServiceInitializer());

        ChannelFuture future = server.bind(0).sync();
        sa = (InetSocketAddress) future.channel().localAddress();
    }

    @TearDown
    public void shutdown() throws InterruptedException, IOException {
        masterGroup.shutdownGracefully().sync();
        workerGroup.shutdownGracefully().sync();
    }

    @Benchmark
    @Fork(value = 2)
    @Measurement(iterations = 10, time = 5)
    @Warmup(iterations = 5, time = 1)
    @Threads(8)
    public long sendMessage() throws IOException {

        var b = ByteBuffer.wrap(data);
        b.limit(b.array().length);
        sockets.get().write(b);

        b = bufs.get();
        b.clear();
        b.limit(b.capacity());
        return sockets.get().read(b);
    }

    public static void main(String[] args) throws RunnerException {
        Options opt = new OptionsBuilder().include(NettyGroupBenchmark.class.getSimpleName()).build();

        new Runner(opt).run();
    }
}