Fix possible IllegalStateException caused by closeNotifyTimeout when using SslHandler

Motivation:

In the SslHandler we schedule a timeout at which we close the Channel if a timeout was detected during close_notify. Because this can race with notify the flushFuture we can see an IllegalStateException when the Channel is closed.

Modifications:

- Use a trySuccess() and tryFailure(...) to guard against race.

Result:

No more race.

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

@@ -15,15 +15,21 @@
  */
 package io.netty.util.concurrent;
 
+import io.netty.util.internal.logging.InternalLogger;
+import io.netty.util.internal.logging.InternalLoggerFactory;
+
+import static io.netty.util.internal.ObjectUtil.checkNotNull;
+
 /**
  * {@link GenericFutureListener} implementation which takes other {@link Future}s
  * and notifies them on completion.
  *
- * @param V the type of value returned by the future
+ * @param <V> the type of value returned by the future
- * @param F the type of future
+ * @param <F> the type of future
  */
 public class PromiseNotifier<V, F extends Future<V>> implements GenericFutureListener<F> {
 
+    private static final InternalLogger logger = InternalLoggerFactory.getInstance(PromiseNotifier.class);
     private final Promise<? super V>[] promises;
 
     /**
@@ -33,9 +39,7 @@ public class PromiseNotifier<V, F extends Future<V>> implements GenericFutureLis
      */
     @SafeVarargs
     public PromiseNotifier(Promise<? super V>... promises) {
-        if (promises == null) {
+        checkNotNull(promises, "promises");
-            throw new NullPointerException("promises");
-        }
         for (Promise<? super V> promise: promises) {
             if (promise == null) {
                 throw new IllegalArgumentException("promises contains null Promise");
@@ -49,15 +53,18 @@ public class PromiseNotifier<V, F extends Future<V>> implements GenericFutureLis
         if (future.isSuccess()) {
             V result = future.get();
             for (Promise<? super V> p: promises) {
-                p.setSuccess(result);
+                if (!p.trySuccess(result)) {
+                    logger.warn("Failed to mark a promise as success because it is done already: {}", p);
+                }
             }
             return;
         }
 
         Throwable cause = future.cause();
         for (Promise<? super V> p: promises) {
-            p.setFailure(cause);
+            if (!p.tryFailure(cause)) {
+                logger.warn("Failed to mark a promise as failure because it's done already: {}", p, cause);
+            }
         }
     }
-
 }

common/src/test/java/io/netty/util/concurrent/PromiseNotifierTest.java

@@ -55,8 +55,8 @@ public class PromiseNotifierTest {
         Future<Void> future = createStrictMock(Future.class);
         expect(future.isSuccess()).andReturn(true);
         expect(future.get()).andReturn(null);
-        expect(p1.setSuccess(null)).andReturn(p1);
+        expect(p1.trySuccess(null)).andReturn(true);
-        expect(p2.setSuccess(null)).andReturn(p2);
+        expect(p2.trySuccess(null)).andReturn(true);
         replay(p1, p2, future);
 
         notifier.operationComplete(future);
@@ -79,8 +79,8 @@ public class PromiseNotifierTest {
         Throwable t = createStrictMock(Throwable.class);
         expect(future.isSuccess()).andReturn(false);
         expect(future.cause()).andReturn(t);
-        expect(p1.setFailure(t)).andReturn(p1);
+        expect(p1.tryFailure(t)).andReturn(true);
-        expect(p2.setFailure(t)).andReturn(p2);
+        expect(p2.tryFailure(t)).andReturn(true);
         replay(p1, p2, future);
 
         notifier.operationComplete(future);

handler/src/main/java/io/netty/handler/ssl/SslHandler.java

@@ -30,6 +30,7 @@ import io.netty.channel.ChannelInboundHandler;
 import io.netty.channel.ChannelOutboundHandler;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.ChannelPromise;
+import io.netty.channel.ChannelPromiseNotifier;
 import io.netty.channel.PendingWriteQueue;
 import io.netty.handler.codec.ByteToMessageDecoder;
 import io.netty.util.concurrent.DefaultPromise;
@@ -1459,7 +1460,12 @@ public class SslHandler extends ByteToMessageDecoder implements ChannelOutboundH
                 @Override
                 public void run() {
                     logger.warn("{} Last write attempt timed out; force-closing the connection.", ctx.channel());
-                    ctx.close(promise);
+
+                    // We notify the promise in the TryNotifyListener as there is a "race" where the close(...) call
+                    // by the timeoutFuture and the close call in the flushFuture listener will be called. Because of
+                    // this we need to use trySuccess() and tryFailure(...) as otherwise we can cause an
+                    // IllegalStateException.
+                    ctx.close(ctx.newPromise()).addListener(new ChannelPromiseNotifier(promise));
                 }
             }, closeNotifyTimeoutMillis, TimeUnit.MILLISECONDS);
         } else {
@@ -1476,7 +1482,12 @@ public class SslHandler extends ByteToMessageDecoder implements ChannelOutboundH
                 }
                 // Trigger the close in all cases to make sure the promise is notified
                 // See https://github.com/netty/netty/issues/2358
-                ctx.close(promise);
+                //
+                // We notify the promise in the ChannelPromiseNotifier as there is a "race" where the close(...) call
+                // by the timeoutFuture and the close call in the flushFuture listener will be called. Because of
+                // this we need to use trySuccess() and tryFailure(...) as otherwise we can cause an
+                // IllegalStateException.
+                ctx.close(ctx.newPromise()).addListener(new ChannelPromiseNotifier(promise));
             }
         });
     }

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

@@ -32,5 +32,4 @@ public final class ChannelPromiseNotifier
     public ChannelPromiseNotifier(ChannelPromise... promises) {
         super(promises);
     }
-
 }