Improved DefaultChannelPipeline#destroy() to avoid spinning continuously in case of custom executors.

Motivation:

The previous DefaultChannelPipeline#destroy() implementation, introduced in netty#3156, is suboptimal as it can cause the for loop to continuously spin if the executor used by a given handler is unable to "recognize" the event loop.
It could be objected that it's the custom executor responsibility to properly implement the inEventLoop() method, but some implementetaions might not be able to do that for performance reasons, and even so, it's always better to be safe against API misuse, in particular when it is not possible to fail fast and the alternative is rather some sutle behaviour.

Modifications:

The patch simply avoids the recursive spin by explicitly passing the "in event loop" condition as a boolean parameter, preserving the same guarantees offered by netty#3156. A unit test has also been added.

Result:

All channel events are correctly called and no high CPU usage is seen anymore.

Author: sbtourist

transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java

@@ -867,35 +867,36 @@ public ChannelPipeline fireChannelUnregistered() {
      * See: https://github.com/netty/netty/issues/3156
      */
     private void destroy() {
-        destroyUp(head.next);
+        destroyUp(head.next, false);
     }
 
-    private void destroyUp(AbstractChannelHandlerContext ctx) {
+    private void destroyUp(AbstractChannelHandlerContext ctx, boolean inEventLoop) {
         final Thread currentThread = Thread.currentThread();
         final AbstractChannelHandlerContext tail = this.tail;
         for (;;) {
             if (ctx == tail) {
-                destroyDown(currentThread, tail.prev);
+                destroyDown(currentThread, tail.prev, inEventLoop);
                 break;
             }
 
             final EventExecutor executor = ctx.executor();
-            if (!executor.inEventLoop(currentThread)) {
+            if (!inEventLoop && !executor.inEventLoop(currentThread)) {
                 final AbstractChannelHandlerContext finalCtx = ctx;
                 executor.unwrap().execute(new OneTimeTask() {
                     @Override
                     public void run() {
-                        destroyUp(finalCtx);
+                        destroyUp(finalCtx, true);
                     }
                 });
                 break;
             }
 
             ctx = ctx.next;
+            inEventLoop = false;
         }
     }
 
-    private void destroyDown(Thread currentThread, AbstractChannelHandlerContext ctx) {
+    private void destroyDown(Thread currentThread, AbstractChannelHandlerContext ctx, boolean inEventLoop) {
         // We have reached at tail; now traverse backwards.
         final AbstractChannelHandlerContext head = this.head;
         for (;;) {
@@ -904,7 +905,7 @@ private void destroyDown(Thread currentThread, AbstractChannelHandlerContext ctx
             }
 
             final EventExecutor executor = ctx.executor();
-            if (executor.inEventLoop(currentThread)) {
+            if (inEventLoop || executor.inEventLoop(currentThread)) {
                 synchronized (this) {
                     remove0(ctx);
                 }
@@ -913,13 +914,14 @@ private void destroyDown(Thread currentThread, AbstractChannelHandlerContext ctx
                 executor.unwrap().execute(new OneTimeTask() {
                     @Override
                     public void run() {
-                        destroyDown(Thread.currentThread(), finalCtx);
+                        destroyDown(Thread.currentThread(), finalCtx, true);
                     }
                 });
                 break;
             }
 
             ctx = ctx.prev;
+            inEventLoop = false;
         }
     }
 

transport/src/test/java/io/netty/channel/DefaultChannelPipelineTest.java

@@ -28,6 +28,9 @@
 import io.netty.util.AbstractReferenceCounted;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.ReferenceCounted;
+import io.netty.util.concurrent.AbstractEventExecutor;
+import io.netty.util.concurrent.EventExecutorGroup;
+import io.netty.util.concurrent.Future;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Test;
@@ -38,6 +41,8 @@
 import java.util.List;
 import java.util.Queue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -572,6 +577,37 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
         assertSame(exception, error.get());
     }
 
+    @Test
+    public void testChannelUnregistrationWithCustomExecutor() throws Exception {
+        final CountDownLatch channelLatch = new CountDownLatch(1);
+        final CountDownLatch handlerLatch = new CountDownLatch(1);
+        ChannelPipeline pipeline = new LocalChannel().pipeline();
+        pipeline.addLast(new ChannelInitializer<Channel>() {
+            @Override
+            protected void initChannel(Channel ch) throws Exception {
+                ch.pipeline().addLast(new WrapperExecutor(),
+                        new ChannelInboundHandlerAdapter() {
+
+                            @Override
+                            public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
+                                channelLatch.countDown();
+                            }
+
+                            @Override
+                            public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+                                handlerLatch.countDown();
+                            }
+                        });
+            }
+        });
+        Channel channel = pipeline.channel();
+        group.register(channel);
+        channel.close();
+        channel.deregister();
+        assertTrue(channelLatch.await(2, TimeUnit.SECONDS));
+        assertTrue(handlerLatch.await(2, TimeUnit.SECONDS));
+    }
+
     private static int next(AbstractChannelHandlerContext ctx) {
         AbstractChannelHandlerContext next = ctx.next;
         if (next == null) {
@@ -683,4 +719,64 @@ public void handlerRemoved(ChannelHandlerContext ctx) {
             afterRemove = true;
         }
     }
+
+    private static final class WrapperExecutor extends AbstractEventExecutor {
+
+        private final ExecutorService wrapped = Executors.newSingleThreadExecutor();
+
+        @Override
+        public boolean isShuttingDown() {
+            return wrapped.isShutdown();
+        }
+
+        @Override
+        public Future<?> shutdownGracefully(long l, long l2, TimeUnit timeUnit) {
+            throw new IllegalStateException();
+        }
+
+        @Override
+        public Future<?> terminationFuture() {
+            throw new IllegalStateException();
+        }
+
+        @Override
+        public void shutdown() {
+            wrapped.shutdown();
+        }
+
+        @Override
+        public List<Runnable> shutdownNow() {
+            return wrapped.shutdownNow();
+        }
+
+        @Override
+        public boolean isShutdown() {
+            return wrapped.isShutdown();
+        }
+
+        @Override
+        public boolean isTerminated() {
+            return wrapped.isTerminated();
+        }
+
+        @Override
+        public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+            return wrapped.awaitTermination(timeout, unit);
+        }
+
+        @Override
+        public EventExecutorGroup parent() {
+            return null;
+        }
+
+        @Override
+        public boolean inEventLoop(Thread thread) {
+            return false;
+        }
+
+        @Override
+        public void execute(Runnable command) {
+            wrapped.execute(command);
+        }
+    }
 }