Repository: netty/netty
Issue: Downloading large file halts during transfer
Labels: defect

### Expected behavior
When downloading a large file (50MB) from a netty based server, I expect to receive the entire file and that the download does not halt during the transfer.

### Actual behavior
Download of file halts during the transfer. Transfer eventually terminates due to idle timeout. 

### Steps to reproduce
1. Setup a Netty HTTP server using Epoll Edge Triggered
2. Set the low water mark to 8192 bytes (8k) and high to 32768 bytes (32k)
3. Transfer the file using 8k chunks, flushing when unwritable

### Minimal yet complete reproducer code (or URL to code)
Will try to create one

### Netty version
4.1.21.Final

### JVM version (e.g. `java -version`)
java version "1.7.0_80"
Java(TM) SE Runtime Environment (build 1.7.0_80-b15)
Java HotSpot(TM) 64-Bit Server VM (build 24.80-b11, mixed mode)

### OS version (e.g. `uname -a`)
Linux

### Notes
FYI, downgrading to Netty 4.1.17.Final fixes the issue.  I believe the issue has been introduced by https://github.com/netty/netty/commit/b215794de31f28355e4469fcc04782f55076c80c. I use a custom handler to chunk `FullHttpResponse` objects and send them outbound, so it may be an issue with that handler in conjunction with the netty commit mentioned above.

/cc @Scottmitch Since you did the work on that commit

Issue comments:
Have the same issue. For now I fixed it with Increasing chunk size from 8k to 128k. In my case file is 4Mb only. It is reproduced only on specific VM.
thanks for reporting. I have some questions...

- Are you using `FileRegion` or do you cache the file in memory and write `ByteBuf`s (sounds like `ByteBuf` based upon how you are chunking and dealing with `FullHttpResponse`)?
- Do you have an `SslHandler` in the pipeline?
- Can you confirm if the data is actually being written (wire shark, LoggingHandler, etc...)
- Are you not getting channelWritabilityChanged notifications when you would expect to?
@Scottmitch I was using default Netty Static File example with small modifications. 

SslHandler is present and this part is used:

```
ctx.writeAndFlush(new HttpChunkedInput(new ChunkedFile(raf, 8 * 1024)),
                            ctx.newProgressivePromise());
```

> Can you confirm if the data is actually being written (wire shark, LoggingHandler, etc...)

Yes. Progress listener shows that server sends every time the random number of chunks (40-60% of the file, browser confirms that). Tested both via browser and curl. I wasn't able to debug locally as it was reproduced only on 1 VM.


@doom369 - Can you list the precise steps you used to reproduce using the "Netty Static File example" (I'm assuming `HttpStaticFileServer`)? What file did you download, did you change any code, what browser were you using, what transport are you able to reproduce with (the example is hard coded to NIO, but this issue report is using EPOLL)?

I tried with a 100mb file [1], and changed the low/high watermark to 8k/32k and was unable to reproduce with curl/chrome.

[1] `dd if=/dev/zero of=LargeTestFile.img bs=1024 count=0 seek=$[1024*100]`
@Scottmitch correct. I used handler from `HttpStaticFileServer` (just handler). My server also uses epoll along with openssl. Browser chrome. I'll try to cut off some minimal reproducer when I have some time.

Problem file - https://www.dropbox.com/s/cfbjy029veybinx/main.344f92c975457d9a7419.js?dl=0
Smaller files, css and html were downloaded fine.
@Scottmitch - Sent you a reproducer check your mail.

> Are you using FileRegion or do you cache the file in memory and write ByteBufs (sounds like ByteBuf based upon how you are chunking and dealing with FullHttpResponse)?

In memory

> Do you have an SslHandler in the pipeline?

Issue shows up with or without SslHandler

> Can you confirm if the data is actually being written (wire shark, LoggingHandler, etc...)

Sometimes curl receives some of the data, other times, it receives none.

> Are you not getting channelWritabilityChanged notifications when you would expect to?

I assume I am not.  Its hard to determine given my custom chunk encoder.
@blucas - reproducer was very helpful to narrow in on the issue!
@Scottmitch - You're welcome! Glad it helped!

Pull request review comments:
@Scottmitch seems like the code is now basically the same as before here... Consider revert the change of the file ? Seems like the only change that is "new" is the final keyword
consider reverting this as the code basically does the same as before.
yes it is the same as before ... just 1 less LOC and byte code (see below). Maybe the JIT can optimize this ... but simple enough to do it in code.

```java
    public void function1(boolean v) {
        if (!v) {
            return;
        }
        doWork();
    }

    public void function2(boolean v) {
        if (v) {
            doWork();
        }
    }
```

```
  public void function1(boolean);
    Code:
       0: iload_1
       1: ifne          5
       4: return
       5: aload_0
       6: invokevirtual #6                  // Method doWork:()V
       9: return

  public void function2(boolean);
    Code:
       0: iload_1
       1: ifeq          8
       4: aload_0
       5: invokevirtual #6                  // Method doWork:()V
       8: return
```
@Scottmitch I guess at the end it really not matters.. So up to you.

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 758052db70b..8d3dc5bbcbd 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
@@ -507,14 +507,13 @@ EpollRecvByteAllocatorHandle newEpollHandle(RecvByteBufAllocator.ExtendedHandle
         }
 
         @Override
-        protected void flush0() {
+        protected final void flush0() {
             // Flush immediately only when there's no pending flush.
             // If there's a pending flush operation, event loop will call forceFlush() later,
             // and thus there's no need to call it now.
-            if (isFlagSet(Native.EPOLLOUT)) {
-                return;
+            if (!isFlagSet(Native.EPOLLOUT)) {
+                super.flush0();
             }
-            super.flush0();
         }
 
         /**
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
index e077b2903b7..0c0710f8fbf 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
@@ -445,6 +445,12 @@ protected void doWrite(ChannelOutboundBuffer in) throws Exception {
         } while (writeSpinCount > 0);
 
         if (writeSpinCount == 0) {
+            // It is possible that we have set EPOLLOUT, woken up by EPOLL because the socket is writable, and then use
+            // our write quantum. In this case we no longer want to set the EPOLLOUT flag because the socket is still
+            // writable (as far as we know). We will find out next time we attempt to write if the socket is writable
+            // and set the EPOLLOUT if necessary.
+            clearFlag(Native.EPOLLOUT);
+
             // We used our writeSpin quantum, and should try to write again later.
             eventLoop().execute(flushTask);
         } else {
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 df0cd6e7fa6..fbdbc4b71a6 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
@@ -498,6 +498,16 @@ public KQueueRecvByteAllocatorHandle recvBufAllocHandle() {
             return allocHandle;
         }
 
+        @Override
+        protected final void flush0() {
+            // Flush immediately only when there's no pending flush.
+            // If there's a pending flush operation, event loop will call forceFlush() later,
+            // and thus there's no need to call it now.
+            if (!writeFilterEnabled) {
+                super.flush0();
+            }
+        }
+
         final void executeReadReadyRunnable(ChannelConfig config) {
             if (readReadyRunnablePending || !isActive() || shouldBreakReadReady(config)) {
                 return;
diff --git a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueStreamChannel.java b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueStreamChannel.java
index 50d2cf9725f..ef5f48e820a 100644
--- a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueStreamChannel.java
+++ b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueStreamChannel.java
@@ -283,6 +283,12 @@ protected void doWrite(ChannelOutboundBuffer in) throws Exception {
         } while (writeSpinCount > 0);
 
         if (writeSpinCount == 0) {
+            // It is possible that we have set the write filter, woken up by KQUEUE because the socket is writable, and
+            // then use our write quantum. In this case we no longer want to set the write filter because the socket is
+            // still writable (as far as we know). We will find out next time we attempt to write if the socket is
+            // writable and set the write filter if necessary.
+            writeFilter(false);
+
             // We used our writeSpin quantum, and should try to write again later.
             eventLoop().execute(flushTask);
         } else {
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 8570b3d936a..5987dff2d7a 100644
--- a/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
+++ b/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
@@ -46,7 +46,12 @@ public abstract class AbstractNioByteChannel extends AbstractNioChannel {
             " (expected: " + StringUtil.simpleClassName(ByteBuf.class) + ", " +
             StringUtil.simpleClassName(FileRegion.class) + ')';
 
-    private Runnable flushTask;
+    private final Runnable flushTask = new Runnable() {
+        @Override
+        public void run() {
+            flush();
+        }
+    };
 
     /**
      * Create a new instance
@@ -266,16 +271,13 @@ protected final void incompleteWrite(boolean setOpWrite) {
         if (setOpWrite) {
             setOpWrite();
         } else {
+            // It is possible that we have set the write OP, woken up by NIO because the socket is writable, and then
+            // use our write quantum. In this case we no longer want to set the write OP because the socket is still
+            // writable (as far as we know). We will find out next time we attempt to write if the socket is writable
+            // and set the write OP if necessary.
+            clearOpWrite();
+
             // Schedule flush again later so other tasks can be picked up in the meantime
-            Runnable flushTask = this.flushTask;
-            if (flushTask == null) {
-                flushTask = this.flushTask = new Runnable() {
-                    @Override
-                    public void run() {
-                        flush();
-                    }
-                };
-            }
             eventLoop().execute(flushTask);
         }
     }
diff --git a/transport/src/main/java/io/netty/channel/nio/AbstractNioChannel.java b/transport/src/main/java/io/netty/channel/nio/AbstractNioChannel.java
index f9ea574347b..01467b98361 100644
--- a/transport/src/main/java/io/netty/channel/nio/AbstractNioChannel.java
+++ b/transport/src/main/java/io/netty/channel/nio/AbstractNioChannel.java
@@ -356,10 +356,9 @@ protected final void flush0() {
             // Flush immediately only when there's no pending flush.
             // If there's a pending flush operation, event loop will call forceFlush() later,
             // and thus there's no need to call it now.
-            if (isFlushPending()) {
-                return;
+            if (!isFlushPending()) {
+                super.flush0();
             }
-            super.flush0();
         }
 
         @Override


Test patch:
diff --git a/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketConditionalWritabilityTest.java b/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketConditionalWritabilityTest.java
new file mode 100644
index 00000000000..d5265a1a4c5
--- /dev/null
+++ b/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketConditionalWritabilityTest.java
@@ -0,0 +1,124 @@
+/*
+ * Copyright 2018 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.testsuite.transport.socket;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.WriteBufferWaterMark;
+import io.netty.util.ReferenceCountUtil;
+import org.junit.Test;
+
+import java.util.concurrent.CountDownLatch;
+
+public class SocketConditionalWritabilityTest extends AbstractSocketTest {
+    @Test(timeout = 30000)
+    public void testConditionalWritability() throws Throwable {
+        run();
+    }
+
+    public void testConditionalWritability(ServerBootstrap sb, Bootstrap cb) throws Throwable {
+        Channel serverChannel = null;
+        Channel clientChannel = null;
+        try {
+            final int expectedBytes = 100 * 1024 * 1024;
+            final int maxWriteChunkSize = 16 * 1024;
+            final CountDownLatch latch = new CountDownLatch(1);
+            sb.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(8 * 1024, 16 * 1024));
+            sb.childHandler(new ChannelInitializer<Channel>() {
+                @Override
+                protected void initChannel(Channel ch) {
+                    ch.pipeline().addLast(new ChannelDuplexHandler() {
+                        private int bytesWritten;
+
+                        @Override
+                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
+                            ReferenceCountUtil.release(msg);
+                            writeRemainingBytes(ctx);
+                        }
+
+                        @Override
+                        public void flush(ChannelHandlerContext ctx) {
+                            if (ctx.channel().isWritable()) {
+                                writeRemainingBytes(ctx);
+                            } else {
+                                ctx.flush();
+                            }
+                        }
+
+                        @Override
+                        public void channelWritabilityChanged(ChannelHandlerContext ctx) {
+                            if (ctx.channel().isWritable()) {
+                                writeRemainingBytes(ctx);
+                            }
+                            ctx.fireChannelWritabilityChanged();
+                        }
+
+                        private void writeRemainingBytes(ChannelHandlerContext ctx) {
+                            while (ctx.channel().isWritable() && bytesWritten < expectedBytes) {
+                                int chunkSize = Math.min(expectedBytes - bytesWritten, maxWriteChunkSize);
+                                bytesWritten += chunkSize;
+                                ctx.write(ctx.alloc().buffer(chunkSize).writeZero(chunkSize));
+                            }
+                            ctx.flush();
+                        }
+                    });
+                }
+            });
+
+            serverChannel = sb.bind().syncUninterruptibly().channel();
+
+            cb.handler(new ChannelInitializer<Channel>() {
+                @Override
+                protected void initChannel(Channel ch) {
+                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
+                        private int totalRead;
+                        @Override
+                        public void channelActive(ChannelHandlerContext ctx) {
+                            ctx.writeAndFlush(ctx.alloc().buffer(1).writeByte(0));
+                        }
+
+                        @Override
+                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
+                            if (msg instanceof ByteBuf) {
+                                totalRead += ((ByteBuf) msg).readableBytes();
+                                if (totalRead == expectedBytes) {
+                                    latch.countDown();
+                                }
+                            }
+                            ReferenceCountUtil.release(msg);
+                        }
+                    });
+                }
+            });
+            clientChannel = cb.connect(serverChannel.localAddress()).syncUninterruptibly().channel();
+            latch.await();
+        } finally {
+            if (serverChannel != null) {
+                serverChannel.close();
+            }
+            if (clientChannel != null) {
+                clientChannel.close();
+            }
+        }
+    }
+}
diff --git a/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollETSocketConditionalWritabilityTest.java b/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollETSocketConditionalWritabilityTest.java
new file mode 100644
index 00000000000..773e3d63c83
--- /dev/null
+++ b/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollETSocketConditionalWritabilityTest.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2018 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.channel.epoll;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.testsuite.transport.TestsuitePermutation;
+import io.netty.testsuite.transport.socket.SocketConditionalWritabilityTest;
+
+import java.util.List;
+
+public class EpollETSocketConditionalWritabilityTest extends SocketConditionalWritabilityTest {
+    @Override
+    protected List<TestsuitePermutation.BootstrapComboFactory<ServerBootstrap, Bootstrap>> newFactories() {
+        return EpollSocketTestPermutation.INSTANCE.socket();
+    }
+
+    @Override
+    protected void configure(ServerBootstrap bootstrap, Bootstrap bootstrap2, ByteBufAllocator allocator) {
+        super.configure(bootstrap, bootstrap2, allocator);
+        bootstrap.option(EpollChannelOption.EPOLL_MODE, EpollMode.EDGE_TRIGGERED)
+                .childOption(EpollChannelOption.EPOLL_MODE, EpollMode.EDGE_TRIGGERED);
+        bootstrap2.option(EpollChannelOption.EPOLL_MODE, EpollMode.EDGE_TRIGGERED);
+    }
+}
diff --git a/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollLTSocketConditionalWritabilityTest.java b/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollLTSocketConditionalWritabilityTest.java
new file mode 100644
index 00000000000..31cb81964da
--- /dev/null
+++ b/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollLTSocketConditionalWritabilityTest.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2018 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.channel.epoll;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.testsuite.transport.TestsuitePermutation;
+import io.netty.testsuite.transport.socket.SocketConditionalWritabilityTest;
+
+import java.util.List;
+
+public class EpollLTSocketConditionalWritabilityTest extends SocketConditionalWritabilityTest {
+    @Override
+    protected List<TestsuitePermutation.BootstrapComboFactory<ServerBootstrap, Bootstrap>> newFactories() {
+        return EpollSocketTestPermutation.INSTANCE.socket();
+    }
+
+    @Override
+    protected void configure(ServerBootstrap bootstrap, Bootstrap bootstrap2, ByteBufAllocator allocator) {
+        super.configure(bootstrap, bootstrap2, allocator);
+        bootstrap.option(EpollChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED)
+                .childOption(EpollChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);
+        bootstrap2.option(EpollChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);
+    }
+}
diff --git a/transport-native-kqueue/src/test/java/io/netty/channel/kqueue/KQueueETSocketConditionalWritabilityTest.java b/transport-native-kqueue/src/test/java/io/netty/channel/kqueue/KQueueETSocketConditionalWritabilityTest.java
new file mode 100644
index 00000000000..1c3da7485ef
--- /dev/null
+++ b/transport-native-kqueue/src/test/java/io/netty/channel/kqueue/KQueueETSocketConditionalWritabilityTest.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2018 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.channel.kqueue;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.testsuite.transport.TestsuitePermutation;
+import io.netty.testsuite.transport.socket.SocketConditionalWritabilityTest;
+
+import java.util.List;
+
+public class KQueueETSocketConditionalWritabilityTest extends SocketConditionalWritabilityTest {
+    @Override
+    protected List<TestsuitePermutation.BootstrapComboFactory<ServerBootstrap, Bootstrap>> newFactories() {
+        return KQueueSocketTestPermutation.INSTANCE.socket();
+    }
+}