Ensure the best effort is made even if future listeners could not be notified / Handle registration failure in a robust manner

- Related: #1187

common/src/main/java/io/netty/util/concurrent/DefaultPromise.java

@@ -481,16 +481,20 @@ public class DefaultPromise<V> extends AbstractFuture<V> implements Promise<V> {
         } else {
             final Object listeners = this.listeners;
             this.listeners = null;
-            executor.execute(new Runnable() {
+            try {
-                @Override
+                executor.execute(new Runnable() {
-                public void run() {
+                    @Override
-                    if (listeners instanceof DefaultPromiseListeners) {
+                    public void run() {
-                        notifyListeners0(DefaultPromise.this, (DefaultPromiseListeners) listeners);
+                        if (listeners instanceof DefaultPromiseListeners) {
-                    } else {
+                            notifyListeners0(DefaultPromise.this, (DefaultPromiseListeners) listeners);
-                        notifyListener0(DefaultPromise.this, (GenericFutureListener<? extends Future<V>>) listeners);
+                        } else {
+                            notifyListener0(DefaultPromise.this, (GenericFutureListener<? extends Future<V>>) listeners);
+                        }
                     }
-                }
+                });
-            });
+            } catch (Throwable t) {
+                logger.error("Failed to notify listener(s). Event loop terminated?", t);
+            }
         }
     }
 
@@ -505,8 +509,10 @@ public class DefaultPromise<V> extends AbstractFuture<V> implements Promise<V> {
     }
 
     @SuppressWarnings("unchecked")
-    public static void notifyListener(final EventExecutor eventExecutor, final Future<?> future,
+    protected static void notifyListener(
-                               final GenericFutureListener<? extends Future<?>> l) {
+            final EventExecutor eventExecutor, final Future<?> future,
+            final GenericFutureListener<? extends Future<?>> l) {
+
         if (eventExecutor.inEventLoop()) {
             final Integer stackDepth = LISTENER_STACK_DEPTH.get();
             if (stackDepth < MAX_LISTENER_STACK_DEPTH) {
@@ -520,12 +526,16 @@ public class DefaultPromise<V> extends AbstractFuture<V> implements Promise<V> {
             }
         }
 
-        eventExecutor.execute(new Runnable() {
+        try {
+            eventExecutor.execute(new Runnable() {
             @Override
             public void run() {
                 notifyListener(eventExecutor, future, l);
             }
         });
+        } catch (Throwable t) {
+            logger.error("Failed to notify a listener. Event loop terminated?", t);
+        }
     }
 
     @SuppressWarnings({ "unchecked", "rawtypes" })

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

@@ -564,30 +564,32 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
                 throw new NullPointerException("eventLoop");
             }
             if (isRegistered()) {
-                throw new IllegalStateException("registered to an event loop already");
+                promise.setFailure(new IllegalStateException("registered to an event loop already"));
+                return;
             }
             if (!isCompatible(eventLoop)) {
-                throw new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName());
+                promise.setFailure(new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
             }
 
             AbstractChannel.this.eventLoop = eventLoop;
 
-            assert eventLoop().inEventLoop();
-
-            // check if the eventLoop which was given is currently in the eventloop.
-            // if that is the case we are safe to call register, if not we need to
-            // schedule the execution as otherwise we may say some race-conditions.
-            //
-            // See https://github.com/netty/netty/issues/654
             if (eventLoop.inEventLoop()) {
                 register0(promise);
             } else {
-                eventLoop.execute(new Runnable() {
+                try {
-                    @Override
+                    eventLoop.execute(new Runnable() {
-                    public void run() {
+                        @Override
-                        register0(promise);
+                        public void run() {
-                    }
+                            register0(promise);
-                });
+                        }
+                    });
+                } catch (Throwable t) {
+                    logger.warn(
+                            "Force-closing a channel whose registration task was unaccepted by an event loop: {}",
+                            AbstractChannel.this, t);
+                    closeForcibly();
+                    promise.setFailure(t);
+                }
             }
         }
 

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

@@ -16,8 +16,8 @@
 package io.netty.channel;
 
 import io.netty.util.concurrent.EventExecutorGroup;
-import io.netty.util.concurrent.TaskScheduler;
 import io.netty.util.concurrent.SingleThreadEventExecutor;
+import io.netty.util.concurrent.TaskScheduler;
 
 import java.util.concurrent.ThreadFactory;
 
@@ -48,31 +48,19 @@ public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor im
 
     @Override
     public ChannelFuture register(Channel channel) {
-        if (channel == null) {
-            throw new NullPointerException("channel");
-        }
         return register(channel, channel.newPromise());
     }
 
     @Override
     public ChannelFuture register(final Channel channel, final ChannelPromise promise) {
-        if (isShutdown()) {
+        if (channel == null) {
-            channel.unsafe().closeForcibly();
+            throw new NullPointerException("channel");
-            promise.setFailure(new EventLoopException("cannot register a channel to a shut down loop"));
+        }
-            return promise;
+        if (promise == null) {
-        }
+            throw new NullPointerException("promise");
-
-        if (inEventLoop()) {
-            channel.unsafe().register(this, promise);
-        } else {
-            execute(new Runnable() {
-                @Override
-                public void run() {
-                    channel.unsafe().register(SingleThreadEventLoop.this, promise);
-                }
-            });
         }
 
+        channel.unsafe().register(this, promise);
         return promise;
     }
 }

transport/src/main/java/io/netty/channel/local/LocalChannel.java

@@ -194,9 +194,10 @@ public class LocalChannel extends AbstractChannel {
 
     @Override
     protected void doClose() throws Exception {
-        if (peer.isActive()) {
+        LocalChannel peer = this.peer;
+        if (peer != null && peer.isActive()) {
             peer.unsafe().close(peer.unsafe().voidFuture());
-            peer = null;
+            this.peer = null;
         }
     }
 

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

@@ -15,6 +15,7 @@
  */
 package io.netty.channel;
 
+import io.netty.channel.local.LocalChannel;
 import io.netty.util.concurrent.TaskScheduler;
 import org.junit.After;
 import org.junit.Before;
@@ -24,11 +25,13 @@ import java.util.Queue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
+import static org.hamcrest.CoreMatchers.*;
 import static org.junit.Assert.*;
 
 public class SingleThreadEventLoopTest {
@@ -244,6 +247,45 @@ public class SingleThreadEventLoopTest {
         assertEquals(NUM_TASKS, ranTasks.get());
     }
 
+    @Test(timeout = 10000)
+    public void testRegistrationAfterTermination() throws Exception {
+        loop.shutdown();
+        while (!loop.isTerminated()) {
+            loop.awaitTermination(1, TimeUnit.DAYS);
+        }
+
+        ChannelFuture f = loop.register(new LocalChannel());
+        f.awaitUninterruptibly();
+        assertFalse(f.isSuccess());
+        assertThat(f.cause(), is(instanceOf(RejectedExecutionException.class)));
+    }
+
+    @Test(timeout = 10000)
+    public void testRegistrationAfterTermination2() throws Exception {
+        loop.shutdown();
+        while (!loop.isTerminated()) {
+            loop.awaitTermination(1, TimeUnit.DAYS);
+        }
+
+        final CountDownLatch latch = new CountDownLatch(1);
+        Channel ch = new LocalChannel();
+        ChannelPromise promise = ch.newPromise();
+        promise.addListener(new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) throws Exception {
+                latch.countDown();
+            }
+        });
+
+        ChannelFuture f = loop.register(ch, promise);
+        f.awaitUninterruptibly();
+        assertFalse(f.isSuccess());
+        assertThat(f.cause(), is(instanceOf(RejectedExecutionException.class)));
+
+        // Ensure the listener was notified.
+        assertFalse(latch.await(1, TimeUnit.SECONDS));
+    }
+
     private static class SingleThreadEventLoopImpl extends SingleThreadEventLoop {
 
         final AtomicInteger cleanedUp = new AtomicInteger();
@@ -274,11 +316,5 @@ public class SingleThreadEventLoopTest {
         protected void cleanup() {
             cleanedUp.incrementAndGet();
         }
-
-        @Override
-        public ChannelFuture register(Channel channel, ChannelPromise future) {
-            // Untested
-            return future;
-        }
     }
 }