Repository: netty/netty
Issue: NioEventLoop task execution might throw UnsupportedOperationException on shutdown

### Expected behavior
Execution instead throws `RejectedExecutionException`.

### Actual behavior
There is a racy `UnsupportedOperationException` instead because the task removal is delegated to `MpscChunkedArrayQueue` that does not support removal. This happens with `SingleThreadEventExecutor` that overrides the `newTaskQueue` to return an MPSC queue instead of the `LinkedBlockingQueue` returned by the base class such as `NioEventLoop`, `EpollEventLoop` and `KQueueEventLoop`.

### Steps to reproduce
Schedule task to an event loop concurrently to an event loop shutdown.

### Minimal yet complete reproducer code (or URL to code)
```
    @Test
    public void testTaskRemovalOnShutdownThrowsUnsupportedOperationException() throws Exception {
        NioEventLoopGroup group = new NioEventLoopGroup(1);
        final NioEventLoop loop = (NioEventLoop) group.next();
        final Runnable task = new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        new Thread() {
            @Override
            public void run() {
                while (true) {
                    loop.execute(task);
                }
            }
        }.start();
        Thread.sleep(10);
        loop.shutdownNow();
        Thread.sleep(100000);
    }
```

this is a racy test, the event loop shutdown race against the the task producing thread, if you run it enough it will fail:

```
Exception in thread "Thread-0" java.lang.UnsupportedOperationException
	at org.jctools.queues.BaseMpscLinkedArrayQueue.iterator(BaseMpscLinkedArrayQueue.java:201)
	at java.util.AbstractCollection.remove(AbstractCollection.java:282)
	at io.netty.util.concurrent.SingleThreadEventExecutor.removeTask(SingleThreadEventExecutor.java:340)
	at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:781)
	at io.netty.channel.nio.NioEventLoopTest$8.run(NioEventLoopTest.java:195)
```

### Netty version
At least 4.1.x

### JVM version (e.g. `java -version`)
N/A

### OS version (e.g. `uname -a`)
N/A

Issue comments:
@vietj thanks! Will provide a fix soon. 

Pull request review comments:
Add a bit of javadoc about  the meaning of the returned boolean? (especially as we no longer delegate a call to the queue which could have effectively provided that information via its javadoc). Ditto the other `removeTask` methods.
Yes this makes things less racy. Why not use a CAS loop here to make even less racy (especially as we have a  volatile field updater available)? Might be overkill I grant.
I see the `shutdown` method is `@Deprecated`. Is that to remove it from the public API but it will still be called by something? A bit of javadoc about the deprecation would be great.
Just noticed that `addTask` already has that `if (isShutdown())  reject()` logic in it!
@davidmoten sure.. 
oh duh! you are right... 
Will? :laughing: 
ah I meant what does true mean and what does false mean. Does true mean it was found on the queue and removed? Does false mean it is either not supported or not found on the queue?
Another thing we could do is just catch `UnsupportedOperationException`. WDYT ?
Why is this code even caring about `isShutdown()`? It looks like this block should just be deleted. `isShutdown()` is checked by `addTask()`/`offerTask()`. If we added the entry to the queue, then it seems we should commit ourselves to it; attempting to remove it (which is not guaranteed to be successful, since the task may have already run) seems weird.

If this `isShutdown()` check was atomic with `startThread()`, I could sort of understand the removal since we know we didn't start the thread, but as it is it just seems like useless effort.

Patch that resolved the issue:
diff --git a/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java b/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java
index 67959016152..dab7e9500d0 100644
--- a/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java
+++ b/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java
@@ -778,8 +778,20 @@ public void execute(Runnable task) {
         addTask(task);
         if (!inEventLoop) {
             startThread();
-            if (isShutdown() && removeTask(task)) {
-                reject();
+            if (isShutdown()) {
+                boolean reject = false;
+                try {
+                    if (removeTask(task)) {
+                        reject = true;
+                    }
+                } catch (UnsupportedOperationException e) {
+                    // The task queue does not support removal so the best thing we can do is to just move on and
+                    // hope we will be able to pick-up the task before its completely terminated.
+                    // In worst case we will log on termination.
+                }
+                if (reject) {
+                    reject();
+                }
             }
         }
 


Test patch:
diff --git a/transport/src/test/java/io/netty/channel/nio/NioEventLoopTest.java b/transport/src/test/java/io/netty/channel/nio/NioEventLoopTest.java
index d3412c2d996..15fcb212442 100644
--- a/transport/src/test/java/io/netty/channel/nio/NioEventLoopTest.java
+++ b/transport/src/test/java/io/netty/channel/nio/NioEventLoopTest.java
@@ -22,6 +22,7 @@
 import io.netty.channel.socket.ServerSocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.util.concurrent.Future;
+import org.hamcrest.core.IsInstanceOf;
 import org.junit.Test;
 
 import java.net.InetSocketAddress;
@@ -29,7 +30,9 @@
 import java.nio.channels.Selector;
 import java.nio.channels.SocketChannel;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.junit.Assert.*;
 
@@ -171,4 +174,41 @@ public void channelUnregistered(SocketChannel ch, Throwable cause) {
             group.shutdownGracefully();
         }
     }
+
+    @SuppressWarnings("deprecation")
+    @Test
+    public void testTaskRemovalOnShutdownThrowsNoUnsupportedOperationException() throws Exception {
+        final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
+        final Runnable task = new Runnable() {
+            @Override
+            public void run() {
+                // NOOP
+            }
+        };
+        // Just run often enough to trigger it normally.
+        for (int i = 0; i < 1000; i++) {
+            NioEventLoopGroup group = new NioEventLoopGroup(1);
+            final NioEventLoop loop = (NioEventLoop) group.next();
+
+            Thread t = new Thread(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        for (;;) {
+                            loop.execute(task);
+                        }
+                    } catch (Throwable cause) {
+                        error.set(cause);
+                    }
+                }
+            });
+            t.start();
+            group.shutdownNow();
+            t.join();
+            group.terminationFuture().syncUninterruptibly();
+            assertThat(error.get(), IsInstanceOf.instanceOf(RejectedExecutionException.class));
+            error.set(null);
+        }
+    }
+
 }