Fixes #2088 - Recycle HTTP/2 channels on the client.

Recycled channels also for FCGI.
Small improvements to HTTP/2 too.

Signed-off-by: Simone Bordet <simone.bordet@gmail.com>

jetty-fcgi/fcgi-client/src/main/java/org/eclipse/jetty/fcgi/client/http/HttpChannelOverFCGI.java

@@ -38,18 +38,17 @@ public class HttpChannelOverFCGI extends HttpChannel
 {
     private final HttpConnectionOverFCGI connection;
     private final Flusher flusher;
-    private final int request;
     private final HttpSenderOverFCGI sender;
     private final HttpReceiverOverFCGI receiver;
     private final FCGIIdleTimeout idle;
+    private int request;
     private HttpVersion version;
 
-    public HttpChannelOverFCGI(final HttpConnectionOverFCGI connection, Flusher flusher, int request, long idleTimeout)
+    public HttpChannelOverFCGI(final HttpConnectionOverFCGI connection, Flusher flusher, long idleTimeout)
     {
         super(connection.getHttpDestination());
         this.connection = connection;
         this.flusher = flusher;
-        this.request = request;
         this.sender = new HttpSenderOverFCGI(this);
         this.receiver = new HttpReceiverOverFCGI(this);
         this.idle = new FCGIIdleTimeout(connection, idleTimeout);
@@ -60,6 +59,11 @@ public class HttpChannelOverFCGI extends HttpChannel
         return request;
     }
 
+    void setRequest(int request)
+    {
+        this.request = request;
+    }
+
     @Override
     protected HttpSender getHttpSender()
     {
@@ -72,6 +76,11 @@ public class HttpChannelOverFCGI extends HttpChannel
         return receiver;
     }
 
+    public boolean isFailed()
+    {
+        return sender.isFailed() || receiver.isFailed();
+    }
+
     @Override
     public void send()
     {

jetty-fcgi/fcgi-client/src/main/java/org/eclipse/jetty/fcgi/client/http/HttpConnectionOverFCGI.java

@@ -23,7 +23,9 @@ import java.nio.ByteBuffer;
 import java.nio.channels.AsynchronousCloseException;
 import java.util.LinkedList;
 import java.util.Map;
+import java.util.Queue;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -56,7 +58,8 @@ public class HttpConnectionOverFCGI extends AbstractConnection implements Connec
     private static final Logger LOG = Log.getLogger(HttpConnectionOverFCGI.class);
 
     private final LinkedList<Integer> requests = new LinkedList<>();
-    private final Map<Integer, HttpChannelOverFCGI> channels = new ConcurrentHashMap<>();
+    private final Map<Integer, HttpChannelOverFCGI> activeChannels = new ConcurrentHashMap<>();
+    private final Queue<HttpChannelOverFCGI> idleChannels = new ConcurrentLinkedQueue<>();
     private final AtomicBoolean closed = new AtomicBoolean();
     private final HttpDestination destination;
     private final Promise<Connection> promise;
@@ -184,7 +187,7 @@ public class HttpConnectionOverFCGI extends AbstractConnection implements Connec
     {
         // Close explicitly only if we are idle, since the request may still
         // be in progress, otherwise close only if we can fail the responses.
-        if (channels.isEmpty())
+        if (activeChannels.isEmpty())
             close();
         else
             failAndClose(new EOFException(String.valueOf(getEndPoint())));
@@ -204,8 +207,14 @@ public class HttpConnectionOverFCGI extends AbstractConnection implements Connec
 
     protected void release(HttpChannelOverFCGI channel)
     {
-        channels.remove(channel.getRequest());
+        if (activeChannels.remove(channel.getRequest()) != null)
-        destination.release(this);
+        {
+            channel.setRequest(0);
+            // Recycle only non-failed channels.
+            if (!channel.isFailed())
+                idleChannels.offer(channel);
+            destination.release(this);
+        }
     }
 
     @Override
@@ -249,19 +258,20 @@ public class HttpConnectionOverFCGI extends AbstractConnection implements Connec
 
     protected void abort(Throwable failure)
     {
-        for (HttpChannelOverFCGI channel : channels.values())
+        for (HttpChannelOverFCGI channel : activeChannels.values())
         {
             HttpExchange exchange = channel.getHttpExchange();
             if (exchange != null)
                 exchange.getRequest().abort(failure);
         }
-        channels.clear();
+        activeChannels.clear();
+        idleChannels.clear();
     }
 
     private void failAndClose(Throwable failure)
     {
         boolean result = false;
-        for (HttpChannelOverFCGI channel : channels.values())
+        for (HttpChannelOverFCGI channel : activeChannels.values())
             result |= channel.responseFailure(failure);
         if (result)
             close(failure);
@@ -286,9 +296,18 @@ public class HttpConnectionOverFCGI extends AbstractConnection implements Connec
         }
     }
 
-    protected HttpChannelOverFCGI newHttpChannel(int id, Request request)
+    protected HttpChannelOverFCGI provideHttpChannel(int id, Request request)
     {
-        return new HttpChannelOverFCGI(this, getFlusher(), id, request.getIdleTimeout());
+        HttpChannelOverFCGI channel = idleChannels.poll();
+        if (channel == null)
+            channel = newHttpChannel(request);
+        channel.setRequest(id);
+        return channel;
+    }
+
+    protected HttpChannelOverFCGI newHttpChannel(Request request)
+    {
+        return new HttpChannelOverFCGI(this, getFlusher(), request.getIdleTimeout());
     }
 
     @Override
@@ -314,10 +333,10 @@ public class HttpConnectionOverFCGI extends AbstractConnection implements Connec
             Request request = exchange.getRequest();
             normalizeRequest(request);
 
-            // FCGI may be multiplexed, so create one channel for each request.
+            // FCGI may be multiplexed, so one channel for each exchange.
             int id = acquireRequest();
-            HttpChannelOverFCGI channel = newHttpChannel(id, request);
+            HttpChannelOverFCGI channel = provideHttpChannel(id, request);
-            channels.put(id, channel);
+            activeChannels.put(id, channel);
 
             return send(channel, exchange);
         }
@@ -351,7 +370,7 @@ public class HttpConnectionOverFCGI extends AbstractConnection implements Connec
         @Override
         public void onBegin(int request, int code, String reason)
         {
-            HttpChannelOverFCGI channel = channels.get(request);
+            HttpChannelOverFCGI channel = activeChannels.get(request);
             if (channel != null)
                 channel.responseBegin(code, reason);
             else
@@ -361,7 +380,7 @@ public class HttpConnectionOverFCGI extends AbstractConnection implements Connec
         @Override
         public void onHeader(int request, HttpField field)
         {
-            HttpChannelOverFCGI channel = channels.get(request);
+            HttpChannelOverFCGI channel = activeChannels.get(request);
             if (channel != null)
                 channel.responseHeader(field);
             else
@@ -371,7 +390,7 @@ public class HttpConnectionOverFCGI extends AbstractConnection implements Connec
         @Override
         public void onHeaders(int request)
         {
-            HttpChannelOverFCGI channel = channels.get(request);
+            HttpChannelOverFCGI channel = activeChannels.get(request);
             if (channel != null)
                 channel.responseHeaders();
             else
@@ -385,7 +404,7 @@ public class HttpConnectionOverFCGI extends AbstractConnection implements Connec
             {
                 case STD_OUT:
                 {
-                    HttpChannelOverFCGI channel = channels.get(request);
+                    HttpChannelOverFCGI channel = activeChannels.get(request);
                     if (channel != null)
                     {
                         CompletableCallback callback = new CompletableCallback()
@@ -431,7 +450,7 @@ public class HttpConnectionOverFCGI extends AbstractConnection implements Connec
         @Override
         public void onEnd(int request)
         {
-            HttpChannelOverFCGI channel = channels.get(request);
+            HttpChannelOverFCGI channel = activeChannels.get(request);
             if (channel != null)
             {
                 if (channel.responseSuccess())
@@ -446,7 +465,7 @@ public class HttpConnectionOverFCGI extends AbstractConnection implements Connec
         @Override
         public void onFailure(int request, Throwable failure)
         {
-            HttpChannelOverFCGI channel = channels.get(request);
+            HttpChannelOverFCGI channel = activeChannels.get(request);
             if (channel != null)
             {
                 if (channel.responseFailure(failure))

jetty-http2/http2-http-client-transport/src/main/java/org/eclipse/jetty/http2/client/http/HttpChannelOverHTTP2.java

@@ -100,7 +100,6 @@ public class HttpChannelOverHTTP2 extends HttpChannel
     @Override
     public void release()
     {
-        setStream(null);
         connection.release(this);
     }
 

jetty-http2/http2-http-client-transport/src/main/java/org/eclipse/jetty/http2/client/http/HttpConnectionOverHTTP2.java

@@ -65,7 +65,7 @@ public class HttpConnectionOverHTTP2 extends HttpConnection implements Sweeper.S
         request.version(HttpVersion.HTTP_2);
         normalizeRequest(request);
 
-        // One connection maps to N channels, so for each exchange we create a new channel.
+        // One connection maps to N channels, so one channel for each exchange.
         HttpChannelOverHTTP2 channel = provideHttpChannel();
         activeChannels.add(channel);
 
@@ -76,15 +76,21 @@ public class HttpConnectionOverHTTP2 extends HttpConnection implements Sweeper.S
     {
         HttpChannelOverHTTP2 channel = idleChannels.poll();
         if (channel == null)
-            channel = new HttpChannelOverHTTP2(getHttpDestination(), this, getSession());
+            channel = newHttpChannel();
         return channel;
     }
 
+    protected HttpChannelOverHTTP2 newHttpChannel()
+    {
+        return new HttpChannelOverHTTP2(getHttpDestination(), this, getSession());
+    }
+
     protected void release(HttpChannelOverHTTP2 channel)
     {
         // Only non-push channels are released.
         if (activeChannels.remove(channel))
         {
+            channel.setStream(null);
             // Recycle only non-failed channels.
             if (!channel.isFailed())
                 idleChannels.offer(channel);

jetty-http2/http2-http-client-transport/src/test/java/org/eclipse/jetty/http2/client/http/HttpClientTransportOverHTTP2Test.java

@@ -247,7 +247,7 @@ public class HttpClientTransportOverHTTP2Test extends AbstractTest
                 return new HttpConnectionOverHTTP2(destination, session)
                 {
                     @Override
-                    protected HttpChannelOverHTTP2 provideHttpChannel()
+                    protected HttpChannelOverHTTP2 newHttpChannel()
                     {
                         return new HttpChannelOverHTTP2(getHttpDestination(), this, getSession())
                         {

tests/test-http-client-transport/src/test/java/org/eclipse/jetty/http/client/HttpChannelAssociationTest.java

@@ -27,6 +27,7 @@ import org.eclipse.jetty.client.HttpClientTransport;
 import org.eclipse.jetty.client.HttpDestination;
 import org.eclipse.jetty.client.HttpExchange;
 import org.eclipse.jetty.client.api.Connection;
+import org.eclipse.jetty.client.api.Request;
 import org.eclipse.jetty.client.http.HttpChannelOverHTTP;
 import org.eclipse.jetty.client.http.HttpClientTransportOverHTTP;
 import org.eclipse.jetty.client.http.HttpConnectionOverHTTP;
@@ -146,7 +147,7 @@ public class HttpChannelAssociationTest extends AbstractTest
                         return new HttpConnectionOverHTTP2(destination, session)
                         {
                             @Override
-                            protected HttpChannelOverHTTP2 provideHttpChannel()
+                            protected HttpChannelOverHTTP2 newHttpChannel()
                             {
                                 return new HttpChannelOverHTTP2(getHttpDestination(), this, getSession())
                                 {
@@ -171,9 +172,9 @@ public class HttpChannelAssociationTest extends AbstractTest
                         return new HttpConnectionOverFCGI(endPoint, destination, promise, isMultiplexed())
                         {
                             @Override
-                            protected HttpChannelOverFCGI newHttpChannel(int id, org.eclipse.jetty.client.api.Request request)
+                            protected HttpChannelOverFCGI newHttpChannel(Request request)
                             {
-                                return new HttpChannelOverFCGI(this, getFlusher(), id, request.getIdleTimeout())
+                                return new HttpChannelOverFCGI(this, getFlusher(), request.getIdleTimeout())
                                 {
                                     @Override
                                     public boolean associate(HttpExchange exchange)