Repository: netty/netty
Issue: Upgrade to JCTools v2.0, use the Unbounded MPSC to replace Chunked
Labels: feature


Issue comments:
@nitsanw can you explain what the `advantages` of the new impl is ?

IIUC Netty originally had an unbounded linked queue here. It then moved to using Chunked which is bounded and just set a very very high boundary on max size. Unbounded should offer a slightly faster and simpler resize logic, and a more accurate representation of the original intention.
Also note that this release fixes an issue with the Growable variant, but I don't think Netty uses it.


Pull request review comments:
I plan to open an issue for these but wanted to discuss first.
@nitsanw - FYI. I didn't use `MpscLinkedQueue` in this PR for GC concerns. I have not done any evaluation ... we can do this as a followup PR if necessary. I did want to upgrade to the latest version of JCTools though.
+1
+1
@Scottmitch why not do this in `PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);` itself ?
see above
we could however `MAX_VALUE` may not necessarily mean "unlimited" in practice. For these particular use cases that is the intention, but the goal was `newMpscQueue(int)` would enforce a limit while `newMpscQueue()` isn't required to enforce a limit.

Patch that resolved the issue:
diff --git a/common/src/main/java/io/netty/util/internal/PlatformDependent.java b/common/src/main/java/io/netty/util/internal/PlatformDependent.java
index 2fd3c985a94..e56ba5817cc 100644
--- a/common/src/main/java/io/netty/util/internal/PlatformDependent.java
+++ b/common/src/main/java/io/netty/util/internal/PlatformDependent.java
@@ -19,9 +19,11 @@
 import io.netty.util.internal.logging.InternalLoggerFactory;
 import org.jctools.queues.MpscArrayQueue;
 import org.jctools.queues.MpscChunkedArrayQueue;
+import org.jctools.queues.MpscUnboundedArrayQueue;
 import org.jctools.queues.SpscLinkedQueue;
 import org.jctools.queues.atomic.MpscAtomicArrayQueue;
-import org.jctools.queues.atomic.MpscLinkedAtomicQueue;
+import org.jctools.queues.atomic.MpscGrowableAtomicArrayQueue;
+import org.jctools.queues.atomic.MpscUnboundedAtomicArrayQueue;
 import org.jctools.queues.atomic.SpscLinkedAtomicQueue;
 import org.jctools.util.Pow2;
 import org.jctools.util.UnsafeAccess;
@@ -51,6 +53,8 @@
 import static io.netty.util.internal.PlatformDependent0.HASH_CODE_C2;
 import static io.netty.util.internal.PlatformDependent0.hashCodeAsciiSanitize;
 import static io.netty.util.internal.PlatformDependent0.unalignedAccess;
+import static java.lang.Math.max;
+import static java.lang.Math.min;
 
 /**
  * Utility that detects various properties specific to the current runtime
@@ -79,7 +83,6 @@ public final class PlatformDependent {
 
     private static final int MPSC_CHUNK_SIZE =  1024;
     private static final int MIN_MAX_MPSC_CAPACITY =  MPSC_CHUNK_SIZE * 2;
-    private static final int DEFAULT_MAX_MPSC_CAPACITY =  MPSC_CHUNK_SIZE * MPSC_CHUNK_SIZE;
     private static final int MAX_ALLOWED_MPSC_CAPACITY = Pow2.MAX_POW2;
 
     private static final long BYTE_ARRAY_BASE_OFFSET = byteArrayBaseOffset0();
@@ -826,25 +829,27 @@ public Object run() {
         }
 
         static <T> Queue<T> newMpscQueue(final int maxCapacity) {
-            if (USE_MPSC_CHUNKED_ARRAY_QUEUE) {
-                // Calculate the max capacity which can not be bigger then MAX_ALLOWED_MPSC_CAPACITY.
-                // This is forced by the MpscChunkedArrayQueue implementation as will try to round it
-                // up to the next power of two and so will overflow otherwise.
-                final int capacity =
-                        Math.max(Math.min(maxCapacity, MAX_ALLOWED_MPSC_CAPACITY), MIN_MAX_MPSC_CAPACITY);
-                return new MpscChunkedArrayQueue<T>(MPSC_CHUNK_SIZE, capacity);
-            } else {
-                return new MpscLinkedAtomicQueue<T>();
-            }
+            // Calculate the max capacity which can not be bigger then MAX_ALLOWED_MPSC_CAPACITY.
+            // This is forced by the MpscChunkedArrayQueue implementation as will try to round it
+            // up to the next power of two and so will overflow otherwise.
+            final int capacity = max(min(maxCapacity, MAX_ALLOWED_MPSC_CAPACITY), MIN_MAX_MPSC_CAPACITY);
+            return USE_MPSC_CHUNKED_ARRAY_QUEUE ? new MpscChunkedArrayQueue<T>(MPSC_CHUNK_SIZE, capacity)
+                                                : new MpscGrowableAtomicArrayQueue<T>(MPSC_CHUNK_SIZE, capacity);
+        }
+
+        static <T> Queue<T> newMpscQueue() {
+            return USE_MPSC_CHUNKED_ARRAY_QUEUE ? new MpscUnboundedArrayQueue<T>(MPSC_CHUNK_SIZE)
+                                                : new MpscUnboundedAtomicArrayQueue<T>(MPSC_CHUNK_SIZE);
         }
     }
 
     /**
      * Create a new {@link Queue} which is safe to use for multiple producers (different threads) and a single
      * consumer (one thread!).
+     * @return A MPSC queue which may be unbounded.
      */
     public static <T> Queue<T> newMpscQueue() {
-        return newMpscQueue(DEFAULT_MAX_MPSC_CAPACITY);
+        return Mpsc.newMpscQueue();
     }
 
     /**
diff --git a/pom.xml b/pom.xml
index 4be0e4421be..4deabc18bc8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -370,7 +370,7 @@
       <dependency>
         <groupId>org.jctools</groupId>
         <artifactId>jctools-core</artifactId>
-        <version>2.0.1</version>
+        <version>2.0.2</version>
       </dependency>
 
       <dependency>
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoop.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoop.java
index 909088fde4b..33e0dc72588 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoop.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoop.java
@@ -171,7 +171,8 @@ void remove(AbstractEpollChannel ch) throws IOException {
     @Override
     protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
         // This event loop never calls takeTask()
-        return PlatformDependent.newMpscQueue(maxPendingTasks);
+        return maxPendingTasks == Integer.MAX_VALUE ? PlatformDependent.<Runnable>newMpscQueue()
+                                                    : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
     }
 
     @Override
diff --git a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueEventLoop.java b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueEventLoop.java
index fea086693b7..ccb91173ee1 100644
--- a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueEventLoop.java
+++ b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueEventLoop.java
@@ -298,7 +298,8 @@ protected void run() {
     @Override
     protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
         // This event loop never calls takeTask()
-        return PlatformDependent.newMpscQueue(maxPendingTasks);
+        return maxPendingTasks == Integer.MAX_VALUE ? PlatformDependent.<Runnable>newMpscQueue()
+                                                    : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
     }
 
     @Override
diff --git a/transport/src/main/java/io/netty/channel/nio/NioEventLoop.java b/transport/src/main/java/io/netty/channel/nio/NioEventLoop.java
index 3de80fca120..331182e0e77 100644
--- a/transport/src/main/java/io/netty/channel/nio/NioEventLoop.java
+++ b/transport/src/main/java/io/netty/channel/nio/NioEventLoop.java
@@ -256,7 +256,8 @@ public SelectorProvider selectorProvider() {
     @Override
     protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
         // This event loop never calls takeTask()
-        return PlatformDependent.newMpscQueue(maxPendingTasks);
+        return maxPendingTasks == Integer.MAX_VALUE ? PlatformDependent.<Runnable>newMpscQueue()
+                                                    : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
     }
 
     @Override


Test patch: