Repository: netty/netty (Java)

Issue:
Handling slow clients / detecting idleness
### Expected behavior
When using `IdleStateHandler` with `observeOutput=true` the `IdleStateEvent.WRITER_IDLE` or  `IdleStateEvent.ALL_IDLE` should not trigger because of slow clients. 
This is same issue as in #6150.

### Actual behavior
The `IdleStateHandler` triggers `IdleStateEvent.WRITER_IDLE` or `IdleStateEvent.ALL_IDLE` even though Netty is still writing to the client socket. 

### Steps to reproduce
Create an minimal server, use `IdleStateHandler` to detect idleness. Create an client that slowly downloads the content from the server. Watch the `IdleStateHandler` trigger before the download is complete.

### Minimal yet complete reproducer code (or URL to code)
https://gist.github.com/magnus-gustafsson/c009c04aedf14a8b426dcb48450cc7d4
When this sample is run with the `IdleStateHandler` in the pipeline we get the following output:

```
Mon Mar 04 09:02:28 CET 2019: Server - listening on: http://localhost:8082/
Mon Mar 04 09:02:28 CET 2019: Client - connecting
Mon Mar 04 09:02:28 CET 2019: Server - Received request, writing response with size 67108864 on : [id: 0xb4450a55, L:/127.0.0.1:8082 - R:/127.0.0.1:38182]
Mon Mar 04 09:02:38 CET 2019: Server - Channel is idle, closing it: [id: 0xb4450a55, L:/127.0.0.1:8082 - R:/127.0.0.1:38182], ALL_IDLE
Mon Mar 04 09:02:38 CET 2019: Server - : Write complete. Success was false on [id: 0xb4450a55, L:/127.0.0.1:8082 ! R:/127.0.0.1:38182]
Mon Mar 04 09:02:41 CET 2019: Client - Done : read 20998517
```
There we see that the `ALL_IDLE` event is triggered. This results in that  the client only manage to download 20998517 bytes.

However, if we remove `IdleStateHandler` from the pipeline we get:
```
Mon Mar 04 09:05:05 CET 2019: Server - listening on: http://localhost:8082/
Mon Mar 04 09:05:05 CET 2019: Client - connecting
Mon Mar 04 09:05:05 CET 2019: Server - Received request, writing response with size 67108864 on : [id: 0xf8335944, L:/127.0.0.1:8082 - R:/127.0.0.1:38252]
Mon Mar 04 09:05:44 CET 2019: Server - : Write complete. Success was true on [id: 0xf8335944, L:/127.0.0.1:8082 ! R:/127.0.0.1:38252]
Mon Mar 04 09:05:47 CET 2019: Client - Done : read 67108949
```
The client manage to download everything from the server.


### Netty version
Same behavior in both 4.1.7.Final and 4.1.33.Final

### JVM version (e.g. `java -version`)
java version "1.8.0_161"
Java(TM) SE Runtime Environment (build 1.8.0_161-b12)
Java HotSpot(TM) 64-Bit Server VM (build 25.161-b12, mixed mode)

### OS version (e.g. `uname -a`)
Linux magnusg 4.9.0-040900-generic #201612111631 SMP Sun Dec 11 21:33:00 UTC 2016 x86_64 x86_64 x86_64 GNU/Linux




Hints:
I mean the idleStateHandler triggers when a channel has not performed a read or write action([in the doc](https://netty.io/4.0/api/io/netty/handler/timeout/IdleStateHandler.html))!

You specify that when the channel is not read for 10 seconds then trigger and in the client, you only read all 10 seconds (without subtracting the delay that it took). Of curse it will trigger the IdleStarthandler I mean what did you expect?

For TCP keep alive the server can not really check if the connection is open (the client crashed) without pinging it. This is why [wikipedia](https://en.wikipedia.org/wiki/Keepalive#TCP_keepalive) sais: "TCP keepalive period is required to be configurable and by default is set to no less than 2 hours."
The underlying socket is still being used, the client is still receiving data. Therefor the idelState should not be triggered.
If you check my gist you can verify this your self. You can also use tcpdump to verify this behavior.
I reproduced the issue, I found that if the client's message processing speed is very slow (TCP window size is 0), it will trigger multiple Idle messages.

By looking at the source code, I found that if the client's processing speed was very slow, `IdleStateHandler` would trigger an Idle event to notify the user to avoid an `OutOfMemoryError` error .

But actually in my tests, if I encounter a slow client, Idle events will be triggered many times, which I think is not reasonable.

When using `IdleStateHandler` with `observeOutput=true`,  `IdleStateHandler` detects the output data changes by executing method `hasOutputChanged`

```java
private boolean hasOutputChanged(ChannelHandlerContext ctx, boolean first) {
    if (observeOutput) {
        Channel channel = ctx.channel();
        Unsafe unsafe = channel.unsafe();
        ChannelOutboundBuffer buf = unsafe.outboundBuffer();

        if (buf != null) {
            int messageHashCode = System.identityHashCode(buf.current());
            // pending write bytes will remain unchanged when flushing a large byte buffer entry.
            long pendingWriteBytes = buf.totalPendingWriteBytes();

            if (messageHashCode != lastMessageHashCode || pendingWriteBytes != lastPendingWriteBytes) {
                lastMessageHashCode = messageHashCode;
                lastPendingWriteBytes = pendingWriteBytes;

                if (!first) {
                    return true;
                }
            }
        }
    }
    return false;
}
```
I noticed that `ChannelOutboundBuffer` total pending write bytes size will remain unchanged when flushing a large byte buffer entry. It doesn't change until all buffer data in entry is flushed into channel, so when IdleStateChannel encounters a slow client, hasOutputChanged returns false, triggering the IDLE event. 

I just submitted a related issue #9005


Patch:
diff --git a/handler/src/main/java/io/netty/handler/timeout/IdleStateHandler.java b/handler/src/main/java/io/netty/handler/timeout/IdleStateHandler.java
--- a/handler/src/main/java/io/netty/handler/timeout/IdleStateHandler.java
+++ b/handler/src/main/java/io/netty/handler/timeout/IdleStateHandler.java
@@ -129,6 +129,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
     private long lastChangeCheckTimeStamp;
     private int lastMessageHashCode;
     private long lastPendingWriteBytes;
+    private long lastFlushProgress;
 
     /**
      * Creates a new instance firing {@link IdleStateEvent}s.
@@ -399,6 +400,7 @@ private void initOutputChanged(ChannelHandlerContext ctx) {
             if (buf != null) {
                 lastMessageHashCode = System.identityHashCode(buf.current());
                 lastPendingWriteBytes = buf.totalPendingWriteBytes();
+                lastFlushProgress = buf.currentProgress();
             }
         }
     }
@@ -443,6 +445,15 @@ private boolean hasOutputChanged(ChannelHandlerContext ctx, boolean first) {
                         return true;
                     }
                 }
+
+                long flushProgress = buf.currentProgress();
+                if (flushProgress != lastFlushProgress) {
+                    lastFlushProgress = flushProgress;
+
+                    if (!first) {
+                        return true;
+                    }
+                }
             }
         }
 
diff --git a/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java b/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java
--- a/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java
+++ b/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java
@@ -220,6 +220,18 @@ public Object current() {
         return entry.msg;
     }
 
+    /**
+     * Return the current message flush progress.
+     * @return {@code 0} if nothing was flushed before for the current message or there is no current message
+     */
+    public long currentProgress() {
+        Entry entry = flushedEntry;
+        if (entry == null) {
+            return 0;
+        }
+        return entry.progress;
+    }
+
     /**
      * Notify the {@link ChannelPromise} of the current message about writing progress.
      */
@@ -227,9 +239,9 @@ public void progress(long amount) {
         Entry e = flushedEntry;
         assert e != null;
         ChannelPromise p = e.promise;
+        long progress = e.progress + amount;
+        e.progress = progress;
         if (p instanceof ChannelProgressivePromise) {
-            long progress = e.progress + amount;
-            e.progress = progress;
             ((ChannelProgressivePromise) p).tryProgress(progress, e.total);
         }
     }


Test patch:
diff --git a/handler/src/test/java/io/netty/handler/timeout/IdleStateHandlerTest.java b/handler/src/test/java/io/netty/handler/timeout/IdleStateHandlerTest.java
--- a/handler/src/test/java/io/netty/handler/timeout/IdleStateHandlerTest.java
+++ b/handler/src/test/java/io/netty/handler/timeout/IdleStateHandlerTest.java
@@ -236,6 +236,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
             channel.writeAndFlush(Unpooled.wrappedBuffer(new byte[] { 1 }));
             channel.writeAndFlush(Unpooled.wrappedBuffer(new byte[] { 2 }));
             channel.writeAndFlush(Unpooled.wrappedBuffer(new byte[] { 3 }));
+            channel.writeAndFlush(Unpooled.wrappedBuffer(new byte[5 * 1024]));
 
             // Establish a baseline. We're not consuming anything and let it idle once.
             idleStateHandler.tickRun();
@@ -283,6 +284,30 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
             assertEquals(0, events.size());
             assertEquals(26L, idleStateHandler.tick(TimeUnit.SECONDS)); // 23s + 2s + 1s
 
+            // Consume part of the message every 2 seconds, then be idle for 1 seconds,
+            // then run the task and we should get an IdleStateEvent because the first trigger
+            idleStateHandler.tick(2L, TimeUnit.SECONDS);
+            assertNotNullAndRelease(channel.consumePart(1024));
+            idleStateHandler.tick(2L, TimeUnit.SECONDS);
+            assertNotNullAndRelease(channel.consumePart(1024));
+            idleStateHandler.tickRun(1L, TimeUnit.SECONDS);
+            assertEquals(1, events.size());
+            assertEquals(31L, idleStateHandler.tick(TimeUnit.SECONDS)); // 26s + 2s + 2s + 1s
+            events.clear();
+
+            // Consume part of the message every 2 seconds, then be idle for 1 seconds,
+            // then consume all the rest of the message, then run the task and we shouldn't
+            // get an IdleStateEvent because the data is flowing and we haven't been idle for long enough!
+            idleStateHandler.tick(2L, TimeUnit.SECONDS);
+            assertNotNullAndRelease(channel.consumePart(1024));
+            idleStateHandler.tick(2L, TimeUnit.SECONDS);
+            assertNotNullAndRelease(channel.consumePart(1024));
+            idleStateHandler.tickRun(1L, TimeUnit.SECONDS);
+            assertEquals(0, events.size());
+            assertEquals(36L, idleStateHandler.tick(TimeUnit.SECONDS)); // 31s + 2s + 2s + 1s
+            idleStateHandler.tick(2L, TimeUnit.SECONDS);
+            assertNotNullAndRelease(channel.consumePart(1024));
+
             // There are no messages left! Advance the ticker by 3 seconds,
             // attempt a consume() but it will be null, then advance the
             // ticker by an another 2 seconds and we should get an IdleStateEvent
@@ -292,7 +317,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
 
             idleStateHandler.tickRun(2L, TimeUnit.SECONDS);
             assertEquals(1, events.size());
-            assertEquals(31L, idleStateHandler.tick(TimeUnit.SECONDS)); // 26s + 3s + 2s
+            assertEquals(43L, idleStateHandler.tick(TimeUnit.SECONDS)); // 36s + 2s + 3s + 2s
 
             // q.e.d.
         } finally {
@@ -379,7 +404,7 @@ protected void doWrite(ChannelOutboundBuffer in) throws Exception {
             // the messages in the ChannelOutboundBuffer.
         }
 
-        public Object consume() {
+        private Object consume() {
             ChannelOutboundBuffer buf = unsafe().outboundBuffer();
             if (buf != null) {
                 Object msg = buf.current();
@@ -391,5 +416,24 @@ public Object consume() {
             }
             return null;
         }
+
+        /**
+         * Consume the part of a message.
+         *
+         * @param byteCount count of byte to be consumed
+         * @return the message currently being consumed
+         */
+        private Object consumePart(int byteCount) {
+            ChannelOutboundBuffer buf = unsafe().outboundBuffer();
+            if (buf != null) {
+                Object msg = buf.current();
+                if (msg != null) {
+                    ReferenceCountUtil.retain(msg);
+                    buf.removeBytes(byteCount);
+                    return msg;
+                }
+            }
+            return null;
+        }
     }
 }