Recover from Selector IOException (#8569)

Motivation:

When the Selector throws an IOException during our EventLoop processing we should rebuild it and transfer the registered Channels. At the moment we will continue trying to use it which will never work.

Modifications:

- Rebuild Selector when an IOException is thrown during any select*(...) methods.
- Add unit test.

Result:

Fixes https://github.com/netty/netty/issues/8566.

transport/src/main/java/io/netty/channel/nio/NioEventLoop.java

@@ -401,7 +401,8 @@ public final class NioEventLoop extends SingleThreadEventLoop {
     protected void run() {
         for (;;) {
             try {
-                switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
+                try {
+                    switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                     case SelectStrategy.CONTINUE:
                         continue;
 
@@ -444,6 +445,13 @@ public final class NioEventLoop extends SingleThreadEventLoop {
                         }
                         // fall through
                     default:
+                    }
+                } catch (IOException e) {
+                    // If we receive an IOException here its because the Selector is messed up. Let's rebuild
+                    // the selector and retry. https://github.com/netty/netty/issues/8566
+                    rebuildSelector0();
+                    handleLoopException(e);
+                    continue;
                 }
 
                 cancelledKeys = 0;

transport/src/test/java/io/netty/channel/nio/NioEventLoopTest.java

@@ -19,16 +19,22 @@ import io.netty.channel.AbstractEventLoopTest;
 import io.netty.channel.Channel;
 import io.netty.channel.EventLoop;
 import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SelectStrategy;
+import io.netty.channel.SelectStrategyFactory;
 import io.netty.channel.socket.ServerSocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.util.IntSupplier;
+import io.netty.util.concurrent.DefaultThreadFactory;
 import io.netty.util.concurrent.Future;
 import org.hamcrest.core.IsInstanceOf;
 import org.junit.Test;
 
+import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.SocketChannel;
+import java.nio.channels.spi.SelectorProvider;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -211,4 +217,45 @@ public class NioEventLoopTest extends AbstractEventLoopTest {
         }
     }
 
+    @Test
+    public void testRebuildSelectorOnIOException() {
+        SelectStrategyFactory selectStrategyFactory = new SelectStrategyFactory() {
+            @Override
+            public SelectStrategy newSelectStrategy() {
+                return new SelectStrategy() {
+
+                    private boolean thrown;
+
+                    @Override
+                    public int calculateStrategy(IntSupplier selectSupplier, boolean hasTasks) throws Exception {
+                        if (!thrown) {
+                            thrown = true;
+                            throw new IOException();
+                        }
+                        return -1;
+                    }
+                };
+            }
+        };
+
+        EventLoopGroup group = new NioEventLoopGroup(1, new DefaultThreadFactory("ioPool"),
+                                                     SelectorProvider.provider(), selectStrategyFactory);
+        final NioEventLoop loop = (NioEventLoop) group.next();
+        try {
+            Channel channel = new NioServerSocketChannel();
+            Selector selector = loop.unwrappedSelector();
+
+            loop.register(channel).syncUninterruptibly();
+
+            Selector newSelector = ((NioEventLoop) channel.eventLoop()).unwrappedSelector();
+            assertTrue(newSelector.isOpen());
+            assertNotSame(selector, newSelector);
+            assertFalse(selector.isOpen());
+
+            channel.close().syncUninterruptibly();
+        } finally {
+            group.shutdownGracefully();
+        }
+    }
+
 }