Repository: netty/netty
Issue: ChannelGroup has race-condition
Labels: defect

Hi,

the ChannelGroup doc says:

``` java
 * If both {@link ServerChannel}s and non-{@link ServerChannel}s exist in the
 * same {@link ChannelGroup}, any requested I/O operations on the group are
 * performed for the {@link ServerChannel}s first and then for the others.
 * <p>
 * This rule is very useful when you shut down a server in one shot:
 *
 * <pre>
 * <strong>{@link ChannelGroup} allChannels =
 *         new {@link DefaultChannelGroup}({@link GlobalEventExecutor}.INSTANCE);</strong>
 *
 * public static void main(String[] args) throws Exception {
 *     {@link ServerBootstrap} b = new {@link ServerBootstrap}(..);
 *     ...
 *     b.childHandler(new MyHandler());
 *
 *     // Start the server
 *     b.getPipeline().addLast("handler", new MyHandler());
 *     {@link Channel} serverChannel = b.bind(..).sync();
 *     <strong>allChannels.add(serverChannel);</strong>
 *
 *     ... Wait until the shutdown signal reception ...
 *
 *     // Close the serverChannel and then all accepted connections.
 *     <strong>allChannels.close().awaitUninterruptibly();</strong>
 * }
 *
 * public class MyHandler extends {@link ChannelHandlerAdapter} {
 *     {@code @Override}
 *     public void channelActive({@link ChannelHandlerContext} ctx) {
 *         // closed on shutdown.
 *         <strong>allChannels.add(ctx.channel());</strong>
 *         super.channelActive(ctx);
 *     }
 * }
```

But, this is _not_ thread-safe in my opinion, you could leak client channels:

``` java
public class TestCloseMain {

    static volatile Channel ch;

    public static void main(String[] args) throws IOException {

        EventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);

        ChannelGroup cg = new DefaultChannelGroup(eventLoopGroup.next());

        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(eventLoopGroup)
                       .channel(NioServerSocketChannel.class)
                       .handler(new ChannelHandlerAdapter() {


                           @Override
                           public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                               System.out.println(msg.getClass() + " -> " + msg);


                               cg.close(); // <- If this happen concurrently = BOOM!

                               super.channelRead(ctx, msg);
                           }
                       })
                       .childOption(ChannelOption.TCP_NODELAY, true)
                       .childHandler(new ChannelInitializer<Channel>() {
                           @Override
                           protected void initChannel(Channel ch) throws Exception {
                               ch.pipeline().addLast(new ChannelHandlerAdapter() {
                                   @Override
                                   public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                       cg.add(ctx.channel());
                                       System.out.println("Still active! " + ctx.channel().parent().isOpen());
                                       super.channelActive(ctx);
                                   }
                               });
                           }
                       });

        cg.add(serverBootstrap.bind(1337).syncUninterruptibly().channel());

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                 .channel(NioSocketChannel.class)
                 .option(ChannelOption.TCP_NODELAY, true)
                 .handler(new ChannelInitializer<Channel>() {
                     @Override
                     protected void initChannel(Channel ch) throws Exception {

                     }
                 });

        ch = bootstrap.connect("localhost", 1337).syncUninterruptibly().channel();

        System.in.read();
        ch.close().awaitUninterruptibly();
    }
}
```

I know, its only using 1 thread, but it demonstrates the race condition:

``` java
cg.close(); // <- If this happen concurrently = BOOM!
```

If the channel group is closed, after the client channel has been registered, the client channel is not closed properly.

Do I miss something ? This sounds quite severe for me.


Issue comments:
The only solution to this bug would be to dispose the ChannelGroup, so once a CG is closed, it remains closed and every new add()-invocation results in closing the channel.

The only other solution is to synchronize on the CG which would be bad.

EDIT: Is there nobody who can verify if I'm right or wrong ?

Can really nobody prove or disprove this statement ? 

@chrisprobst I'm not sure I understand the actual problem... Can you show a unit test that demonstrate the issue ?

I'm actually on holiday (small laptop, no ide, no real internet etc.), but I will try to explain the problem.
- Status-quo:
  - Doc suggests closing ServerChannel + ChildChannels by using a ChannelGroup
  - Doc suggests adding ChildChannels to the ChannelGroup in their corresponding channelActive method
  - Docs says, this is enough to close everything
- Problem lies between accepting a channel and invoking ChildChannel.channelActive
  - Time between ServerChannel.close() and ChildChannel.channelActive is _not_ atomic
- Flow to reproduce the problem
  - Let the ServerChannel "read" (accept) the incoming ChildChannel
    - ChildChannel.channelActive is scheduled for later invocation 
  - THEN: Call ChannelGroup.close()
    - ServerChannel gets close()'d first, then all _registered_ ChildChannels
      - BUT: The previously accepted ChildChannel is _not_ yet registered
      - The previously started and scheduled ChildChannel will work just fine and launch, but it shouldn't =(
      - Boom: Race-Condition, which can lead to never closed ChildChannels

I hope, this made it clear. It's actually very easy to reproduce, my initial source example (I know, not a testable unit) shows how.

This hit me quite hard, I simply couldn't find the bug, but it was the evil ChannelGroup xD.
I know, usually this is a very short time frame and unlikely to happen. But it _is_ a race condition, IMHO.

@chrisprobst I agree this is a race condition. I didn't think deep on how we can fix this, but the easiest workaround is to shut down the event loop, which closes every single channel it manages.

Hi Trustin,

maybe we should consider creating a ChannelGroup, which remains closed. So every channel, which adds itself after the ChannelGroup has been closed, will also be closed during adding.

EDIT: And of course, changing the doc =).

@chrisprobst That sounds like a good idea IMHO. Care to sign the [ICLA](http://netty.io/s/icla) and sending us a PR?  What do you think @normanmaurer ? 

I believe, I've already done die signing, but I'm not sure. Can I check this ?

About the PR: I could do the change when I'm back from my holiday ;) But maybe it's too close, after holiday I'm in Russia for a few days. Mid September sounds realistic.

The only problem is, that using this technique, the channel group will not be Lock-Free anymore, because IHMO we would need a Lock, since adding to the group depends on the state of the flag, which seems to be unsolvable with Atomics\* or volatiles. I have to refresh my knowledge about concurrent stuff I think.

Maybe Doug Lea can help :+1: 

Hey @trustin @normanmaurer ,

I've just thought about it a bit deeper. I think the ChannelGroup is not the best spot for solving this race condition. A closed ChannelGroup is messier than I first thought, how should other methods react to a closed state ?

Maybe we should follow the path of less resistance - just change the doc, so nobody will do the same mistake as me.

We could just enhance the doc, which suggests using it as Server/Child-Channel manager.
So instead of saying, that a ChannelGroup on its own is enough, we could suggest using a simple bool flag, which should be very fine and should guard against this nasty race condition.
The following shows by idea:

```
// In the ChildHandler we can do:

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        channelGroup.add(ctx.channel());
        if(volatileClosedFlag == true) { ctx.channel().close(); } // Must be checked AFTER adding!!

        super.channelActive(ctx);
    }

// If you want to close ALL channels:
// [...]
volatileClosedFlag = true; // Set this BEFORE closing
group.close();
// [...]
```

My Java-Memory-Model knowledge is a bit rusty, but I think this should do it.

IMHO we need at this point no locks, as long as we check the flag for being set AFTER adding child channels to it.

So the time-flow is something like: 
- set flag to true -> close channels async -> add child to channel group -> check for flag, close() if true

Concurrency is always a bit of a brain twister, maybe I'm wrong. But I believe this should be safe.

Best Regards,
Chris

Ok, to keep it simple:

```
public final class CloseableChannelGroup extends DefaultChannelGroup {

    private volatile boolean closed = false;

    public CloseableChannelGroup(EventExecutor executor) {
        super(executor);
    }

    public CloseableChannelGroup(String name, EventExecutor executor) {
        super(name, executor);
    }

    @Override
    public boolean add(Channel channel) {
        boolean addResult = super.add(channel);

        // First add channel, than check if closed
        // Seems inefficient at first, but this way a volatile
        // gives us enough synchronization to be thread-safe
        // If true: Close right away...
        // If false: Channel will definitely be closed by the group (closed=true always happens-before close();)
        if (closed) {
            channel.close();
        }
        return addResult;
    }

    @Override
    public ChannelGroupFuture close(ChannelMatcher matcher) {
        // It is important to set the flag to true, before closing channels
        // Our invariants are: closed=true -> closed() -> add() -> closed==true?
        closed = true;
        return super.close(matcher);
    }

    @Override
    public ChannelGroupFuture close() {
        // It is important to set the flag to true, before closing channels
        // Our invariants are: closed=true -> closed() -> add() -> closed==true?
        closed = true;
        return super.close();
    }
}
```

This should be enough. Don't know if you @normanmaurer  @trustin  agree with style and stuff..

Regards,
Chris

@chrisprobst Yeah, you've signed the ICLA on 4/30/2015 0:56:11. Thanks for the heads up.

Re: The proposed change - LGTM. You are gonna incorporate the change into DefaultChannelFutureGroup, right?

@trustin Hi, I'm confused. Why the DefaultChannelFutureGroup? I'm not sure how to do fix the RC there.
IMHO, we have 2 choices. Change the DefaultChannelGroup (bad, because for compatibility), or create a new ChannelGroup based on the DefaultChannelGroup and let it be closable only once.

Re: Yes, I could do the change.

Ugh, I meant incorporating the change into `DefaultChannelGroup`. 

I certainly would, but it would change the semantics of the DefaultChannelGroup. You couldn't reuse a ChannelGroup after it is closed.

_Is this expected ?_

If that's the only way to solve the race condition, I think that's fine. To be honest, I haven't thought people would reuse it.

IMHO its the only option, if you want to be lock-free. Something like a reset button would definitely require locks. And a locked channel group... meh. We could add a specification parameter at construction time like (stay closed) other similar.

I will prepare a pull request with the changes (including doc).


Pull request review comments:

Patch that resolved the issue:
diff --git a/transport/src/main/java/io/netty/channel/group/DefaultChannelGroup.java b/transport/src/main/java/io/netty/channel/group/DefaultChannelGroup.java
index 34029f7a3f9..a16938c7679 100644
--- a/transport/src/main/java/io/netty/channel/group/DefaultChannelGroup.java
+++ b/transport/src/main/java/io/netty/channel/group/DefaultChannelGroup.java
@@ -52,13 +52,15 @@ public void operationComplete(ChannelFuture future) throws Exception {
             remove(future.channel());
         }
     };
+    private final boolean stayClosed;
+    private volatile boolean closed;
 
     /**
      * Creates a new group with a generated name and the provided {@link EventExecutor} to notify the
      * {@link ChannelGroupFuture}s.
      */
     public DefaultChannelGroup(EventExecutor executor) {
-        this("group-0x" + Integer.toHexString(nextId.incrementAndGet()), executor);
+        this(executor, false);
     }
 
     /**
@@ -67,11 +69,33 @@ public DefaultChannelGroup(EventExecutor executor) {
      * duplicate check is done against group names.
      */
     public DefaultChannelGroup(String name, EventExecutor executor) {
+        this(name, executor, false);
+    }
+
+    /**
+     * Creates a new group with a generated name and the provided {@link EventExecutor} to notify the
+     * {@link ChannelGroupFuture}s. {@code stayClosed} defines whether or not, this group can be closed
+     * more than once. Adding channels to a closed group will immediately close them, too. This makes it
+     * easy, to shutdown server and child channels at once.
+     */
+    public DefaultChannelGroup(EventExecutor executor, boolean stayClosed) {
+        this("group-0x" + Integer.toHexString(nextId.incrementAndGet()), executor, stayClosed);
+    }
+
+    /**
+     * Creates a new group with the specified {@code name} and {@link EventExecutor} to notify the
+     * {@link ChannelGroupFuture}s. {@code stayClosed} defines whether or not, this group can be closed
+     * more than once. Adding channels to a closed group will immediately close them, too. This makes it
+     * easy, to shutdown server and child channels at once. Please note that different groups can have
+     * the same name, which means no duplicate check is done against group names.
+     */
+    public DefaultChannelGroup(String name, EventExecutor executor, boolean stayClosed) {
         if (name == null) {
             throw new NullPointerException("name");
         }
         this.name = name;
         this.executor = executor;
+        this.stayClosed = stayClosed;
     }
 
     @Override
@@ -122,6 +146,23 @@ public boolean add(Channel channel) {
         if (added) {
             channel.closeFuture().addListener(remover);
         }
+
+        if (stayClosed && closed) {
+
+            // First add channel, than check if closed.
+            // Seems inefficient at first, but this way a volatile
+            // gives us enough synchronization to be thread-safe.
+            //
+            // If true: Close right away.
+            // (Might be closed a second time by ChannelGroup.close(), but this is ok)
+            //
+            // If false: Channel will definitely be closed by the ChannelGroup.
+            // (Because closed=true always happens-before ChannelGroup.close())
+            //
+            // See https://github.com/netty/netty/issues/4020
+            channel.close();
+        }
+
         return added;
     }
 
@@ -273,6 +314,16 @@ public ChannelGroupFuture close(ChannelMatcher matcher) {
         Map<Channel, ChannelFuture> futures =
                 new LinkedHashMap<Channel, ChannelFuture>(size());
 
+        if (stayClosed) {
+            // It is important to set the closed to true, before closing channels.
+            // Our invariants are:
+            // closed=true happens-before ChannelGroup.close()
+            // ChannelGroup.add() happens-before checking closed==true
+            //
+            // See https://github.com/netty/netty/issues/4020
+            closed = true;
+        }
+
         for (Channel c: serverChannels.values()) {
             if (matcher.matches(c)) {
                 futures.put(c, c.close());


Test patch: