Repository: netty/netty
Issue: A lot of ChannelInputShutdownReadComplete events fired for NIO channel with ALLOW_HALF_CLOSURE
Labels: defect

### Expected behavior
Single `ChannelInputShutdownReadComplete` event will be fired for closed socket.

### Actual behavior
`ChannelInputShutdownReadComplete` event fired on each NIO event loop run.
`NioEventLoop` begin consuming a lot of CPU.

### Steps to reproduce
Create upstream connection channel with enabled `ALLOW_HALF_CLOSURE` and disabled `AUTO_READ`, do a single HTTP keep-alive request and wait until connection being closed by remote server.

### Netty version
4.1.16.Final

### JVM version (e.g. `java -version`)
Java(TM) SE Runtime Environment (build 1.8.0_131-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.131-b11, mixed mode)

### OS version (e.g. `uname -a`)
Darwin 16.7.0 Darwin Kernel Version 16.7.0: Thu Jun 15 17:36:27 PDT 2017; root:xnu-3789.70.16~2/RELEASE_X86_64 x86_64

macOS Sierra 10.12.6

Also confirmed on Linux.

### Suspicious code fragment

from `io.netty.channel.nio.AbstractNioByteChannel::read`

https://github.com/netty/netty/blob/f57d8f00e17a3cbd127aeb2b499c3f2787c002e4/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java#L124-L133

`readPending` not cleared when `allocHandle.lastBytesRead() <= 0`.

Negative read bytes indicates socket closed state, but each consecutive selection of this socket key with `OP_READ` will succeed immediately (may be related to http://bugs.java.com/bugdatabase/view_bug.do?bug_id=4531726).

### Fix proposal

Call `removeReadOp()` on closed sockets.

Issue comments:
Thank you for fast response. It will fix issue for non-autoread channels (original issue), but after looking at affected code I'm afraid that channels with `AUTO_READ` can be hurt with this wakeups.
@khitrin so you are saying it is not fixed with the PR ?
I've built netty with the PR and can confirm that original issue is fixed.

But I think we need further investigation of this case.

1. For 4.1.16.Final (from Maven central) event spam begins immediately when upstream break the connection. Also I checked that setting `readPending` to `false` by debugger helps, so the PR clearly must fix the issue.

2. For 4.1.17.Final-SNAPSHOT (with the PR) single event fired on next `.read()` call.
It is absolutely correct behaviour, but why 4.1.16.Final fires event without `.read()` and why it fixed by this PR?
I will try to rebuild 4.1.16.Final on Mac and test it on Mac and Linux platforms. 

Also similar issue may affect channels with auto read enabled and `readPending` disabling might be not enough here.

also @Scottmitch ....
Finally confirmed on Linux.
@khitrin when you say confirmed... what do you mean ? Confirmed as this pr fix it or what ?
@normanmaurer, I've checked against @khitrin's source with 4.1.16.Final where issue is present on Linux and with 4.1.17.Final-SNAPSHOT from your PR #7259 which behaves same as on MacOSX. So, PR fixes issue same way as on Mac.
I'm still seeing this problem in 4.1.22.Final on Linux (Ubuntu 16.04, OpenJDK 1.8.0_151), which appears to have all the abovementioned fixes in place. I'm using the default NIO channel implementation, and am seeing thousands of these events fired when I shutdown the input.

hmm interesting... @Scottmitch Can you have a look ?
See https://github.com/netty/netty/pull/7801
@khitrin   how do you find the Suspicious code?

Pull request review comments:
Replace `DefaultSocketChannelConfig` with `SocketChannelConfig`. After this its ready to be merged :)

Patch that resolved the issue:
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollChannel.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollChannel.java
index 8d3dc5bbcbd..c1af9975cd2 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollChannel.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollChannel.java
@@ -33,6 +33,7 @@
 import io.netty.channel.RecvByteBufAllocator;
 import io.netty.channel.socket.ChannelInputShutdownEvent;
 import io.netty.channel.socket.ChannelInputShutdownReadComplete;
+import io.netty.channel.socket.SocketChannelConfig;
 import io.netty.channel.unix.FileDescriptor;
 import io.netty.channel.unix.Socket;
 import io.netty.channel.unix.UnixChannel;
@@ -241,8 +242,8 @@ final boolean shouldBreakEpollInReady(ChannelConfig config) {
     }
 
     private static boolean isAllowHalfClosure(ChannelConfig config) {
-        return config instanceof EpollSocketChannelConfig &&
-                ((EpollSocketChannelConfig) config).isAllowHalfClosure();
+        return config instanceof SocketChannelConfig &&
+                ((SocketChannelConfig) config).isAllowHalfClosure();
     }
 
     final void clearEpollIn() {
diff --git a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueChannel.java b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueChannel.java
index f10417ebe17..698e59c7f16 100644
--- a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueChannel.java
+++ b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueChannel.java
@@ -33,6 +33,7 @@
 import io.netty.channel.RecvByteBufAllocator;
 import io.netty.channel.socket.ChannelInputShutdownEvent;
 import io.netty.channel.socket.ChannelInputShutdownReadComplete;
+import io.netty.channel.socket.SocketChannelConfig;
 import io.netty.channel.unix.FileDescriptor;
 import io.netty.channel.unix.UnixChannel;
 import io.netty.util.ReferenceCountUtil;
@@ -322,8 +323,8 @@ final boolean shouldBreakReadReady(ChannelConfig config) {
     }
 
     private static boolean isAllowHalfClosure(ChannelConfig config) {
-        return config instanceof KQueueSocketChannelConfig &&
-                ((KQueueSocketChannelConfig) config).isAllowHalfClosure();
+        return config instanceof SocketChannelConfig &&
+                ((SocketChannelConfig) config).isAllowHalfClosure();
     }
 
     final void clearReadFilter() {
diff --git a/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java b/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
index ffa0d67d479..db87b100b0c 100644
--- a/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
+++ b/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
@@ -21,7 +21,6 @@
 import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelMetadata;
-import io.netty.channel.ChannelOption;
 import io.netty.channel.ChannelOutboundBuffer;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.FileRegion;
@@ -29,6 +28,7 @@
 import io.netty.channel.internal.ChannelUtils;
 import io.netty.channel.socket.ChannelInputShutdownEvent;
 import io.netty.channel.socket.ChannelInputShutdownReadComplete;
+import io.netty.channel.socket.SocketChannelConfig;
 import io.netty.util.internal.StringUtil;
 
 import java.io.IOException;
@@ -54,6 +54,7 @@ public void run() {
             ((AbstractNioUnsafe) unsafe()).flush0();
         }
     };
+    private boolean inputClosedSeenErrorOnRead;
 
     /**
      * Create a new instance
@@ -84,17 +85,27 @@ public ChannelMetadata metadata() {
         return METADATA;
     }
 
+    final boolean shouldBreakReadReady(ChannelConfig config) {
+        return isInputShutdown0() && (inputClosedSeenErrorOnRead || !isAllowHalfClosure(config));
+    }
+
+    private static boolean isAllowHalfClosure(ChannelConfig config) {
+        return config instanceof SocketChannelConfig &&
+                ((SocketChannelConfig) config).isAllowHalfClosure();
+    }
+
     protected class NioByteUnsafe extends AbstractNioUnsafe {
 
         private void closeOnRead(ChannelPipeline pipeline) {
             if (!isInputShutdown0()) {
-                if (Boolean.TRUE.equals(config().getOption(ChannelOption.ALLOW_HALF_CLOSURE))) {
+                if (isAllowHalfClosure(config())) {
                     shutdownInput();
                     pipeline.fireUserEventTriggered(ChannelInputShutdownEvent.INSTANCE);
                 } else {
                     close(voidPromise());
                 }
             } else {
+                inputClosedSeenErrorOnRead = true;
                 pipeline.fireUserEventTriggered(ChannelInputShutdownReadComplete.INSTANCE);
             }
         }
@@ -120,6 +131,10 @@ private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Thro
         @Override
         public final void read() {
             final ChannelConfig config = config();
+            if (shouldBreakReadReady(config)) {
+                clearReadPending();
+                return;
+            }
             final ChannelPipeline pipeline = pipeline();
             final ByteBufAllocator allocator = config.getAllocator();
             final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
diff --git a/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java b/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java
index 477f39ac67b..6eda848f82e 100644
--- a/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java
+++ b/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java
@@ -27,6 +27,7 @@
 import io.netty.channel.FileRegion;
 import io.netty.channel.RecvByteBufAllocator;
 import io.netty.channel.socket.ChannelInputShutdownEvent;
+import io.netty.channel.socket.ChannelInputShutdownReadComplete;
 import io.netty.util.internal.StringUtil;
 
 import java.io.IOException;
@@ -73,6 +74,7 @@ private void closeOnRead(ChannelPipeline pipeline) {
             } else {
                 unsafe().close(unsafe().voidPromise());
             }
+            pipeline.fireUserEventTriggered(ChannelInputShutdownReadComplete.INSTANCE);
         }
     }
 


Test patch:
diff --git a/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketHalfClosedTest.java b/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketHalfClosedTest.java
index b5600d4252c..40b5e86ff73 100644
--- a/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketHalfClosedTest.java
+++ b/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketHalfClosedTest.java
@@ -40,10 +40,82 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 public class SocketHalfClosedTest extends AbstractSocketTest {
+    @Test(timeout = 10000)
+    public void testHalfClosureOnlyOneEventWhenAutoRead() throws Throwable {
+        run();
+    }
+
+    public void testHalfClosureOnlyOneEventWhenAutoRead(ServerBootstrap sb, Bootstrap cb) throws Throwable {
+        Channel serverChannel = null;
+        try {
+            cb.option(ChannelOption.ALLOW_HALF_CLOSURE, true)
+                    .option(ChannelOption.AUTO_READ, true);
+            sb.childHandler(new ChannelInitializer<Channel>() {
+                @Override
+                protected void initChannel(Channel ch) {
+                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
+                        @Override
+                        public void channelActive(ChannelHandlerContext ctx) {
+                            ((DuplexChannel) ctx).shutdownOutput();
+                        }
+
+                        @Override
+                        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+                            ctx.close();
+                        }
+                    });
+                }
+            });
+
+            final AtomicInteger shutdownEventReceivedCounter = new AtomicInteger();
+            final AtomicInteger shutdownReadCompleteEventReceivedCounter = new AtomicInteger();
+
+            cb.handler(new ChannelInitializer<Channel>() {
+                @Override
+                protected void initChannel(Channel ch) {
+                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
+
+                        @Override
+                        public void userEventTriggered(final ChannelHandlerContext ctx, Object evt) {
+                            if (evt == ChannelInputShutdownEvent.INSTANCE) {
+                                shutdownEventReceivedCounter.incrementAndGet();
+                            } else if (evt == ChannelInputShutdownReadComplete.INSTANCE) {
+                                shutdownReadCompleteEventReceivedCounter.incrementAndGet();
+                                ctx.executor().schedule(new Runnable() {
+                                    @Override
+                                    public void run() {
+                                        ctx.close();
+                                    }
+                                }, 100, MILLISECONDS);
+                            }
+                        }
+
+                        @Override
+                        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+                            ctx.close();
+                        }
+                    });
+                }
+            });
+
+            serverChannel = sb.bind().sync().channel();
+            Channel clientChannel = cb.connect(serverChannel.localAddress()).sync().channel();
+            clientChannel.closeFuture().await();
+            assertEquals(1, shutdownEventReceivedCounter.get());
+            assertEquals(1, shutdownReadCompleteEventReceivedCounter.get());
+        } finally {
+            if (serverChannel != null) {
+                serverChannel.close().sync();
+            }
+        }
+    }
+
     @Test
     public void testAllDataReadAfterHalfClosure() throws Throwable {
         run();