Repository: netty/netty
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



Issue comments:
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

Hi,
I just rerun the test I attached to this issue with netty-all:4.1.35.Final and this bug has not been fixed in #9020 

If the server writes a response that has an `ByteBuf` that is larger then what the client can download in the `IdleStateHandler` idle timeout, it will trigger the timeout.

Pull request review comments:
private ?
nit you can remove the `else` as you return in the if block
nit you can remove the else as you return in the previous if block 
...if nothing was flushed before for the current message or there is no current message
Done
Done
Done
Done
I would perform the assignment to flushProgress and the check directly on the if expression to avoid the currentProgress() load to happen regardless the pendingWriteBytes value
Do you mean to change it to this way ?
```java
if (messageHashCode != lastMessageHashCode ||
                    pendingWriteBytes != lastPendingWriteBytes ||  buf.currentProgress() != lastFlushProgress)
```
This helps to reduce calls to the `currentProgress()` method, but in the if block we need to assign the current progress to `lastFlushProgress `, so there will be an additional call.

How about changing it to this way? Although poor readability.

```java
long flushProgress = lastFlushProgress;
if (messageHashCode != lastMessageHashCode ||
                    pendingWriteBytes != lastPendingWriteBytes ||  (flushProgress = buf.currentProgress()) != lastFlushPorgress) {
    lastFlushPorgress = flushProgress;
}
```
Sorry for my bad explanation of the change request :P
Yes I was meaning exactly the second case you have shown, lgtm
as the is not a volatile access I think I would just keep it as it is as it is easier to maintain .
Or alternative (which may be better) just use a second if block in case the previous branch is true.
@normanmaurer Just like this ?
```java
if (messageHashCode != lastMessageHashCode || pendingWriteBytes != lastPendingWriteBytes) {
    // some code
     if(!first){
        return true;
    }
}

long  flushProgress = buf.currentProgress();
if ( lastFlushProgress != flushProgress) {
    lastFlushProgress = flushProgress;
    if(!first){
      return true;
    }
}
```
@qeesung yes
Okay, let me update it.
thanks a lot! Just make sure you also make it pass in terms of check style as your code has some check style problems ;)
Done :-)

Patch that resolved the issue:
diff --git a/handler/src/main/java/io/netty/handler/timeout/IdleStateHandler.java b/handler/src/main/java/io/netty/handler/timeout/IdleStateHandler.java
index 299e4c7ec79..ee80dcf7e0d 100644
--- 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
index d3a934a8297..e042490f30b 100644
--- 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
index a27364f4393..668f3f1ea67 100644
--- 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;
+        }
     }
 }