Introduction

Recently, I implemented a UDP packet processing feature using Netty for statistical analysis of business data. Since Netty's default UDP handling relies on a single thread, unlike TCP's master-slave reactor model that supports multi-threading, I explored whether there were performance optimization techniques available for UDP packet reception.

Linux Kernel 3.9 Feature Impact on Netty Performance

Typically, Netty processes UDP packets through a single NIOEventLoop thread, which means only one socket thread listens on a network port and forwards received data to the application layer. This limitation restricts tuning options to minimizing the time between packet reception and application processing, or scaling horizontally across multiple servers to increase throughput.

However, this isn't necessarily the case. Linux kernel version 3.9 introduced the SO_REUSEPORT feature, enabling multiple socket threads to bind to the same port on a single machine. This allows Netty to handle high-concurrency UDP packets across multiple threads, reducing packet loss and maximizing CPU usage through kernel-level load balancing.

Implementing UDP Port Reuse with Netty on Linux

Adding Netty Dependencies

To use Netty, we first need to add the Maven dependency. Here, we're using version 4.1.58.Final:

<!--netty-->
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.58.Final</version>
</dependency>

Creating the Server Bootstrap Class

We'll create a Netty server bootstrap class. Since Netty defaults to Java NIO, and Linux supports epoll which is more efficient than standard NIO due to its internal event array management, we'll check for epoll support and enable SO_REUSEPORT accordingly.

/**
 * UDP server implementation
 */
@Component
public class UdpServer {

    private static final Logger logger = LoggerFactory.getLogger(UdpServer.class);

    private EventLoopGroup bossGroup;

    private Channel serverChannel;

    /**
     * Initialize Netty server
     */
    public void initialize(int port) {

        logger.info("Epoll.isAvailable():{}", Epoll.isAvailable());

        // Use EpollEventLoopGroup if available, otherwise fallback to NioEventLoopGroup
        bossGroup = Epoll.isAvailable() ? new EpollEventLoopGroup() : new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(bossGroup)
                     .channel(Epoll.isAvailable() ? EpollDatagramChannel.class : NioDatagramChannel.class)
                     .option(ChannelOption.SO_BROADCAST, true)
                     .option(ChannelOption.SO_RCVBUF, 1024 * 1024)
                     .handler(new PacketHandler());

            // Enable SO_REUSEPORT for better performance on Linux
            if (Epoll.isAvailable()) {
                logger.info("Enabling SO_REUSEPORT");
                bootstrap.option(EpollChannelOption.SO_REUSEPORT, true);
            }

            // Create multiple threads for binding to the same port if Epoll is available
            if (Epoll.isAvailable()) {
                int cpuCount = Runtime.getRuntime().availableProcessors();
                logger.info("Using Epoll with SO_REUSEPORT and {} CPUs", cpuCount);
                for (int i = 0; i < cpuCount; i++) {
                    logger.info("Worker-{} binding to port", i);
                    ChannelFuture future = bootstrap.bind(port).sync();
                    if (!future.isSuccess()) {
                        logger.error("Failed to bind on port {}", port);
                        throw new Exception(String.format("Failed to bind on [host = %s , port = %d].", "192.168.2.128", port), future.cause());
                    } else {
                        logger.info("Successfully bound to port {}", port);
                    }
                }
            } else {
                ChannelFuture future = bootstrap.bind(port).sync();
                if (!future.isSuccess()) {
                    logger.error("Failed to bind on port {}", port);
                    throw new Exception(String.format("Failed to bind on [host = %s , port = %d].", "127.0.0.1", port), future.cause());
                } else {
                    logger.info("Successfully bound to port {}", port);
                }
            }

        } catch (Exception e) {
            logger.error("Error occurred: {}", e.getMessage(), e);
        }
    }
}

Since this is part of a Spring Boot project, we also need a startup task:

@Component
public class StartupTask implements CommandLineRunner {

    private static final Logger logger = LoggerFactory.getLogger(StartupTask.class);

    @Autowired
    private UdpServer udpServer;

    @Override
    public void run(String... args) {
        logger.info("Initializing Netty server on port: {}", 7000);
        udpServer.initialize(7000);
    }
}

Packet Processing Handler

The handler simply logs received packets and increments a counter for performance monitoring:

/**
 * UDP packet handler
 */
@Component
@ChannelHandler.Sharable
public class PacketHandler extends SimpleChannelInboundHandler<DatagramPacket> {

    private static final Logger logger = LoggerFactory.getLogger(PacketHandler.class);

    private static final AtomicInteger packetCounter = new AtomicInteger(0);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) {
        try {
            int size = packet.content().readableBytes();
            byte[] buffer = new byte[size];
            packet.content().getBytes(packet.content().readerIndex(), buffer);
            logger.info("Received UDP packet: {} Packet count: {}", new String(buffer), packetCounter.incrementAndGet());
        } catch (Exception e) {
            logger.error("Failed to process packet: {}", e.getMessage(), e);
        }
    }
}

Load Testing with JMeter

After deployment, we can use JMeter to send 1 million UDP packets to verify that no packets are lost. The test results confirm successful receipt of all packets.

To demonstrate the performance gain, we can compare with a single-threaded approach by reverting to the older code without SO_REUSEPORT:

public void initialize(int port) {
    logger.info("Epoll.isAvailable():{}", Epoll.isAvailable());

    bossGroup = Epoll.isAvailable() ? new EpollEventLoopGroup() : new NioEventLoopGroup();

    try {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(bossGroup)
                 .channel(Epoll.isAvailable() ? EpollDatagramChannel.class : NioDatagramChannel.class)
                 .option(ChannelOption.SO_BROADCAST, true)
                 .option(ChannelOption.SO_RCVBUF, 1024 * 1024)
                 .handler(new PacketHandler());

        ChannelFuture future = bootstrap.bind(port).sync();
        if (!future.isSuccess()) {
            logger.error("Failed to bind on port {}", port);
            throw new Exception(String.format("Failed to bind on [host = %s , port = %d].", "127.0.0.1", port), future.cause());
        } else {
            logger.info("Successfully bound to port {}", port);
        }
    } catch (Exception e) {
        logger.error("Error occurred: {}", e.getMessage(), e);
    }
}

The comparison shows that the single-threaded version expereinces packet loss under high concurrency. Therefore, for high-load UDP services with Netty, leveraging the SO_REUSEPORT feature on Linux is highly recommended.

Summary

This article demonstrates how to enhance Netty-based UDP service performance on Linux by utilizing the SO_REUSEPORT kernel feature. By distributing incoming packets across multiple threads, we achieve better throughput and reduced packet loss.

References

• Linux Netty High-Performance UDP Service (SO_REUSEPORT): https://blog.csdn.net/monokai/article/details/108453746
• Netty Network Transmission Notes: https://www.sharkchili.com/pages/710071/#前言