EPOLLRDHUP infinite loop

Motivation:
If  is enabled and a channel is half closed it is possible for the EPOLL event loop to get into an infinite loop by continuously being woken up on the EPOLLRDHUP event.

Modifications:
- Ensure that the EPOLLRDHUP event is unregistered for to prevent infinite loop.

Result:
1 less infinite loop.

transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollChannel.java

@@ -22,7 +22,9 @@ import io.netty.buffer.Unpooled;
 import io.netty.channel.AbstractChannel;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelMetadata;
+import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoop;
+import io.netty.channel.socket.ChannelInputShutdownEvent;
 import io.netty.channel.unix.FileDescriptor;
 import io.netty.channel.unix.UnixChannel;
 import io.netty.util.ReferenceCountUtil;
@@ -40,6 +42,7 @@ abstract class AbstractEpollChannel extends AbstractChannel implements UnixChann
     protected int flags = Native.EPOLLET;
 
     protected volatile boolean active;
+    private volatile boolean inputShutdown;
 
     AbstractEpollChannel(int fd, int flag) {
         this(null, fd, flag, false);
@@ -175,6 +178,10 @@ abstract class AbstractEpollChannel extends AbstractChannel implements UnixChann
         loop.add(this);
     }
 
+    protected final boolean isInputShutdown0() {
+        return inputShutdown;
+    }
+
     @Override
     protected abstract AbstractEpollUnsafe newUnsafe();
 
@@ -304,8 +311,47 @@ abstract class AbstractEpollChannel extends AbstractChannel implements UnixChann
         /**
          * Called once EPOLLRDHUP event is ready to be processed
          */
-        void epollRdHupReady() {
-            // NOOP
+        final void epollRdHupReady() {
+            if (isActive()) {
+                // If it is still active, we need to call epollInReady as otherwise we may miss to
+                // read pending data from the underlying file descriptor.
+                // See https://github.com/netty/netty/issues/3709
+                epollInReady();
+
+                // Clear the EPOLLRDHUP flag to prevent continuously getting woken up on this event.
+                clearEpollRdHup();
+            }
+            // epollInReady may call this, but we should ensure that it gets called.
+            shutdownInput();
+        }
+
+        /**
+         * Clear the {@link Native#EPOLLRDHUP} flag from EPOLL, and close on failure.
+         */
+        private void clearEpollRdHup() {
+            try {
+                clearFlag(Native.EPOLLRDHUP);
+            } catch (IOException e) {
+                pipeline().fireExceptionCaught(e);
+                close(voidPromise());
+            }
+        }
+
+        /**
+         * Shutdown the input side of the channel.
+         */
+        void shutdownInput() {
+            if (!inputShutdown) { // Best effort check on volatile variable to prevent multiple shutdowns
+                inputShutdown = true;
+                if (isOpen()) {
+                    if (Boolean.TRUE.equals(config().getOption(ChannelOption.ALLOW_HALF_CLOSURE))) {
+                        clearEpollIn0();
+                        pipeline().fireUserEventTriggered(ChannelInputShutdownEvent.INSTANCE);
+                    } else {
+                        close(voidPromise());
+                    }
+                }
+            }
         }
 
         @Override

transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java

@@ -22,7 +22,6 @@ import io.netty.channel.Channel;
 import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
-import io.netty.channel.ChannelOption;
 import io.netty.channel.ChannelOutboundBuffer;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.ChannelPromise;
@@ -30,7 +29,6 @@ import io.netty.channel.ConnectTimeoutException;
 import io.netty.channel.DefaultFileRegion;
 import io.netty.channel.EventLoop;
 import io.netty.channel.RecvByteBufAllocator;
-import io.netty.channel.socket.ChannelInputShutdownEvent;
 import io.netty.channel.unix.FileDescriptor;
 import io.netty.util.internal.EmptyArrays;
 import io.netty.util.internal.MpscLinkedQueueNode;
@@ -71,7 +69,6 @@ public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel {
     private SocketAddress requestedRemoteAddress;
     private final Queue<SpliceInTask> spliceQueue = PlatformDependent.newMpscQueue();
 
-    private volatile boolean inputShutdown;
     private volatile boolean outputShutdown;
 
     // Lazy init these if we need to splice(...)
@@ -507,10 +504,6 @@ public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel {
                 "unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
     }
 
-    protected boolean isInputShutdown0() {
-        return inputShutdown;
-    }
-
     protected boolean isOutputShutdown0() {
         return outputShutdown || !isActive();
     }
@@ -598,18 +591,6 @@ public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel {
 
         private RecvByteBufAllocator.Handle allocHandle;
 
-        private void closeOnRead(ChannelPipeline pipeline) {
-            inputShutdown = true;
-            if (isOpen()) {
-                if (Boolean.TRUE.equals(config().getOption(ChannelOption.ALLOW_HALF_CLOSURE))) {
-                    clearEpollIn0();
-                    pipeline.fireUserEventTriggered(ChannelInputShutdownEvent.INSTANCE);
-                } else {
-                    close(voidPromise());
-                }
-            }
-        }
-
         private boolean handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Throwable cause, boolean close) {
             if (byteBuf != null) {
                 if (byteBuf.isReadable()) {
@@ -622,7 +603,7 @@ public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel {
             pipeline.fireChannelReadComplete();
             pipeline.fireExceptionCaught(cause);
             if (close || cause instanceof IOException) {
-                closeOnRead(pipeline);
+                shutdownInput();
                 return true;
             }
             return false;
@@ -766,18 +747,6 @@ public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel {
             }
         }
 
-        @Override
-        void epollRdHupReady() {
-            if (isActive()) {
-                // If it is still active, we need to call epollInReady as otherwise we may miss to
-                // read pending data from the underyling file descriptor.
-                // See https://github.com/netty/netty/issues/3709
-                epollInReady();
-            } else {
-                closeOnRead(pipeline());
-            }
-        }
-
         @Override
         void epollInReady() {
             final ChannelConfig config = config();
@@ -860,7 +829,7 @@ public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel {
                 allocHandle.record(totalReadAmount);
 
                 if (close) {
-                    closeOnRead(pipeline);
+                    shutdownInput();
                     close = false;
                 }
             } catch (Throwable t) {