Repository: eclipse-vertx/vert.x

Problem statement:
Socket idle-closed in middle of slowly progressing sendFile() response
I tried streaming music to a music player from vertx using `sendFile()`, but the connection is closed after the time given idleTimeout (I have 60 seconds) set in HttpServerOptions has passed after the request started.

I tracked the socket closing down and it seems this triggers the closing:
`channelIdle(ctx, event);` in `io.netty.handler.timeout.IdleStateHandler.AllIdleTimeoutTask`. I guess some other piece of code forgets to mark the connection "busy" while sendFile is active..

vert.x 3.3.3, netty 4.1.1.Final


Hints:
can you run wireshark to monitor the TCP activity over the wire ?

it could be also that lot of data is sent to the client at once and it buffers more than 60 seconds of music and then the buffer is full which cause no data to be sent.

Ok wireshark showed a lot of data at the beginning and then got throttled by "tcp window full" but kept sending packets of 65kB (local TCP connections allows big packets) every 3.5 seconds or so. Fits the ~140kbit/s quality of the music. At 60.015 seconds after response was sent I got a ClosedChannelException to the sendFile result handler. The data kept flowing though on wireshark. At 290.893 seconds the FIN flag was finally raised in a packet sent from the server and data flow stopped. Finally at 341.009 the buffer was empty in the music player and the client responded with FIN, completing the closure of the connection.

So I did a new test. I created a 120G sparse file using the command `truncate -s 120000000000 zero`. I'm testing on the same machine, so throughput is about 1GB/s so 120G is enough for >100 seconds, well above the 60 seconds idle timeout.

I then ran wget against the url serving that file, downloading at maximum speed all the time. Here's the result:

```
--2016-11-20 19:21:17--  http://localhost:8200/zero
Resolving localhost (localhost)... 127.0.0.1
Connecting to localhost (localhost)|127.0.0.1|:8200... connected.
HTTP request sent, awaiting response... 200 OK
Length: 120000000000 (112G) [application/octet-stream]
Saving to: ‘/dev/null’

/dev/null            70%[=============>      ]  78.56G  1.09GB/s    in 79s     

2016-11-20 19:22:36 (1023 MB/s) - Connection closed at byte 84356054245. Retrying.
```

So the connection still breaks even when sending at full speed.

For reference, the documentation of `idleTimeout` is:

> Set the idle timeout, in seconds. zero means don't timeout. This determines if a connection will timeout and be closed if no data is received within the timeout.

If the Linux sendfile syscall is used to send the file (I didn't manage to find out), then it means that it's not actually possible to monitor whether there is activity on the socket or not in software since the operating system is transfering the whole file in a single call.

I don't know if the bug is in vertx or in netty but the idle timeout originated from netty. If either is indeed using syscalls for the transfer then I guess the idle timeout should be disabled for the time of the syscall..

So I wrote a workaround to use AsyncFile + pump instead -- speed dropped to 300MB/s but no more idle problems..

Another workaround would be to disable the idle timeout and track the idle state myself, then I could still use sendFile.. Would allow me to have separate timeouts for "no request active" and "request active".
can you provide a reproducer for this case, so one can have a look at it ?
@vietj sure, here you go: https://github.com/xkr47/vertx-i1720

See [MainVerticle.java](https://github.com/xkr47/vertx-i1720/blob/master/src/main/java/io/vertx/i1720/MainVerticle.java) for instructions.
This method is apparently used for implementing sendFile by netty: `java.nio.channels.FileChannel.transferTo(...,  socket);` which in the implementation calls a native function with the filedescriptor of the socket, so highly probable that it uses the sendfile operating system call.
yes indeed you are right
So any solutions ? I am also facing similar issue
someone needs to investigate it!

Patch:
diff --git a/src/main/java/io/vertx/core/http/impl/HttpServerResponseImpl.java b/src/main/java/io/vertx/core/http/impl/HttpServerResponseImpl.java
--- a/src/main/java/io/vertx/core/http/impl/HttpServerResponseImpl.java
+++ b/src/main/java/io/vertx/core/http/impl/HttpServerResponseImpl.java
@@ -13,8 +13,10 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelPromise;
 import io.netty.handler.codec.http.*;
+import io.netty.util.concurrent.GenericFutureListener;
 import io.vertx.codegen.annotations.Nullable;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Future;
@@ -449,11 +451,12 @@ private void doSendFile(String filename, long offset, long length, Handler<Async
       }
       prepareHeaders(bytesWritten);
 
+      ChannelFuture channelFuture;
       RandomAccessFile raf = null;
       try {
         raf = new RandomAccessFile(file, "r");
         conn.writeToChannel(new AssembledHttpResponse(head, version, status, headers));
-        conn.sendFile(raf, Math.min(offset, file.length()), contentLength);
+        channelFuture = conn.sendFile(raf, Math.min(offset, file.length()), contentLength);
       } catch (IOException e) {
         try {
           if (raf != null) {
@@ -469,12 +472,14 @@ private void doSendFile(String filename, long offset, long length, Handler<Async
         }
         return;
       }
+      written = true;
 
       // write an empty last content to let the http encoder know the response is complete
-      ChannelPromise channelFuture = conn.channelFuture();
-      conn.writeToChannel(LastHttpContent.EMPTY_LAST_CONTENT, channelFuture);
-      written = true;
+      channelFuture.addListener(future -> {
+        conn.writeToChannel(LastHttpContent.EMPTY_LAST_CONTENT);
+      });
 
+      // signal completion handler when there is one
       if (resultHandler != null) {
         ContextInternal ctx = vertx.getOrCreateContext();
         channelFuture.addListener(future -> {
diff --git a/src/main/java/io/vertx/core/net/NetSocket.java b/src/main/java/io/vertx/core/net/NetSocket.java
--- a/src/main/java/io/vertx/core/net/NetSocket.java
+++ b/src/main/java/io/vertx/core/net/NetSocket.java
@@ -53,6 +53,12 @@ public interface NetSocket extends ReadStream<Buffer>, WriteStream<Buffer> {
   @Override
   NetSocket fetch(long amount);
 
+  /**
+   * {@inheritDoc}
+   * <p>
+   * This handler might be called after the close handler when the socket is paused and there are still
+   * buffers to deliver.
+   */
   @Override
   NetSocket endHandler(Handler<Void> endHandler);
 
diff --git a/src/main/java/io/vertx/core/net/impl/ConnectionBase.java b/src/main/java/io/vertx/core/net/impl/ConnectionBase.java
--- a/src/main/java/io/vertx/core/net/impl/ConnectionBase.java
+++ b/src/main/java/io/vertx/core/net/impl/ConnectionBase.java
@@ -49,6 +49,7 @@ public abstract class ConnectionBase {
    */
   public static final VertxException CLOSED_EXCEPTION = new VertxException("Connection was closed", true);
   private static final Logger log = LoggerFactory.getLogger(ConnectionBase.class);
+  private static final int MAX_REGION_SIZE = 1024 * 1024;
 
   private final VoidChannelPromise voidPromise;
   protected final VertxInternal vertx;
@@ -271,6 +272,38 @@ public boolean isSSL() {
     return chctx.pipeline().get(SslHandler.class) != null;
   }
 
+  /**
+   * Send a file as a file region for zero copy transfer to the socket.
+   *
+   * The implementation splits the file into multiple regions to avoid stalling the pipeline
+   * and producing idle timeouts for very large files.
+   *
+   * @param file the file to send
+   * @param offset the file offset
+   * @param length the file length
+   * @param writeFuture the write future to be completed when the transfer is done or failed
+   */
+  private void sendFileRegion(RandomAccessFile file, long offset, long length, ChannelPromise writeFuture) {
+    if (length < MAX_REGION_SIZE) {
+      writeToChannel(new DefaultFileRegion(file.getChannel(), offset, length), writeFuture);
+    } else {
+      ChannelPromise promise = chctx.newPromise();
+      FileRegion region = new DefaultFileRegion(file.getChannel(), offset, MAX_REGION_SIZE);
+      // Retain explicitly this file region so the underlying channel is not closed by the NIO channel when it
+      // as been sent as we need it again
+      region.retain();
+      writeToChannel(region, promise);
+      promise.addListener(future -> {
+        if (future.isSuccess()) {
+          sendFileRegion(file, offset + MAX_REGION_SIZE, length - MAX_REGION_SIZE, writeFuture);
+        } else {
+          future.cause().printStackTrace();
+          writeFuture.setFailure(future.cause());
+        }
+      });
+    }
+  }
+
   protected ChannelFuture sendFile(RandomAccessFile raf, long offset, long length) throws IOException {
     // Write the content.
     ChannelPromise writeFuture = chctx.newPromise();
@@ -279,8 +312,7 @@ protected ChannelFuture sendFile(RandomAccessFile raf, long offset, long length)
       writeToChannel(new ChunkedFile(raf, offset, length, 8192), writeFuture);
     } else {
       // No encryption - use zero-copy.
-      FileRegion region = new DefaultFileRegion(raf.getChannel(), offset, length);
-      writeToChannel(region, writeFuture);
+      sendFileRegion(raf, offset, length, writeFuture);
     }
     if (writeFuture != null) {
       writeFuture.addListener(fut -> raf.close());
diff --git a/src/main/java/io/vertx/core/net/impl/NetSocketImpl.java b/src/main/java/io/vertx/core/net/impl/NetSocketImpl.java
--- a/src/main/java/io/vertx/core/net/impl/NetSocketImpl.java
+++ b/src/main/java/io/vertx/core/net/impl/NetSocketImpl.java
@@ -90,6 +90,7 @@ public NetSocketImpl(VertxInternal vertx, ChannelHandlerContext channel, SocketA
     pending = new InboundBuffer<>(context);
     pending.drainHandler(v -> doResume());
     pending.handler(NULL_MSG_HANDLER);
+    pending.emptyHandler(v -> checkEnd());
   }
 
   synchronized void registerEventBusHandler() {
@@ -337,7 +338,6 @@ public void end() {
 
   @Override
   protected void handleClosed() {
-    Handler<Void> handler;
     MessageConsumer consumer;
     synchronized (this) {
       if (closed) {
@@ -346,17 +346,24 @@ protected void handleClosed() {
       closed = true;
       consumer = registration;
       registration = null;
-      handler = endHandler;
-    }
-    if (handler != null) {
-      handler.handle(null);
     }
+    checkEnd();
     super.handleClosed();
     if (consumer != null) {
       consumer.unregister();
     }
   }
 
+  private void checkEnd() {
+    Handler<Void> handler;
+    synchronized (this) {
+      if (!closed || pending.size() > 0 || (handler = endHandler) == null) {
+        return;
+      }
+    }
+    handler.handle(null);
+  }
+
   public synchronized void handleMessage(Object msg) {
     checkContext();
     if (!pending.write(msg)) {


Test patch:
diff --git a/src/test/java/io/vertx/core/http/Http1xTest.java b/src/test/java/io/vertx/core/http/Http1xTest.java
--- a/src/test/java/io/vertx/core/http/Http1xTest.java
+++ b/src/test/java/io/vertx/core/http/Http1xTest.java
@@ -31,6 +31,7 @@
 import org.junit.Test;
 
 import java.io.File;
+import java.io.RandomAccessFile;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -4662,4 +4663,36 @@ public void testClosingVertxCloseSharedServers() throws Exception {
       assertTrue(server.isClosed());
     });
   }
+
+  @Test
+  public void testHttpServerWithIdleTimeoutSendChunkedFile() throws Exception {
+    int expected = 16 * 1024 * 1024; // We estimate this will take more than 200ms to transfer with a 1ms pause in chunks
+    File sent = TestUtils.tmpFile(".dat", expected);
+    server.close();
+    server = vertx
+      .createHttpServer(createBaseServerOptions().setIdleTimeout(400).setIdleTimeoutUnit(TimeUnit.MILLISECONDS))
+      .requestHandler(
+        req -> {
+          req.response().sendFile(sent.getAbsolutePath());
+        });
+    startServer();
+    client.getNow(8080, "localhost", "/", resp -> {
+      long now = System.currentTimeMillis();
+      int[] length = {0};
+      resp.handler(buff -> {
+        length[0] += buff.length();
+        resp.pause();
+        vertx.setTimer(1, id -> {
+          resp.resume();
+        });
+      });
+      resp.exceptionHandler(this::fail);
+      resp.endHandler(v -> {
+        assertEquals(expected, length[0]);
+        assertTrue(System.currentTimeMillis() - now > 1000);
+        testComplete();
+      });
+    });
+    await();
+  }
 }
diff --git a/src/test/java/io/vertx/core/http/HttpTest.java b/src/test/java/io/vertx/core/http/HttpTest.java
--- a/src/test/java/io/vertx/core/http/HttpTest.java
+++ b/src/test/java/io/vertx/core/http/HttpTest.java
@@ -131,7 +131,7 @@ public void testListenDomainSocketAddress() throws Exception {
     Assume.assumeTrue("Native transport must be enabled", vx.isNativeTransportEnabled());
     NetClient netClient = vx.createNetClient();
     HttpServer httpserver = vx.createHttpServer().requestHandler(req -> req.response().end());
-    File sockFile = TestUtils.tmpFile("vertx", ".sock");
+    File sockFile = TestUtils.tmpFile(".sock");
     SocketAddress sockAddress = SocketAddress.domainSocketAddress(sockFile.getAbsolutePath());
     httpserver.listen(sockAddress, onSuccess(server -> {
       netClient.connect(sockAddress, onSuccess(sock -> {
diff --git a/src/test/java/io/vertx/core/net/NetTest.java b/src/test/java/io/vertx/core/net/NetTest.java
--- a/src/test/java/io/vertx/core/net/NetTest.java
+++ b/src/test/java/io/vertx/core/net/NetTest.java
@@ -49,10 +49,7 @@
 
 import javax.net.ssl.SSLPeerUnverifiedException;
 import javax.security.cert.X509Certificate;
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.OutputStreamWriter;
+import java.io.*;
 import java.nio.charset.StandardCharsets;
 import java.util.*;
 import java.util.concurrent.*;
@@ -83,7 +80,7 @@ public void setUp() throws Exception {
     super.setUp();
     if (USE_DOMAIN_SOCKETS) {
       assertTrue("Native transport not enabled", USE_NATIVE_TRANSPORT);
-      tmp = TestUtils.tmpFile("vertx", ".sock");
+      tmp = TestUtils.tmpFile(".sock");
       testAddress = SocketAddress.domainSocketAddress(tmp.getAbsolutePath());
     } else {
       testAddress = SocketAddress.inetSocketAddress(1234, "localhost");
@@ -1821,7 +1818,7 @@ public void testSharedServersRoundRobinButFirstStartAndStopServer() throws Excep
   @Test
   public void testClosingVertxCloseSharedServers() throws Exception {
     int numServers = 2;
-    Vertx vertx = Vertx.vertx();
+    Vertx vertx = Vertx.vertx(getOptions());
     List<NetServerImpl> servers = new ArrayList<>();
     for (int i = 0;i < numServers;i++) {
       NetServer server = vertx.createNetServer().connectHandler(so -> {
@@ -3356,6 +3353,91 @@ public void testServerNetSocketShouldBeClosedWhenTheClosedHandlerIsCalled() thro
     await();
   }
 
+  @Test
+  public void testServerWithIdleTimeoutSendChunkedFile() throws Exception {
+    testdleTimeoutSendChunkedFile(true);
+  }
+
+  @Test
+  public void testClientWithIdleTimeoutSendChunkedFile() throws Exception {
+    testdleTimeoutSendChunkedFile(false);
+  }
+
+  private void testdleTimeoutSendChunkedFile(boolean idleOnServer) throws Exception {
+    int expected = 16 * 1024 * 1024; // We estimate this will take more than 200ms to transfer with a 1ms pause in chunks
+    File sent = TestUtils.tmpFile(".dat", expected);
+    server.close();
+    Consumer<NetSocket> sender = so -> {
+      so.sendFile(sent.getAbsolutePath());
+    };
+    Consumer<NetSocket> receiver = so -> {
+      int[] len = { 0 };
+      long now = System.currentTimeMillis();
+      so.handler(buff -> {
+        len[0] += buff.length();
+        so.pause();
+        vertx.setTimer(1, id -> {
+          so.resume();
+        });
+      });
+      so.exceptionHandler(this::fail);
+      so.endHandler(v -> {
+        assertEquals(0, expected - len[0]);
+        assertTrue(System.currentTimeMillis() - now > 200);
+        testComplete();
+      });
+    };
+    server = vertx
+      .createNetServer(new NetServerOptions().setIdleTimeout(200).setIdleTimeoutUnit(TimeUnit.MILLISECONDS))
+      .connectHandler((idleOnServer ? sender : receiver)::accept);
+    startServer();
+    client.close();
+    client = vertx.createNetClient(new NetClientOptions().setIdleTimeout(200).setIdleTimeoutUnit(TimeUnit.MILLISECONDS));
+    client.connect(testAddress, onSuccess(idleOnServer ? receiver : sender));
+    await();
+  }
+
+  @Test
+  public void testHalfCloseCallsEndHandlerAfterBuffersAreDelivered() throws Exception {
+    // Synchronized on purpose
+    StringBuffer expected = new StringBuffer();
+    server.connectHandler(so -> {
+      Context ctx = vertx.getOrCreateContext();
+      for (int i = 0;i < 8;i++) {
+        int val = i;
+        ctx.runOnContext(v -> {
+          String chunk = "chunk-" + val + "\r\n";
+          so.write(chunk);
+          expected.append(chunk);
+        });
+      }
+      ctx.runOnContext(v -> {
+        // This will half close the connection
+        so.close();
+      });
+    });
+    startServer();
+    client.connect(testAddress, "localhost", onSuccess(so -> {
+      so.pause();
+      AtomicBoolean closed = new AtomicBoolean();
+      AtomicBoolean ended = new AtomicBoolean();
+      Buffer received = Buffer.buffer();
+      so.handler(received::appendBuffer);
+      so.closeHandler(v -> {
+        assertFalse(ended.get());
+        assertEquals(Buffer.buffer(), received);
+        closed.set(true);
+        so.resume();
+      });
+      so.endHandler(v -> {
+        assertEquals(expected.toString(), received.toString());
+        ended.set(true);
+        testComplete();
+      });
+    }));
+    await();
+  }
+
   protected void startServer(SocketAddress remoteAddress) throws Exception {
     startServer(remoteAddress, vertx.getOrCreateContext());
   }
diff --git a/src/test/java/io/vertx/core/streams/InboundBufferTest.java b/src/test/java/io/vertx/core/streams/InboundBufferTest.java
--- a/src/test/java/io/vertx/core/streams/InboundBufferTest.java
+++ b/src/test/java/io/vertx/core/streams/InboundBufferTest.java
@@ -17,6 +17,7 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -678,4 +679,23 @@ public void testCheckThatPauseAfterResumeWontDoAnyEmission() {
     });
     await();
   }
+
+  @Test
+  public void testBufferSignalingFullImmediately() {
+    context.runOnContext(v1 -> {
+      buffer = new InboundBuffer<>(context, 0L);
+      List<Integer> emitted = new ArrayList<>();
+      buffer.drainHandler(v -> {
+        assertEquals(Arrays.asList(0, 1), emitted);
+        testComplete();
+      });
+      buffer.handler(emitted::add);
+      assertTrue(emit());
+      assertEquals(Collections.singletonList(0), emitted);
+      buffer.pause();
+      assertFalse(emit());
+      buffer.resume();
+    });
+    await();
+  }
 }
diff --git a/src/test/java/io/vertx/it/TransportTest.java b/src/test/java/io/vertx/it/TransportTest.java
--- a/src/test/java/io/vertx/it/TransportTest.java
+++ b/src/test/java/io/vertx/it/TransportTest.java
@@ -87,7 +87,7 @@ private void testNetServer(VertxOptions options) {
 
   @Test
   public void testDomainSocketServer() throws Exception {
-    File sock = TestUtils.tmpFile("vertx", ".sock");
+    File sock = TestUtils.tmpFile(".sock");
     vertx = Vertx.vertx();
     NetServer server = vertx.createNetServer();
     server.connectHandler(so -> {});
@@ -100,7 +100,7 @@ public void testDomainSocketServer() throws Exception {
 
   @Test
   public void testDomainSocketClient() throws Exception {
-    File sock = TestUtils.tmpFile("vertx", ".sock");
+    File sock = TestUtils.tmpFile(".sock");
     vertx = Vertx.vertx();
     NetClient client = vertx.createNetClient();
     client.connect(SocketAddress.domainSocketAddress(sock.getAbsolutePath()), onFailure(err -> {
diff --git a/src/test/java/io/vertx/test/core/TestUtils.java b/src/test/java/io/vertx/test/core/TestUtils.java
--- a/src/test/java/io/vertx/test/core/TestUtils.java
+++ b/src/test/java/io/vertx/test/core/TestUtils.java
@@ -24,6 +24,7 @@
 import javax.security.cert.X509Certificate;
 import java.io.ByteArrayOutputStream;
 import java.io.File;
+import java.io.RandomAccessFile;
 import java.nio.file.Files;
 import java.util.EnumSet;
 import java.util.List;
@@ -409,12 +410,23 @@ public static String loopbackAddress() {
   /**
    * Create a temp file that does not exists.
    */
-  public static File tmpFile(String prefix, String suffix) throws Exception {
-    File tmp = Files.createTempFile(prefix, suffix).toFile();
+  public static File tmpFile(String suffix) throws Exception {
+    File tmp = Files.createTempFile("vertx", suffix).toFile();
     assertTrue(tmp.delete());
     return tmp;
   }
 
+  /**
+   * Create a temp file that exists and with a specified {@code length}. The file will be deleted at VM exit.
+   */
+  public static File tmpFile(String suffix, long length) throws Exception {
+    File tmp = File.createTempFile("vertx", suffix);
+    tmp.deleteOnExit();
+    RandomAccessFile f = new RandomAccessFile(tmp, "rw");
+    f.setLength(length);
+    return tmp;
+  }
+
   public static TestLoggerFactory testLogging(Runnable runnable) {
     InternalLoggerFactory prev = InternalLoggerFactory.getDefaultFactory();
     TestLoggerFactory factory = new TestLoggerFactory();