Repository: netty/netty
Issue: ChannelInitializer.initChannel() executed more than once when used with context executor
Labels: defect

### Expected behavior

Expected ChannelInitializer.initChannel() to be executed one. 

### Actual behavior

A subsequent execution causes exception to be thrown because it can't find itself or remove itself again.

```
2018-12-01 01:22:02,527 [WARN] [pool-2-thread-1] ChannelInitializer ? Failed to initialize a channel. Closing: [id: 0xce54469f, L:local:test - R:local:E:d49bc512]
java.lang.AssertionError: expected object to not be null
	at org.testng.Assert.fail(Assert.java:93) ~[testng-6.11.jar:?]
	at org.testng.Assert.assertNotNull(Assert.java:422) ~[testng-6.11.jar:?]
	at org.testng.Assert.assertNotNull(Assert.java:407) ~[testng-6.11.jar:?]
	at test.TestChannelInitializer$2.initChannel(TestChannelInitializer.java:93) ~[espresso-router-impl/:?]
	at io.netty.channel.ChannelInitializer.initChannel(ChannelInitializer.java:113) [netty-all-4.1.22.Final.jar:4.1.22.Final]
	at io.netty.channel.ChannelInitializer.channelRegistered(ChannelInitializer.java:76) [netty-all-4.1.22.Final.jar:4.1.22.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRegistered(AbstractChannelHandlerContext.java:149) [netty-all-4.1.22.Final.jar:4.1.22.Final]
	at io.netty.channel.AbstractChannelHandlerContext.access$000(AbstractChannelHandlerContext.java:38) [netty-all-4.1.22.Final.jar:4.1.22.Final]
	at io.netty.channel.AbstractChannelHandlerContext$1.run(AbstractChannelHandlerContext.java:140) [netty-all-4.1.22.Final.jar:4.1.22.Final]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_121]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_121]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [?:1.8.0_121]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [?:1.8.0_121]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_121]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_121]
	at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]
2018-12-01 01:22:02,533 [WARN] [pool-2-thread-1] LoggingHandler - [id: 0xce54469f, L:local:test ! R:local:E:d49bc512] READ: Hello World
2018-12-01 01:22:02,533 [WARN] [pool-2-thread-1] LoggingHandler - [id: 0xce54469f, L:local:test ! R:local:E:d49bc512] READ COMPLETE
2018-12-01 01:22:02,533 [WARN] [pool-2-thread-1] LoggingHandler - [id: 0xce54469f, L:local:test ! R:local:E:d49bc512] ACTIVE
2018-12-01 01:22:02,533 [WARN] [pool-2-thread-1] LoggingHandler - [id: 0xce54469f, L:local:test ! R:local:E:d49bc512] REGISTERED
2018-12-01 01:22:02,534 [WARN] [pool-2-thread-1] LoggingHandler - [id: 0xce54469f, L:local:test ! R:local:E:d49bc512] INACTIVE
2018-12-01 01:22:02,534 [WARN] [pool-2-thread-1] LoggingHandler - [id: 0xce54469f, L:local:test ! R:local:E:d49bc512] UNREGISTERED

java.lang.AssertionError: expected [1] but found [2]
Expected :1
Actual   :2


	at org.testng.Assert.fail(Assert.java:93)
	at org.testng.Assert.failNotEquals(Assert.java:512)
	at org.testng.Assert.assertEqualsImpl(Assert.java:134)
	at org.testng.Assert.assertEquals(Assert.java:115)
	at org.testng.Assert.assertEquals(Assert.java:388)
	at org.testng.Assert.assertEquals(Assert.java:398)
	at test.TestChannelInitializer.test(TestChannelInitializer.java:137)

```

### Steps to reproduce

See repro code below.

### Minimal yet complete reproducer code (or URL to code)

```java
package test;

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.DefaultEventLoopGroup;
import io.netty.channel.local.LocalAddress;
import io.netty.channel.local.LocalChannel;
import io.netty.channel.local.LocalServerChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.AbstractEventExecutor;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Future;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.testng.Assert;
import org.testng.annotations.Test;


public class TestChannelInitializer {

  @Test
  public void test() throws InterruptedException {

    AtomicInteger invokeCount = new AtomicInteger();
    AtomicInteger completeCount = new AtomicInteger();

    LocalAddress addr = new LocalAddress("test");

    ChannelHandler logger = new LoggingHandler(LogLevel.WARN);

    ScheduledExecutorService execService = Executors.newSingleThreadScheduledExecutor();

    EventExecutor exec = new AbstractEventExecutor() {
      @Override
      public void shutdown() {
        throw new IllegalStateException();
      }

      @Override
      public boolean inEventLoop(Thread thread) {
        return false;
      }

      @Override
      public boolean isShuttingDown() {
        return false;
      }

      @Override
      public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
        throw new IllegalStateException();
      }

      @Override
      public Future<?> terminationFuture() {
        throw new IllegalStateException();
      }

      @Override
      public boolean isShutdown() {
        return execService.isShutdown();
      }

      @Override
      public boolean isTerminated() {
        return execService.isTerminated();
      }

      @Override
      public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return execService.awaitTermination(timeout, unit);
      }

      @Override
      public void execute(Runnable command) {
        execService.schedule(command, 1, TimeUnit.NANOSECONDS);
      }
    };

    ChannelInitializer<Channel> otherInitializer = new ChannelInitializer<Channel>() {
      @Override
      protected void initChannel(Channel ch) throws Exception {
        invokeCount.incrementAndGet();
        ChannelHandlerContext ctx = ch.pipeline().context(this);

        Assert.assertNotNull(ctx); // FAILS HERE

        ch.pipeline().addAfter(ctx.executor(), ctx.name(), null, logger);
        completeCount.decrementAndGet();
      }
    };

    DefaultEventLoopGroup group = new DefaultEventLoopGroup(1);

    ServerBootstrap serverBootstrap = new ServerBootstrap()
        .channel(LocalServerChannel.class)
        .group(group)
        .localAddress(addr)
        .childHandler(new ChannelInitializer<LocalChannel>() {
          @Override
          protected void initChannel(LocalChannel ch) throws Exception {
            ch.pipeline().addLast(exec, otherInitializer);
          }
        });

    Channel server = serverBootstrap.bind().sync().channel();

    Bootstrap clientBootstrap = new Bootstrap()
        .channel(LocalChannel.class)
        .group(group)
        .remoteAddress(addr)
        .handler(new ChannelInitializer<LocalChannel>() {
          @Override
          protected void initChannel(LocalChannel ch) throws Exception {
          }
        });

    Channel client = clientBootstrap.connect().sync().channel();

    client.writeAndFlush("Hello World").sync();

    Thread.sleep(1000);

    client.close().sync();
    server.close().sync();

    execService.shutdown();
    execService.awaitTermination(10, TimeUnit.SECONDS);

    Assert.assertEquals(invokeCount.get(), 1); // This seems to be 2 instead of 1
    Assert.assertEquals(invokeCount.get(), completeCount.get());
  }
}
```

### Netty version

4.1.22.Final

### JVM version (e.g. `java -version`)

```
java version "1.8.0_121"
Java(TM) SE Runtime Environment (build 1.8.0_121-b13)
Java HotSpot(TM) 64-Bit Server VM (build 25.121-b13, mixed mode)
```

### OS version (e.g. `uname -a`)

RHEL7
3.10.0-514.55.4.el7.x86_64 #1 SMP Fri Aug 10 17:03:02 UTC 2018 x86_64 x86_64 x86_64 GNU/Linux

Issue comments:
@atcurtis can you please try netty 4.1.32.Final and let me know if the same problem happens there as well? I think I remember that we fixed something related to this at some point. 

Please let me know if after the upgrade the problem still persists.
I have tried with 4.1.32.Final and it fails exactly the same except with slightly different line numbers:
```
	at io.netty.channel.ChannelInitializer.initChannel(ChannelInitializer.java:115)
	at io.netty.channel.ChannelInitializer.channelRegistered(ChannelInitializer.java:76)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRegistered(AbstractChannelHandlerContext.java:149)
	at io.netty.channel.AbstractChannelHandlerContext.access$000(AbstractChannelHandlerContext.java:38)
	at io.netty.channel.AbstractChannelHandlerContext$1.run(AbstractChannelHandlerContext.java:140)
```
I will have a look

> Am 03.12.2018 um 21:08 schrieb Antony T Curtis <notifications@github.com>:
> 
> I have tried with 4.1.32.Final and it fails exactly the same except with slightly different line numbers:
> 
> 	at io.netty.channel.ChannelInitializer.initChannel(ChannelInitializer.java:115)
> 	at io.netty.channel.ChannelInitializer.channelRegistered(ChannelInitializer.java:76)
> 	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRegistered(AbstractChannelHandlerContext.java:149)
> 	at io.netty.channel.AbstractChannelHandlerContext.access$000(AbstractChannelHandlerContext.java:38)
> 	at io.netty.channel.AbstractChannelHandlerContext$1.run(AbstractChannelHandlerContext.java:140)
> —
> You are receiving this because you commented.
> Reply to this email directly, view it on GitHub, or mute the thread.

@atcurtis check https://github.com/netty/netty/pull/8620

Pull request review comments:
What's the point of this line? Wouldn't we only get here if `handlerRemoved()` was called? Is the concern that someone could override `handlerRemoved()`?
This will retain `ctx` until the channel is closed, even if `handlerRemoved()` is called. If this is a one-off initializer (not reused) then that could greatly increase the lifetime of some objects used during handshakes and the like.

Is this seriously the only way we can clean up properly? Is the problem that the methods can be overridden? If so, then maybe we should fix that in Netty 5? Or maybe we could avoid the check-for-double-register problem by having an alternative approach to 26aa34853a8974d212e12b98e708790606bea5fa in Netty 5?

It is possible to cancel the future to clean up, but the amount of effort necessary to do something simple makes me question if this is more of systemic problem that should be addressed in Netty 5.
yes :/
@ejona86 yeah I think its the best we can do for netty 4 to ensure we not leak if the user overrides `handlerRemoved(...)`.Thats also why I check `isRemoved()` first. 
Perhaps we could check handlerRemoved() was overridden at instantiation time using reflection and take the fast path?
@trustin we could ... that said I think it does not really worth it as `ctx.isRemoved()` should be true if the `EventExecutor` is not some funky custom implemention that is is used with `ChannelInitializer`. So I would prefer to keep things simple for now.
Agreed.

Patch that resolved the issue:
diff --git a/transport/src/main/java/io/netty/channel/ChannelInitializer.java b/transport/src/main/java/io/netty/channel/ChannelInitializer.java
index 9ea1b182219..18344d200fa 100644
--- a/transport/src/main/java/io/netty/channel/ChannelInitializer.java
+++ b/transport/src/main/java/io/netty/channel/ChannelInitializer.java
@@ -18,11 +18,12 @@
 import io.netty.bootstrap.Bootstrap;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelHandler.Sharable;
-import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
-import java.util.concurrent.ConcurrentMap;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * A special {@link ChannelInboundHandler} which offers an easy way to initialize a {@link Channel} once it was
@@ -53,9 +54,10 @@
 public abstract class ChannelInitializer<C extends Channel> extends ChannelInboundHandlerAdapter {
 
     private static final InternalLogger logger = InternalLoggerFactory.getInstance(ChannelInitializer.class);
-    // We use a ConcurrentMap as a ChannelInitializer is usually shared between all Channels in a Bootstrap /
+    // We use a Set as a ChannelInitializer is usually shared between all Channels in a Bootstrap /
     // ServerBootstrap. This way we can reduce the memory usage compared to use Attributes.
-    private final ConcurrentMap<ChannelHandlerContext, Boolean> initMap = PlatformDependent.newConcurrentHashMap();
+    private final Set<ChannelHandlerContext> initMap = Collections.newSetFromMap(
+            new ConcurrentHashMap<ChannelHandlerContext, Boolean>());
 
     /**
      * This method will be called once the {@link Channel} was registered. After the method returns this instance
@@ -108,9 +110,14 @@ public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
         }
     }
 
+    @Override
+    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+        initMap.remove(ctx);
+    }
+
     @SuppressWarnings("unchecked")
     private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
-        if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) { // Guard against re-entrance.
+        if (initMap.add(ctx)) { // Guard against re-entrance.
             try {
                 initChannel((C) ctx.channel());
             } catch (Throwable cause) {
@@ -125,14 +132,25 @@ private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
         return false;
     }
 
-    private void remove(ChannelHandlerContext ctx) {
+    private void remove(final ChannelHandlerContext ctx) {
         try {
             ChannelPipeline pipeline = ctx.pipeline();
             if (pipeline.context(this) != null) {
                 pipeline.remove(this);
             }
         } finally {
-            initMap.remove(ctx);
+            // The removal may happen in an async fashion if the EventExecutor we use does something funky.
+            if (ctx.isRemoved()) {
+                initMap.remove(ctx);
+            } else {
+                // Ensure we always remove from the Map in all cases to not produce a memory leak.
+                ctx.channel().closeFuture().addListener(new ChannelFutureListener() {
+                    @Override
+                    public void operationComplete(ChannelFuture future) {
+                        initMap.remove(ctx);
+                    }
+                });
+            }
         }
     }
 }


Test patch:
diff --git a/transport/src/test/java/io/netty/channel/ChannelInitializerTest.java b/transport/src/test/java/io/netty/channel/ChannelInitializerTest.java
index 26b5e4e9fcf..2ac1bcdefa9 100644
--- a/transport/src/test/java/io/netty/channel/ChannelInitializerTest.java
+++ b/transport/src/test/java/io/netty/channel/ChannelInitializerTest.java
@@ -21,12 +21,16 @@
 import io.netty.channel.local.LocalAddress;
 import io.netty.channel.local.LocalChannel;
 import io.netty.channel.local.LocalServerChannel;
+import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.concurrent.Future;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
 import java.util.Iterator;
 import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -35,6 +39,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertSame;
 
 public class ChannelInitializerTest {
@@ -249,6 +254,127 @@ private void testChannelRegisteredEventPropagation(ChannelInitializer<LocalChann
         }
     }
 
+    @SuppressWarnings("deprecation")
+    @Test(timeout = 10000)
+    public void testChannelInitializerEventExecutor() throws Throwable {
+        final AtomicInteger invokeCount = new AtomicInteger();
+        final AtomicInteger completeCount = new AtomicInteger();
+        final AtomicReference<Throwable> errorRef = new AtomicReference<Throwable>();
+        LocalAddress addr = new LocalAddress("test");
+
+        final EventExecutor executor = new DefaultEventLoop() {
+            private final ScheduledExecutorService execService = Executors.newSingleThreadScheduledExecutor();
+
+            @Override
+            public void shutdown() {
+                execService.shutdown();
+            }
+
+            @Override
+            public boolean inEventLoop(Thread thread) {
+                // Always return false which will ensure we always call execute(...)
+                return false;
+            }
+
+            @Override
+            public boolean isShuttingDown() {
+                return false;
+            }
+
+            @Override
+            public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
+                throw new IllegalStateException();
+            }
+
+            @Override
+            public Future<?> terminationFuture() {
+                throw new IllegalStateException();
+            }
+
+            @Override
+            public boolean isShutdown() {
+                return execService.isShutdown();
+            }
+
+            @Override
+            public boolean isTerminated() {
+                return execService.isTerminated();
+            }
+
+            @Override
+            public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+                return execService.awaitTermination(timeout, unit);
+            }
+
+            @Override
+            public void execute(Runnable command) {
+                execService.execute(command);
+            }
+        };
+
+        ServerBootstrap serverBootstrap = new ServerBootstrap()
+                .channel(LocalServerChannel.class)
+                .group(group)
+                .localAddress(addr)
+                .childHandler(new ChannelInitializer<LocalChannel>() {
+                    @Override
+                    protected void initChannel(LocalChannel ch) {
+                        ch.pipeline().addLast(executor, new ChannelInitializer<Channel>() {
+                            @Override
+                            protected void initChannel(Channel ch) {
+                                invokeCount.incrementAndGet();
+                                ChannelHandlerContext ctx = ch.pipeline().context(this);
+                                assertNotNull(ctx);
+                                ch.pipeline().addAfter(ctx.executor(),
+                                        ctx.name(), null, new ChannelInboundHandlerAdapter() {
+                                            @Override
+                                            public void channelRead(ChannelHandlerContext ctx, Object msg)  {
+                                                // just drop on the floor.
+                                            }
+                                        });
+                                completeCount.incrementAndGet();
+                            }
+
+                            @Override
+                            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+                                errorRef.set(cause);
+                            }
+                        });
+                    }
+                });
+
+        Channel server = serverBootstrap.bind().sync().channel();
+
+        Bootstrap clientBootstrap = new Bootstrap()
+                .channel(LocalChannel.class)
+                .group(group)
+                .remoteAddress(addr)
+                .handler(new ChannelInboundHandlerAdapter());
+
+        Channel client = clientBootstrap.connect().sync().channel();
+        client.writeAndFlush("Hello World").sync();
+
+        client.close().sync();
+        server.close().sync();
+
+        client.closeFuture().sync();
+        server.closeFuture().sync();
+
+        // Give some time to execute everything that was submitted before.
+        Thread.sleep(1000);
+
+        executor.shutdown();
+        assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+
+        assertEquals(invokeCount.get(), 1);
+        assertEquals(invokeCount.get(), completeCount.get());
+
+        Throwable cause = errorRef.get();
+        if (cause != null) {
+            throw cause;
+        }
+    }
+
     private static void closeChannel(Channel c) {
         if (c != null) {
             c.close().syncUninterruptibly();