Move SegmentedDatagramPacket to transport-native-unix-common (#11121)

Motivation:

As we can supported SegmentedDatagramPacket in multiple native
transports (like in epoll and io_uring) we should just move it to
unix-common so we can share code.

Modification:

- Move SegmentedDatagrampPacket to transport-native-unixu
- Mark the SegmentedDatagramPacket in epoll as deprecated
- Update code to use updated package.

Result:

Possibility of code re-use

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

@@ -67,6 +67,17 @@ public final class EpollDatagramChannel extends AbstractEpollChannel implements
     private final EpollDatagramChannelConfig config;
     private volatile boolean connected;
 
+    /**
+     * Returns {@code true} if {@link io.netty.channel.unix.SegmentedDatagramPacket} is supported natively.
+     *
+     * @return {@code true} if supported, {@code false} otherwise.
+     */
+    public static boolean isSegmentedDatagramPacketSupported() {
+        return Epoll.isAvailable() &&
+                // We only support it together with sendmmsg(...)
+                Native.IS_SUPPORTING_SENDMMSG && Native.IS_SUPPORTING_UDP_SEGMENT;
+    }
+
     /**
      * Create a new instance which selects the {@link InternetProtocolFamily} to use depending
      * on the Operation Systems default which will be chosen.
@@ -299,7 +310,7 @@ public final class EpollDatagramChannel extends AbstractEpollChannel implements
                 // Check if sendmmsg(...) is supported which is only the case for GLIBC 2.14+
                 if (Native.IS_SUPPORTING_SENDMMSG && in.size() > 1 ||
                         // We only handle UDP_SEGMENT in sendmmsg.
-                        in.current() instanceof SegmentedDatagramPacket) {
+                        in.current() instanceof io.netty.channel.unix.SegmentedDatagramPacket) {
                     NativeDatagramPacketArray array = cleanDatagramPacketArray();
                     array.add(in, isConnected(), maxMessagesPerWrite);
                     int cnt = array.count();
@@ -377,12 +388,12 @@ public final class EpollDatagramChannel extends AbstractEpollChannel implements
 
     @Override
     protected Object filterOutboundMessage(Object msg) {
-        if (msg instanceof SegmentedDatagramPacket) {
+        if (msg instanceof io.netty.channel.unix.SegmentedDatagramPacket) {
             if (!Native.IS_SUPPORTING_UDP_SEGMENT) {
                 throw new UnsupportedOperationException(
                         "unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
             }
-            SegmentedDatagramPacket packet = (SegmentedDatagramPacket) msg;
+            io.netty.channel.unix.SegmentedDatagramPacket packet = (io.netty.channel.unix.SegmentedDatagramPacket) msg;
             ByteBuf content = packet.content();
             return UnixChannelUtil.isBufferCopyNeededForWrite(content) ?
                     packet.replace(newDirectBuffer(packet, content)) : msg;
@@ -569,8 +580,9 @@ public final class EpollDatagramChannel extends AbstractEpollChannel implements
 
     private static void addDatagramPacketToOut(DatagramPacket packet,
                                               RecyclableArrayList out) {
-        if (packet instanceof SegmentedDatagramPacket) {
+        if (packet instanceof io.netty.channel.unix.SegmentedDatagramPacket) {
-            SegmentedDatagramPacket segmentedDatagramPacket = (SegmentedDatagramPacket) packet;
+            io.netty.channel.unix.SegmentedDatagramPacket segmentedDatagramPacket =
+                    (io.netty.channel.unix.SegmentedDatagramPacket) packet;
             ByteBuf content = segmentedDatagramPacket.content();
             InetSocketAddress recipient = segmentedDatagramPacket.recipient();
             InetSocketAddress sender = segmentedDatagramPacket.sender();
@@ -636,7 +648,7 @@ public final class EpollDatagramChannel extends AbstractEpollChannel implements
             byteBuf.writerIndex(bytesReceived);
             InetSocketAddress local = localAddress();
             DatagramPacket packet = msg.newDatagramPacket(byteBuf, local);
-            if (!(packet instanceof SegmentedDatagramPacket)) {
+            if (!(packet instanceof io.netty.channel.unix.SegmentedDatagramPacket)) {
                 processPacket(pipeline(), allocHandle, bytesReceived, packet);
                 byteBuf = null;
             } else {
@@ -686,7 +698,7 @@ public final class EpollDatagramChannel extends AbstractEpollChannel implements
             if (received == 1) {
                 // Single packet fast-path
                 DatagramPacket packet = packets[0].newDatagramPacket(byteBuf, local);
-                if (!(packet instanceof SegmentedDatagramPacket)) {
+                if (!(packet instanceof io.netty.channel.unix.SegmentedDatagramPacket)) {
                     processPacket(pipeline(), allocHandle, datagramSize, packet);
                     byteBuf = null;
                     return true;

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

@@ -120,8 +120,8 @@ final class NativeDatagramPacketArray {
                 DatagramPacket packet = (DatagramPacket) msg;
                 ByteBuf buf = packet.content();
                 int segmentSize = 0;
-                if (packet instanceof SegmentedDatagramPacket) {
+                if (packet instanceof io.netty.channel.unix.SegmentedDatagramPacket) {
-                    int seg = ((SegmentedDatagramPacket) packet).segmentSize();
+                    int seg = ((io.netty.channel.unix.SegmentedDatagramPacket) packet).segmentSize();
                     // We only need to tell the kernel that we want to use UDP_SEGMENT if there are multiple
                     // segments in the packet.
                     if (buf.readableBytes() > seg) {

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

@@ -16,18 +16,14 @@
 package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.channel.socket.DatagramPacket;
-import io.netty.util.internal.ObjectUtil;
 
 import java.net.InetSocketAddress;
 
 /**
- * Allows to use <a href="https://blog.cloudflare.com/accelerating-udp-packet-transmission-for-quic/">GSO</a>
+ * @deprecated use {@link io.netty.channel.unix.SegmentedDatagramPacket}.
- * if the underlying OS supports it. Before instance and use this class you should check {@link #isSupported()}.
  */
-public final class SegmentedDatagramPacket extends DatagramPacket {
+@Deprecated
-
+public final class SegmentedDatagramPacket extends io.netty.channel.unix.SegmentedDatagramPacket {
-    private final int segmentSize;
 
     /**
      * Create a new instance.
@@ -37,9 +33,8 @@ public final class SegmentedDatagramPacket extends DatagramPacket {
      * @param recipient     the recipient.
      */
     public SegmentedDatagramPacket(ByteBuf data, int segmentSize, InetSocketAddress recipient) {
-        super(data, recipient);
+        super(data, segmentSize, recipient);
         checkIsSupported();
-        this.segmentSize = ObjectUtil.checkPositive(segmentSize, "segmentSize");
     }
 
     /**
@@ -51,9 +46,8 @@ public final class SegmentedDatagramPacket extends DatagramPacket {
      */
     public SegmentedDatagramPacket(ByteBuf data, int segmentSize,
                                    InetSocketAddress recipient, InetSocketAddress sender) {
-        super(data, recipient, sender);
+        super(data, segmentSize, recipient, sender);
         checkIsSupported();
-        this.segmentSize = ObjectUtil.checkPositive(segmentSize, "segmentSize");
     }
 
     /**
@@ -65,33 +59,24 @@ public final class SegmentedDatagramPacket extends DatagramPacket {
                 Native.IS_SUPPORTING_SENDMMSG && Native.IS_SUPPORTING_UDP_SEGMENT;
     }
 
-    /**
-     * Return the size of each segment (the last segment can be smaller).
-     *
-     * @return size of segments.
-     */
-    public int segmentSize() {
-        return segmentSize;
-    }
-
     @Override
     public SegmentedDatagramPacket copy() {
-        return new SegmentedDatagramPacket(content().copy(), segmentSize, recipient(), sender());
+        return new SegmentedDatagramPacket(content().copy(), segmentSize(), recipient(), sender());
     }
 
     @Override
     public SegmentedDatagramPacket duplicate() {
-        return new SegmentedDatagramPacket(content().duplicate(), segmentSize, recipient(), sender());
+        return new SegmentedDatagramPacket(content().duplicate(), segmentSize(), recipient(), sender());
     }
 
     @Override
     public SegmentedDatagramPacket retainedDuplicate() {
-        return new SegmentedDatagramPacket(content().retainedDuplicate(), segmentSize, recipient(), sender());
+        return new SegmentedDatagramPacket(content().retainedDuplicate(), segmentSize(), recipient(), sender());
     }
 
     @Override
     public SegmentedDatagramPacket replace(ByteBuf content) {
-        return new SegmentedDatagramPacket(content, segmentSize, recipient(), sender());
+        return new SegmentedDatagramPacket(content, segmentSize(), recipient(), sender());
     }
 
     @Override

transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollDatagramUnicastTest.java

@@ -99,14 +99,14 @@ public class EpollDatagramUnicastTest extends DatagramUnicastTest {
             // Only supported for the native epoll transport.
             return;
         }
-        Assume.assumeTrue(SegmentedDatagramPacket.isSupported());
+        Assume.assumeTrue(EpollDatagramChannel.isSegmentedDatagramPacketSupported());
         Channel sc = null;
         Channel cc = null;
 
         try {
             cb.handler(new SimpleChannelInboundHandler<Object>() {
                 @Override
-                public void messageReceived(ChannelHandlerContext ctx, Object msgs) throws Exception {
+                public void messageReceived(ChannelHandlerContext ctx, Object msgs) {
                     // Nothing will be sent.
                 }
             });
@@ -148,7 +148,7 @@ public class EpollDatagramUnicastTest extends DatagramUnicastTest {
             } else {
                 buffer = Unpooled.directBuffer(bufferCapacity).writeZero(bufferCapacity);
             }
-            cc.writeAndFlush(new SegmentedDatagramPacket(buffer, segmentSize, addr)).sync();
+            cc.writeAndFlush(new io.netty.channel.unix.SegmentedDatagramPacket(buffer, segmentSize, addr)).sync();
 
             if (!latch.await(10, TimeUnit.SECONDS)) {
                 Throwable error = errorRef.get();

transport-native-unix-common/src/main/java/io/netty/channel/unix/SegmentedDatagramPacket.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.channel.unix;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.socket.DatagramPacket;
+import io.netty.util.internal.ObjectUtil;
+
+import java.net.InetSocketAddress;
+
+/**
+ * Allows to use <a href="https://blog.cloudflare.com/accelerating-udp-packet-transmission-for-quic/">GSO</a>
+ * if the underlying OS supports it. Before using this you should ensure your system support it.
+ */
+public class SegmentedDatagramPacket extends DatagramPacket {
+
+    private final int segmentSize;
+
+    /**
+     * Create a new instance.
+     *
+     * @param data          the {@link ByteBuf} which must be continguous.
+     * @param segmentSize   the segment size.
+     * @param recipient     the recipient.
+     */
+    public SegmentedDatagramPacket(ByteBuf data, int segmentSize, InetSocketAddress recipient) {
+        super(data, recipient);
+        this.segmentSize = ObjectUtil.checkPositive(segmentSize, "segmentSize");
+    }
+
+    /**
+     * Create a new instance.
+     *
+     * @param data          the {@link ByteBuf} which must be continguous.
+     * @param segmentSize   the segment size.
+     * @param recipient     the recipient.
+     */
+    public SegmentedDatagramPacket(ByteBuf data, int segmentSize,
+                                   InetSocketAddress recipient, InetSocketAddress sender) {
+        super(data, recipient, sender);
+        this.segmentSize = ObjectUtil.checkPositive(segmentSize, "segmentSize");
+    }
+
+    /**
+     * Return the size of each segment (the last segment can be smaller).
+     *
+     * @return size of segments.
+     */
+    public int segmentSize() {
+        return segmentSize;
+    }
+
+    @Override
+    public SegmentedDatagramPacket copy() {
+        return new SegmentedDatagramPacket(content().copy(), segmentSize, recipient(), sender());
+    }
+
+    @Override
+    public SegmentedDatagramPacket duplicate() {
+        return new SegmentedDatagramPacket(content().duplicate(), segmentSize, recipient(), sender());
+    }
+
+    @Override
+    public SegmentedDatagramPacket retainedDuplicate() {
+        return new SegmentedDatagramPacket(content().retainedDuplicate(), segmentSize, recipient(), sender());
+    }
+
+    @Override
+    public SegmentedDatagramPacket replace(ByteBuf content) {
+        return new SegmentedDatagramPacket(content, segmentSize, recipient(), sender());
+    }
+
+    @Override
+    public SegmentedDatagramPacket retain() {
+        super.retain();
+        return this;
+    }
+
+    @Override
+    public SegmentedDatagramPacket retain(int increment) {
+        super.retain(increment);
+        return this;
+    }
+
+    @Override
+    public SegmentedDatagramPacket touch() {
+        super.touch();
+        return this;
+    }
+
+    @Override
+    public SegmentedDatagramPacket touch(Object hint) {
+        super.touch(hint);
+        return this;
+    }
+}