Repository: netty/netty
Issue: Support pluggable EventLoop task queue

The default task queue used by EPoll and NIO EventLoop implementations is provided by JCTools, and while it is efficient it has some undesirable properties.  Specifically, it is not non-blocking, and can lead to the EventLoop busy-spinning waiting for a task that may not be provided promptly because the offering thread's execution has been suspended by the operating system.  This is a rare scenario, but it will happen and some application maintainers might like to avoid it.

Unfortunately, there is no simple fix in the use of the queue; the `relaxedPoll()` method appears to be a candidate, but does not provide volatile visibility guarantees to a preceding `offer()`, meaning a wakeup could be missed (although we could write to a dummy volatile variable after offering to provide this).  A queue identical to the JCTools one but using a `putVolatile` to set the item in the queue's backing array would suffice to avoid the busy-spin blocking of the EventLoop, but the issue of future tasks being unreachable until the thread completes would remain.

Ideally, the queue used by these event loops would be pluggable, so that users can choose their ideal point on the spectrum of tradeoffs.

A simple replacement that guarantees progress is the JDK `ConcurrentLinkedQueue`, but at the cost of slightly more expensive `offer()` and `poll()`; the above mentioned modification to the JCTools `offer()` could be mixed with `relaxedPoll()` to simply avoid busy-spin blocking, and a custom Mpsc queue implementation could guarantee forward progress with only marginally higher cost.  Permitting the application maintainer to pick their poison would be tremendously helpful.

Issue comments:
@nitsanw FYI... 
AFAIK the only misbehaviours could happen by using offer of a single-producer JCTools queue or just relaxedPoll instead of poll.
The first one seems not the case here, while on consumer side just using a mix of relaxedPoll()/drain() with poll()/isEmpty() is enough to guarantee strict empty recognition a-la CLQ.
IMO just using poll() could be a simple fix that would force a JCTools queue to use the same semantic of CLQ::poll.
@franz1981 I'm afraid I don't quite follow what you're saying, but we currently _do_ use `poll()` and this has the problems I describe.

However, I have just taken another look at `wakeup()`, and since we _always_ perform a `compareAndSet`, we actually could use `relaxedPoll` to solve the busy-spin issue.  However, we probably _shouldn't_ be doing an unconditional `compareAndSet`, since this is costly - it should ordinarily be guarded by a cheap test of its current value.
@belliottsmith when you say we should guard you think of doing something like this:

```
if (!inEventLoop && !wakenUp.get() && wakenUp.compareAndSet(false, true)) {
    selector.wakeup();
}
```
?
@belliottsmith 

The poll() always return null if there is no pending offering and the only (internal) spinning I'm aware of in JCTools is while awaiting the message slot to be filled by the producer (because JCTools's offer happen in 2 stages: producer sequence increment AND message slot fill).
If the event loop logic fall into a spinning i don't see how it can be related to JCTools unless:

1. the queue used is single producer
2. the event loop logic that perform the sleep/awake is not correct (assuming is using poll() as you've suggested)

I think that the awake logic could use a different strategy too, that would avoid using compareAndSwap on offer side: https://github.com/JCTools/JCTools/blob/master/jctools-experimental/src/main/java/org/jctools/queues/blocking/ScParkTakeStrategy.java#L19



@normanmaurer right
@franz1981 the two stages are not atomic, and a thread can be suspended by the operating system in between these two steps

also, the EventLoop should definitely not be parking; it should only be voluntarily suspending its execution when entering `select()`
@belliottsmith Correct: that means that you would likely to see a spin into the queue::poll call: that's the issue you're seeing? Because I cannot see how using a putVolatile while setting the message slot could make any difference...@nitsanw can confirm/reject it
@belliottsmith I think I would even argue that we may not need the CAS at all and could just use a volatile. At worse we would call `wakeup` multiple times. That said maybe just add a `get` before is good enough as a failed CAS should be at least cheaper then an extra selector wakeup. WDYT ?
@normanmaurer I think the get + CAS is probably safest for now, the problem being the negotiation between the consumer resetting it to `false` before going to sleep, and a producer seeing the `true` before this happens and failing to prevent it going to sleep.  It _would_ be possible to avoid a CAS here, but it would require a bit more complex surgery and thought to modify all of the places we actually go to sleep.  On modern chips a CAS isn't so costly anyway, and probably less costly than actually triggering the `wakeup` (though I don't know how actually costly this is, the code comments suggest it is costly) - it might be something to consider for 5.0 as part of wider refactors, but doesn't intuitively feel like it fits the risk/reward tradeoff for 4.x to me.
@belliottsmith @normanmaurer I think this can be closed now that #9247 is merged?
@njhill yes!
@belliottsmith "Specifically, it is not non-blocking, and can lead to the EventLoop busy-spinning waiting for a task that may not be provided promptly because the offering thread's execution has been suspended by the operating system." - I assume you are referring to the following:
https://github.com/JCTools/JCTools/blob/58a08a14928ca966d55543d3e4f6f40a44a65518/jctools-core/src/main/java/org/jctools/queues/BaseMpscLinkedArrayQueue.java#L265

```
            if (casProducerIndex(pIndex, pIndex + 2))
            {
                break;
            }
        }
        // INDEX visible before ELEMENT
/* Producer thread is suspended here */
        final long offset = modifiedCalcElementOffset(pIndex, mask);
        soElement(buffer, offset, e); // release element e
        return true;
```

If this happens the consumer thread will spin-wait for the element to appear in `poll()`:

```
        Object e = lvElement(buffer, offset);// LoadLoad
        if (e == null)
        {
            if (index != lvProducerIndex())
            {
/* index is visible, but element is not visible until the producer is able to make progress */
                do
                {
                    e = lvElement(buffer, offset);
                }
                while (e == null);
            }
            else
            {
                return null;
            }
        }
```

This makes the queue blocking. What is worse, successful `offer` calls from other producers cannot overtake the "stuck producer" and the consumer cannot attend to them (I've always thought of this as a bubble in the queue). The producers can make progress, but the consumer can't skip ahead and deq the items until the "stuck producer" releases it.

If this is the issue under discussion, I'm not sure what the following suggestion means: "A queue identical to the JCTools one but using a putVolatile to set the item in the queue's backing array would suffice to avoid the busy-spin blocking of the EventLoop"

The problem is that the index is visible before the element. Making the element "more visible" (by using a stronger barrier) will not make a difference. The index and the element are not atomically published (as @belliottsmith points out later). Using CLQ will provide better progress guarantees here AFAIK (IIRC the node is only published when fully initialized, so the "bubble" in the queue cannot happen) and I think the feature request is sensible.
@belliottsmith @franz1981 reading through the discussion and the java doc I realize JCTools should do a better job documenting the progress guarantees on this and other queues. I've filed a bug: https://github.com/JCTools/JCTools/issues/259
> If this is the issue under discussion, I'm not sure what the following suggestion means: "A queue identical to the JCTools one but using a putVolatile to set the item in the queue's backing array would suffice to avoid the busy-spin blocking of the EventLoop"

It means the `eventLoop` (by using `relaxedPoll()`) would at least be able to proceed with other work such as reading/writing to a socket, or could stop if it had no other work to do, freeing up a thread for the producer to complete its work.

Obviously a preferable solution is to use a truly non-blocking queue, but blocking the `eventLoop` is worse than just blocking the external work queue.
I was referring in particular to: "using a **putVolatile** to set the item in the queue's backing array would suffice to avoid the busy-spin blocking of the EventLoop".
I agree that using a `relaxedPoll` will return control to the EventLoop, giving you more options in how to utilize time until an element is visible. I don't see how `putVolatile` makes a difference here.
Thanks.
Well, in combination with #9109 you might appreciate my reasoning.  Long story short, it depends on the behaviour of the code that wraps it to provide blocking behaviour.  `putVolatile` instead of `putOrdered` would permit negotiating if a wake-up is _not_ necessary with only `LoadLoad` barriers.  However, _today_, Netty imposes a full CAS in each side of that transaction, so it wouldn't actually be necessary.

(It's been a while since I had the context to think about this, so I may bow out of further discussion for now, as I don't really have time to re-research my answers/thoughts)
Reading through the discussion, I understand you have some reservations regarding the semantics/guarantees of JCTools queues, but I can't see how they add up to anything actionable.

If you have a concern with a particular detail please file a bug with JCTools and I'll be happy to address it.

Making the element write a volatile write makes no difference to either ticket AFAICT. It imposes further ordering constraints on the offer, which are irrelevant to the offer and do not help the `poll` code where these tickets seems to focus. The `offer` method as a whole already has volatile write semantics (in this implementation) due to the successful CAS operation.
I have no concerns about the semantics, that would imply I thought they were overall problematic.  Every concurrent structure's semantics matter only in the context they're used.

IMO, this project should be using a truly non-blocking queue, but that doesn't mean the algorithm employed by JCTools is of general concern.  

(I will concede that I consider the default spin-lock blocking behaviour to be a bug, for any real-time application, but that's a personal stance on the matter of unbounded user-space spin locks and priority inversion)
"IMO, this project should be using a truly non-blocking queue" - The trade offs are between higher per-element costs (allocation + barriers) and the JCTools option which is lock-less but still risks blocking. The configurable queue impl solves this issue and offers choice, so I think this is solved.

"I consider the default spin-lock blocking behaviour to be a bug" - Would you prefer to put a yield in the loop instead? If you have suggestion I'm happy to hear it.



Given that a relaxed poll can return NULL either due to emptiness and a "not yet" published element I believe that it can be improved by make these 2 cases explicit and let a user to choose what to do while awaiting element publication, but it would mean an additional xxxPoll method: it could be useful, but most JCtools users love and uses it because qs are "kinda" replacement for j.u.c. queues and I'm not sure that Netty would love to be bounded to use MessagePassingQueue API instead of the more general java Queue one.
I've been down enough of these rabbit holes to know any answer will only lead to another, so I hope you don't mind if I bow out for real for now.  I only provided that qualifier as a post-script because I felt I had been dishonest by disclaiming _any_ concerns about JCTools in response to your claim I had expressed general concerns (which I do not feel I had, I was only discussing their semantics in the context of this project).  JCTools is a great library, that you should be proud of; it would be surprising if nobody had any concerns or criticisms.  Perhaps we will have some time to discuss them in person one day.
"JCTools is a great library, that you should be proud of" - Thanks :-)

"it would be surprising if nobody had any concerns or criticisms." - Indeed, and I am keen to address these concerns where I can. So please, if possible, make time to submit issues if appropriate. Even if they remain unresolved they will be informing other users of the risks/rewards/tradeoffs/decision making etc.

Thanks for your time, I hope we have an opportunity to discuss in person in the near future.

Pull request review comments:

Patch that resolved the issue:
diff --git a/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java b/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java
index 92b648889e7..a7aee9a4c6b 100644
--- a/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java
+++ b/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java
@@ -167,6 +167,17 @@ protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor
         rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
     }
 
+    protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
+                                        boolean addTaskWakesUp, Queue<Runnable> taskQueue,
+                                        RejectedExecutionHandler rejectedHandler) {
+        super(parent);
+        this.addTaskWakesUp = addTaskWakesUp;
+        this.maxPendingTasks = DEFAULT_MAX_PENDING_EXECUTOR_TASKS;
+        this.executor = ThreadExecutorMap.apply(executor, this);
+        this.taskQueue = ObjectUtil.checkNotNull(taskQueue, "taskQueue");
+        rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
+    }
+
     /**
      * @deprecated Please use and override {@link #newTaskQueue(int)}.
      */
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 d99f4a5d48c..29d71d66a32 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
@@ -17,6 +17,7 @@
 
 import io.netty.channel.EventLoop;
 import io.netty.channel.EventLoopGroup;
+import io.netty.channel.MultithreadEventLoopGroup;
 import io.netty.channel.SelectStrategy;
 import io.netty.channel.SingleThreadEventLoop;
 import io.netty.channel.epoll.AbstractEpollChannel.AbstractEpollUnsafe;
@@ -80,8 +81,10 @@ public int get() throws Exception {
     private static final long MAX_SCHEDULED_TIMERFD_NS = 999999999;
 
     EpollEventLoop(EventLoopGroup parent, Executor executor, int maxEvents,
-                   SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {
-        super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);
+                   SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,
+                   MultithreadEventLoopGroup.EventLoopTaskQueueFactory queueFactory) {
+        super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory),
+                rejectedExecutionHandler);
         selectStrategy = ObjectUtil.checkNotNull(strategy, "strategy");
         if (maxEvents == 0) {
             allowGrowing = true;
@@ -140,6 +143,14 @@ public int get() throws Exception {
         }
     }
 
+    private static Queue<Runnable> newTaskQueue(
+            MultithreadEventLoopGroup.EventLoopTaskQueueFactory queueFactory) {
+        if (queueFactory == null) {
+            return newTaskQueue0(DEFAULT_MAX_PENDING_TASKS);
+        }
+        return queueFactory.newTaskQueue(DEFAULT_MAX_PENDING_TASKS);
+    }
+
     /**
      * Return a cleared {@link IovArray} that can be used for writes in this {@link EventLoop}.
      */
@@ -217,9 +228,13 @@ void remove(AbstractEpollChannel ch) throws IOException {
 
     @Override
     protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
+        return newTaskQueue0(maxPendingTasks);
+    }
+
+    private static Queue<Runnable> newTaskQueue0(int maxPendingTasks) {
         // This event loop never calls takeTask()
         return maxPendingTasks == Integer.MAX_VALUE ? PlatformDependent.<Runnable>newMpscQueue()
-                                                    : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
+                : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
     }
 
     /**
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoopGroup.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoopGroup.java
index acb4212fe97..1a383a78c35 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoopGroup.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoopGroup.java
@@ -119,6 +119,13 @@ public EpollEventLoopGroup(int nThreads, Executor executor, EventExecutorChooser
         super(nThreads, executor, chooserFactory, 0, selectStrategyFactory, rejectedExecutionHandler);
     }
 
+    public EpollEventLoopGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory,
+                               SelectStrategyFactory selectStrategyFactory,
+                               RejectedExecutionHandler rejectedExecutionHandler,
+                               EventLoopTaskQueueFactory queueFactory) {
+        super(nThreads, executor, chooserFactory, 0, selectStrategyFactory, rejectedExecutionHandler, queueFactory);
+    }
+
     /**
      * Sets the percentage of the desired amount of time spent for I/O in the child event loops.  The default value is
      * {@code 50}, which means the event loop will try to spend the same amount of time for I/O as for non-I/O tasks.
@@ -131,7 +138,9 @@ public void setIoRatio(int ioRatio) {
 
     @Override
     protected EventLoop newChild(Executor executor, Object... args) throws Exception {
+        EventLoopTaskQueueFactory queueFactory = args.length == 4 ? (EventLoopTaskQueueFactory) args[3] : null;
         return new EpollEventLoop(this, executor, (Integer) args[0],
-                ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
+                ((SelectStrategyFactory) args[1]).newSelectStrategy(),
+                (RejectedExecutionHandler) args[2], queueFactory);
     }
 }
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 a968118a671..567dd67c315 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
@@ -17,6 +17,7 @@
 
 import io.netty.channel.EventLoop;
 import io.netty.channel.EventLoopGroup;
+import io.netty.channel.MultithreadEventLoopGroup;
 import io.netty.channel.SelectStrategy;
 import io.netty.channel.SingleThreadEventLoop;
 import io.netty.channel.kqueue.AbstractKQueueChannel.AbstractKQueueUnsafe;
@@ -71,8 +72,10 @@ public int get() throws Exception {
     private volatile int ioRatio = 50;
 
     KQueueEventLoop(EventLoopGroup parent, Executor executor, int maxEvents,
-                    SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {
-        super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);
+                    SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,
+                    MultithreadEventLoopGroup.EventLoopTaskQueueFactory queueFactory) {
+        super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory),
+                rejectedExecutionHandler);
         selectStrategy = ObjectUtil.checkNotNull(strategy, "strategy");
         this.kqueueFd = Native.newKQueue();
         if (maxEvents == 0) {
@@ -90,6 +93,14 @@ public int get() throws Exception {
         }
     }
 
+    private static Queue<Runnable> newTaskQueue(
+            MultithreadEventLoopGroup.EventLoopTaskQueueFactory queueFactory) {
+        if (queueFactory == null) {
+            return newTaskQueue0(DEFAULT_MAX_PENDING_TASKS);
+        }
+        return queueFactory.newTaskQueue(DEFAULT_MAX_PENDING_TASKS);
+    }
+
     void add(AbstractKQueueChannel ch) {
         assert inEventLoop();
         AbstractKQueueChannel old = channels.put(ch.fd().intValue(), ch);
@@ -305,9 +316,13 @@ protected void run() {
 
     @Override
     protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
+        return newTaskQueue0(maxPendingTasks);
+    }
+
+    private static Queue<Runnable> newTaskQueue0(int maxPendingTasks) {
         // This event loop never calls takeTask()
         return maxPendingTasks == Integer.MAX_VALUE ? PlatformDependent.<Runnable>newMpscQueue()
-                                                    : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
+                : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
     }
 
     /**
diff --git a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueEventLoopGroup.java b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueEventLoopGroup.java
index fe32aa5a7b9..77325c4c235 100644
--- a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueEventLoopGroup.java
+++ b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueEventLoopGroup.java
@@ -116,6 +116,14 @@ public KQueueEventLoopGroup(int nThreads, Executor executor, EventExecutorChoose
         super(nThreads, executor, chooserFactory, 0, selectStrategyFactory, rejectedExecutionHandler);
     }
 
+    public KQueueEventLoopGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory,
+                                SelectStrategyFactory selectStrategyFactory,
+                                RejectedExecutionHandler rejectedExecutionHandler,
+                                EventLoopTaskQueueFactory queueFactory) {
+        super(nThreads, executor, chooserFactory, 0, selectStrategyFactory,
+                rejectedExecutionHandler, queueFactory);
+    }
+
     /**
      * Sets the percentage of the desired amount of time spent for I/O in the child event loops.  The default value is
      * {@code 50}, which means the event loop will try to spend the same amount of time for I/O as for non-I/O tasks.
@@ -128,7 +136,10 @@ public void setIoRatio(int ioRatio) {
 
     @Override
     protected EventLoop newChild(Executor executor, Object... args) throws Exception {
+        EventLoopTaskQueueFactory queueFactory = args.length == 4 ? (EventLoopTaskQueueFactory) args[3] : null;
+
         return new KQueueEventLoop(this, executor, (Integer) args[0],
-                ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
+                ((SelectStrategyFactory) args[1]).newSelectStrategy(),
+                (RejectedExecutionHandler) args[2], queueFactory);
     }
 }
diff --git a/transport/src/main/java/io/netty/channel/MultithreadEventLoopGroup.java b/transport/src/main/java/io/netty/channel/MultithreadEventLoopGroup.java
index a9bc23dfe96..ee322184719 100644
--- a/transport/src/main/java/io/netty/channel/MultithreadEventLoopGroup.java
+++ b/transport/src/main/java/io/netty/channel/MultithreadEventLoopGroup.java
@@ -23,6 +23,7 @@
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
+import java.util.Queue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ThreadFactory;
 
@@ -96,4 +97,21 @@ public ChannelFuture register(ChannelPromise promise) {
     public ChannelFuture register(Channel channel, ChannelPromise promise) {
         return next().register(channel, promise);
     }
+
+    /**
+     * Factory used to create {@link Queue} instances that will be used to store tasks for an {@link EventLoop}.
+     *
+     * Generally speaking the returned {@link Queue} MUST be thread-safe and depending on the {@link EventLoop}
+     * implementation must be of type {@link java.util.concurrent.BlockingQueue}.
+     */
+    public interface EventLoopTaskQueueFactory {
+
+        /**
+         * Returns a new {@link Queue} to use.
+         * @param maxCapacity the maximum amount of elements that can be stored in the {@link Queue} at a given point
+         *                    in time.
+         * @return the new queue.
+         */
+        Queue<Runnable> newTaskQueue(int maxCapacity);
+    }
 }
diff --git a/transport/src/main/java/io/netty/channel/SingleThreadEventLoop.java b/transport/src/main/java/io/netty/channel/SingleThreadEventLoop.java
index 1fe2d3fe2b5..9abb39fc389 100644
--- a/transport/src/main/java/io/netty/channel/SingleThreadEventLoop.java
+++ b/transport/src/main/java/io/netty/channel/SingleThreadEventLoop.java
@@ -59,6 +59,13 @@ protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,
         tailTasks = newTaskQueue(maxPendingTasks);
     }
 
+    protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,
+                                    boolean addTaskWakesUp, Queue<Runnable> taskQueue, Queue<Runnable> tailTaskQueue,
+                                    RejectedExecutionHandler rejectedExecutionHandler) {
+        super(parent, executor, addTaskWakesUp, taskQueue, rejectedExecutionHandler);
+        tailTasks = ObjectUtil.checkNotNull(tailTaskQueue, "tailTaskQueue");
+    }
+
     @Override
     public EventLoopGroup parent() {
         return (EventLoopGroup) super.parent();
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 bb9a1e25cf6..edd41a57905 100644
--- a/transport/src/main/java/io/netty/channel/nio/NioEventLoop.java
+++ b/transport/src/main/java/io/netty/channel/nio/NioEventLoop.java
@@ -19,6 +19,7 @@
 import io.netty.channel.ChannelException;
 import io.netty.channel.EventLoop;
 import io.netty.channel.EventLoopException;
+import io.netty.channel.MultithreadEventLoopGroup;
 import io.netty.channel.SelectStrategy;
 import io.netty.channel.SingleThreadEventLoop;
 import io.netty.util.IntSupplier;
@@ -131,8 +132,10 @@ public Void run() {
     private boolean needsToSelectAgain;
 
     NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
-                 SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {
-        super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);
+                 SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,
+                 MultithreadEventLoopGroup.EventLoopTaskQueueFactory queueFactory) {
+        super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory),
+                rejectedExecutionHandler);
         if (selectorProvider == null) {
             throw new NullPointerException("selectorProvider");
         }
@@ -146,6 +149,14 @@ public Void run() {
         selectStrategy = strategy;
     }
 
+    private static Queue<Runnable> newTaskQueue(
+            MultithreadEventLoopGroup.EventLoopTaskQueueFactory queueFactory) {
+        if (queueFactory == null) {
+            return newTaskQueue0(DEFAULT_MAX_PENDING_TASKS);
+        }
+        return queueFactory.newTaskQueue(DEFAULT_MAX_PENDING_TASKS);
+    }
+
     private static final class SelectorTuple {
         final Selector unwrappedSelector;
         final Selector selector;
@@ -265,9 +276,13 @@ public SelectorProvider selectorProvider() {
 
     @Override
     protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
+        return newTaskQueue0(maxPendingTasks);
+    }
+
+    private static Queue<Runnable> newTaskQueue0(int maxPendingTasks) {
         // This event loop never calls takeTask()
         return maxPendingTasks == Integer.MAX_VALUE ? PlatformDependent.<Runnable>newMpscQueue()
-                                                    : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
+                : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
     }
 
     /**
diff --git a/transport/src/main/java/io/netty/channel/nio/NioEventLoopGroup.java b/transport/src/main/java/io/netty/channel/nio/NioEventLoopGroup.java
index 833b7549785..8598e010439 100644
--- a/transport/src/main/java/io/netty/channel/nio/NioEventLoopGroup.java
+++ b/transport/src/main/java/io/netty/channel/nio/NioEventLoopGroup.java
@@ -101,6 +101,15 @@ public NioEventLoopGroup(int nThreads, Executor executor, EventExecutorChooserFa
         super(nThreads, executor, chooserFactory, selectorProvider, selectStrategyFactory, rejectedExecutionHandler);
     }
 
+    public NioEventLoopGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory,
+                             final SelectorProvider selectorProvider,
+                             final SelectStrategyFactory selectStrategyFactory,
+                             final RejectedExecutionHandler rejectedExecutionHandler,
+                             final EventLoopTaskQueueFactory taskQueueFactory) {
+        super(nThreads, executor, chooserFactory, selectorProvider, selectStrategyFactory,
+                rejectedExecutionHandler, taskQueueFactory);
+    }
+
     /**
      * Sets the percentage of the desired amount of time spent for I/O in the child event loops.  The default value is
      * {@code 50}, which means the event loop will try to spend the same amount of time for I/O as for non-I/O tasks.
@@ -123,7 +132,8 @@ public void rebuildSelectors() {
 
     @Override
     protected EventLoop newChild(Executor executor, Object... args) throws Exception {
+        EventLoopTaskQueueFactory queueFactory = args.length == 4 ? (EventLoopTaskQueueFactory) args[3] : null;
         return new NioEventLoop(this, executor, (SelectorProvider) args[0],
-            ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
+            ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2], queueFactory);
     }
 }


Test patch:
diff --git a/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollEventLoopTest.java b/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollEventLoopTest.java
index f250c272e64..71c56b6d863 100644
--- a/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollEventLoopTest.java
+++ b/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollEventLoopTest.java
@@ -51,7 +51,7 @@ public void testScheduleBigDelayNotOverflow() {
 
         final EventLoopGroup group = new EpollEventLoop(null,
                 new ThreadPerTaskExecutor(new DefaultThreadFactory(getClass())), 0,
-                DefaultSelectStrategyFactory.INSTANCE.newSelectStrategy(), RejectedExecutionHandlers.reject()) {
+                DefaultSelectStrategyFactory.INSTANCE.newSelectStrategy(), RejectedExecutionHandlers.reject(), null) {
             @Override
             void handleLoopException(Throwable t) {
                 capture.set(t);
diff --git a/transport/src/test/java/io/netty/channel/nio/NioEventLoopTest.java b/transport/src/test/java/io/netty/channel/nio/NioEventLoopTest.java
index 7db891afc62..99c949b0600 100644
--- a/transport/src/test/java/io/netty/channel/nio/NioEventLoopTest.java
+++ b/transport/src/test/java/io/netty/channel/nio/NioEventLoopTest.java
@@ -17,15 +17,20 @@
 
 import io.netty.channel.AbstractEventLoopTest;
 import io.netty.channel.Channel;
+import io.netty.channel.DefaultSelectStrategyFactory;
 import io.netty.channel.EventLoop;
 import io.netty.channel.EventLoopGroup;
+import io.netty.channel.MultithreadEventLoopGroup;
 import io.netty.channel.SelectStrategy;
 import io.netty.channel.SelectStrategyFactory;
 import io.netty.channel.socket.ServerSocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.util.IntSupplier;
+import io.netty.util.concurrent.DefaultEventExecutorChooserFactory;
 import io.netty.util.concurrent.DefaultThreadFactory;
 import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.RejectedExecutionHandlers;
+import io.netty.util.concurrent.ThreadPerTaskExecutor;
 import org.hamcrest.core.IsInstanceOf;
 import org.junit.Ignore;
 import org.junit.Test;
@@ -36,9 +41,12 @@
 import java.nio.channels.Selector;
 import java.nio.channels.SocketChannel;
 import java.nio.channels.spi.SelectorProvider;
+import java.util.Queue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.junit.Assert.*;
@@ -281,4 +289,35 @@ public void testChannelsRegistered()  {
             group.shutdownGracefully();
         }
     }
+
+    @Test
+    public void testCustomQueue()  {
+        final AtomicBoolean called = new AtomicBoolean();
+        NioEventLoopGroup group = new NioEventLoopGroup(1,
+                new ThreadPerTaskExecutor(new DefaultThreadFactory(NioEventLoopGroup.class)),
+                DefaultEventExecutorChooserFactory.INSTANCE, SelectorProvider.provider(),
+                DefaultSelectStrategyFactory.INSTANCE, RejectedExecutionHandlers.reject(),
+                new MultithreadEventLoopGroup.EventLoopTaskQueueFactory() {
+                    @Override
+                    public Queue<Runnable> newTaskQueue(int maxCapacity) {
+                        called.set(true);
+                        return new LinkedBlockingQueue<Runnable>(maxCapacity);
+                    }
+        });
+
+        final NioEventLoop loop = (NioEventLoop) group.next();
+
+        try {
+            loop.submit(new Runnable() {
+                @Override
+                public void run() {
+                    // NOOP.
+                }
+            }).syncUninterruptibly();
+            assertTrue(called.get());
+        } finally {
+            group.shutdownGracefully();
+        }
+    }
+
 }