[grid] Closing input pipeline when 404 is returned

In some cases the resource is not found, and if we have not read completely the input
stream, a read/write lock will happen since the client gets a response and won't stream
the remaining content. Closing the pipe avoids a lock and leaves the RequestConverter
ready for upcoming requests.

Fixes SeleniumHQ/docker-selenium#1646
Fixes SeleniumHQ/docker-selenium#1605

Author: diemol

java/src/org/openqa/selenium/netty/server/RequestConverter.java

@@ -70,7 +70,7 @@ class RequestConverter extends SimpleChannelInboundHandler<HttpObject> {
   protected void channelRead0(
     ChannelHandlerContext ctx,
     HttpObject msg) throws Exception {
-    LOG.fine("Incoming message: " + msg);
+    LOG.log(Debug.getDebugLogLevel(), "Incoming message: " + msg);
 
     if (msg instanceof io.netty.handler.codec.http.HttpRequest) {
       LOG.log(Debug.getDebugLogLevel(), "Start of http request: " + msg);
@@ -84,7 +84,7 @@ protected void channelRead0(
       }
 
       if (nettyRequest.headers().contains("Sec-WebSocket-Version") &&
-        "upgrade".equals(nettyRequest.headers().get("Connection"))) {
+          "upgrade".equals(nettyRequest.headers().get("Connection"))) {
         // Pass this on to later in the pipeline.
         ReferenceCountUtil.retain(msg);
         ctx.fireChannelRead(msg);
@@ -124,7 +124,7 @@ protected void channelRead0(
     }
 
     if (msg instanceof LastHttpContent) {
-      LOG.fine("Closing input pipe.");
+      LOG.log(Debug.getDebugLogLevel(), "Closing input pipe.");
       PipedOutputStream target = out;
 
       EXECUTOR.submit(() -> {
@@ -139,17 +139,20 @@ protected void channelRead0(
 
   @Override
   public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-    LOG.fine("Closing input pipe, channel became inactive.");
-    PipedOutputStream target = out;
+    LOG.log(Debug.getDebugLogLevel(), "Closing input pipe, channel became inactive.");
+    closeInputPipe();
+    super.channelInactive(ctx);
+  }
 
+  public void closeInputPipe() {
+    PipedOutputStream target = out;
     SHUTDOWN_EXECUTOR.submit(() -> {
       try {
         target.close();
       } catch (IOException e) {
         throw new UncheckedIOException(e);
       }
     });
-    super.channelInactive(ctx);
   }
 
   private HttpRequest createRequest(

java/src/org/openqa/selenium/netty/server/ResponseConverter.java

@@ -18,6 +18,9 @@
 package org.openqa.selenium.netty.server;
 
 import com.google.common.io.ByteStreams;
+
+import org.openqa.selenium.remote.http.HttpResponse;
+
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
@@ -29,7 +32,6 @@
 import io.netty.handler.codec.http.HttpChunkedInput;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.stream.ChunkedStream;
-import org.openqa.selenium.remote.http.HttpResponse;
 
 import java.io.InputStream;
 
@@ -56,6 +58,13 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
     }
 
     HttpResponse seResponse = (HttpResponse) msg;
+    if (seResponse.getStatus() == 404) {
+      // In some cases the resource is not found, and if we have not read completely the input
+      // stream, a read/write lock will happen since the client gets a response and won't stream
+      // the remaining content. Closing the pipe avoids a lock and leaves the RequestConverter
+      // ready for upcoming requests.
+      ((RequestConverter) ctx.pipeline().context("se-request").handler()).closeInputPipe();
+    }
 
     // We may not know how large the response is, but figure it out if we can.
     byte[] ary = new byte[CHUNK_SIZE];