Reduce direct memory overhead per EpollEventLoop when using EpollDatagramChannel (#8825)

Motivation:

When using a linux distribution that supports sendmmsg(...) we allocated enough direct memory per EpollEventLoop to be able to write IOV_MAX number of iovecs per message that can be written per sendmmsg.
The number of messages that can be written per sendmmsg(...) call is limited by UIO_MAX_IOV.

In practice this resulted in an allocation of 16MB direct memory per EpollEventLoop instance that stayed allocated until the EpollEventLoop was shutdown which happens as part of the shutdown of the enclosing EpollEVentLoopGroup.

This resulted in quite some heavy direct memory usage in practice even when in practice we have very slim changes to ever need all of the memory.

Modification:

Adjust NativeDatagramPacketArray to share one IovArray instance across all NativeDatagramPacket instances it holds. This limits the max number of iovecs we can write across all messages to IOV_MAX per sendmmsg(...) call.
This in practice will still be enough to allow us to write multiple messages with one syscall while keep the memory overhead to a minimum.

Result:

Smaller direct memory footprint per EpollEventLoop when using EpollDatagramChannel on distributions that support sendmmsg(...).
Fixes https://github.com/netty/netty/issues/8814

transport-native-epoll/src/main/java/io/netty/channel/epoll/NativeDatagramPacketArray.java

@@ -33,6 +33,10 @@ final class NativeDatagramPacketArray implements ChannelOutboundBuffer.MessagePr
 
     // Use UIO_MAX_IOV as this is the maximum number we can write with one sendmmsg(...) call.
     private final NativeDatagramPacket[] packets = new NativeDatagramPacket[UIO_MAX_IOV];
+
+    // We share one IovArray for all NativeDatagramPackets to reduce memory overhead. This will allow us to write
+    // up to IOV_MAX iovec across all messages in one sendmmsg(...) call.
+    private final IovArray iovArray = new IovArray();
     private int count;
 
     NativeDatagramPacketArray() {
@@ -47,6 +51,8 @@ final class NativeDatagramPacketArray implements ChannelOutboundBuffer.MessagePr
      */
     boolean add(DatagramPacket packet) {
         if (count == packets.length) {
+            // We already filled up to UIO_MAX_IOV messages. This is the max allowed per sendmmsg(...) call, we will
+            // try again later.
             return false;
         }
         ByteBuf content = packet.content();
@@ -56,9 +62,13 @@ final class NativeDatagramPacketArray implements ChannelOutboundBuffer.MessagePr
         }
         NativeDatagramPacket p = packets[count];
         InetSocketAddress recipient = packet.recipient();
-        if (!p.init(content, recipient)) {
+
+        int offset = iovArray.count();
+        if (!iovArray.add(content)) {
+            // Not enough space to hold the whole content, we will try again later.
             return false;
         }
+        p.init(iovArray.memoryAddress(offset), iovArray.count() - offset, recipient);
 
         count++;
         return true;
@@ -85,13 +95,11 @@ final class NativeDatagramPacketArray implements ChannelOutboundBuffer.MessagePr
 
     void clear() {
         this.count = 0;
+        this.iovArray.clear();
     }
 
     void release() {
-        // Release all packets
+        iovArray.release();
-        for (NativeDatagramPacket datagramPacket : packets) {
-            datagramPacket.release();
-        }
     }
 
     /**
@@ -99,10 +107,6 @@ final class NativeDatagramPacketArray implements ChannelOutboundBuffer.MessagePr
      */
     @SuppressWarnings("unused")
     static final class NativeDatagramPacket {
-        // Each NativeDatagramPackets holds a IovArray which is used for gathering writes.
-        // This is ok as NativeDatagramPacketArray is always obtained from an EpollHandler
-        // field so the memory needed is quite small anyway.
-        private final IovArray array = new IovArray();
 
         // This is the actual struct iovec*
         private long memoryAddress;
@@ -112,21 +116,9 @@ final class NativeDatagramPacketArray implements ChannelOutboundBuffer.MessagePr
         private int scopeId;
         private int port;
 
-        private void release() {
+        private void init(long memoryAddress, int count, InetSocketAddress recipient) {
-            array.release();
+            this.memoryAddress = memoryAddress;
-        }
+            this.count = count;
-
-        /**
-         * Init this instance and return {@code true} if the init was successful.
-         */
-        private boolean init(ByteBuf buf, InetSocketAddress recipient) {
-            array.clear();
-            if (!array.add(buf)) {
-                return false;
-            }
-            // always start from offset 0
-            memoryAddress = array.memoryAddress(0);
-            count = array.count();
 
             InetAddress address = recipient.getAddress();
             if (address instanceof Inet6Address) {
@@ -137,7 +129,6 @@ final class NativeDatagramPacketArray implements ChannelOutboundBuffer.MessagePr
                 scopeId = 0;
             }
             port = recipient.getPort();
-            return true;
         }
     }
 }