Improve connection pool to correctly handle concurrency connection up…

…dates
eclipse-vertx · Nov 11, 2017 · 4c69f70 · 4c69f70
1 parent 0e72898
commit 4c69f70
Show file tree

Hide file tree

Showing 11 changed files with 116 additions and 69 deletions.
diff --git a/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java b/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java
@@ -460,7 +460,7 @@ public void reset(long code) {
   private void requestEnded() {
     context.runOnContext(v -> {
       if (pipelining && requests.size() < pipeliningLimit) {
-        listener.onRecycle(this);
+        listener.recycle(this);
       }
     });
   }
@@ -471,7 +471,7 @@ private void responseEnded() {
     } else {
       context.runOnContext(v -> {
         if (currentRequest == null) {
-          listener.onRecycle(this);
+          listener.recycle(this);
         }
       });
     }
@@ -630,7 +630,7 @@ protected void handleMessage(NetSocketImpl connection, ContextImpl context, Chan
         ByteBuf buf = (ByteBuf) msg;
         connection.handleMessageReceived(buf);
       }
-    }.removeHandler(sock -> listener.onClose(this)));
+    }.removeHandler(sock -> listener.closed(this)));
     return socket;
   }
 

diff --git a/src/main/java/io/vertx/core/http/impl/Http2ClientConnection.java b/src/main/java/io/vertx/core/http/impl/Http2ClientConnection.java
@@ -70,8 +70,8 @@ public Channel channel() {
   }
 
   @Override
-  protected void onConcurrencyChange() {
+  protected void concurrencyChanged() {
-    listener.onRecycle(this);
+    listener.concurrencyChanged(this);
   }
 
   @Override
@@ -82,7 +82,7 @@ public HttpClientMetrics metrics() {
   @Override
   void onStreamClosed(Http2Stream nettyStream) {
     super.onStreamClosed(nettyStream);
-    listener.onRecycle(this);
+    listener.recycle(this);
   }
 
   public synchronized HttpClientStream createStream() throws Http2Exception {

diff --git a/src/main/java/io/vertx/core/http/impl/Http2ConnectionBase.java b/src/main/java/io/vertx/core/http/impl/Http2ConnectionBase.java
@@ -91,12 +91,14 @@ static ByteBuf safeBuffer(ByteBuf buf, ByteBufAllocator allocator) {
   private boolean closed;
   private boolean goneAway;
   private int windowSize;
+  private long maxConcurrentStreams;
 
   public Http2ConnectionBase(ContextImpl context, VertxHttp2ConnectionHandler handler) {
     super(context.owner(), handler.context(), context);
     this.handler = handler;
     this.handlerContext = chctx;
     this.windowSize = handler.connection().local().flowController().windowSize(handler.connection().connectionStream());
+    this.maxConcurrentStreams = io.vertx.core.http.Http2Settings.DEFAULT_MAX_CONCURRENT_STREAMS;
   }
 
   VertxInternal vertx() {
@@ -210,12 +212,17 @@ public synchronized void onSettingsAckRead(ChannelHandlerContext ctx) {
   protected void onConnect() {
   }
 
-  protected void onConcurrencyChange() {
+  protected void concurrencyChanged() {
   }
 
   @Override
   public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) {
-    boolean update = !Objects.equals(remoteSettings.maxConcurrentStreams(), settings.maxConcurrentStreams());
+    boolean changed = false;
+    if (settings.maxConcurrentStreams() != null) {
+      long val = settings.maxConcurrentStreams();
+      changed = val != maxConcurrentStreams;
+      maxConcurrentStreams = val;
+    }
     remoteSettings = settings;
     synchronized (this) {
       Handler<io.vertx.core.http.Http2Settings> handler = remoteSettingsHandler;
@@ -225,8 +232,8 @@ public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) {
         });
       }
     }
-    if (update) {
+    if (changed) {
-      onConcurrencyChange();
+      concurrencyChanged();
     }
   }
 

diff --git a/src/main/java/io/vertx/core/http/impl/HttpChannelConnector.java b/src/main/java/io/vertx/core/http/impl/HttpChannelConnector.java
@@ -262,7 +262,7 @@ private void http1xConnected(ConnectionListener<HttpClientConnection> listener,
       handler.handle(Future.succeededFuture(conn));
     });
     clientHandler.removeHandler(conn -> {
-      listener.onClose(conn);
+      listener.closed(conn);
     });
     ch.pipeline().addLast("handler", clientHandler);
   }
@@ -297,7 +297,7 @@ private void http2Connected(ConnectionListener<HttpClientConnection> listener,
         if (metrics != null) {
           metrics.endpointDisconnected(endpointMetric, conn.metric());
         }
-        listener.onClose(conn);
+        listener.closed(conn);
       });
     } catch (Exception e) {
       connectFailed(ch, resultHandler, e);

diff --git a/src/main/java/io/vertx/core/http/impl/HttpClientImpl.java b/src/main/java/io/vertx/core/http/impl/HttpClientImpl.java
@@ -957,10 +957,11 @@ private void getConnectionForWebsocket(boolean ssl,
       public void initConnection(HttpClientConnection conn) {
       }
       @Override
-      public void handleConnection(HttpClientConnection conn) {
+      public boolean handleConnection(HttpClientConnection conn) {
         conn.getContext().executeFromIO(() -> {
           handler.handle((Http1xClientConnection) conn);
         });
+        return true;
       }
       @Override
       public void handleFailure(ContextInternal ctx, Throwable failure) {
@@ -972,10 +973,6 @@ public void handleFailure(ContextInternal ctx, Throwable failure) {
           connectionExceptionHandler.handle(failure);
         }
       }
-      @Override
-      public boolean isCancelled() {
-        return false;
-      }
     });
   }
 

diff --git a/src/main/java/io/vertx/core/http/impl/HttpClientRequestImpl.java b/src/main/java/io/vertx/core/http/impl/HttpClientRequestImpl.java
@@ -667,20 +667,19 @@ public void initConnection(HttpClientConnection conn) {
         }
 
         @Override
-        public void handleConnection(HttpClientConnection conn) throws Exception {
+        public boolean handleConnection(HttpClientConnection conn) throws Exception {
+          // No need to synchronize as the thread is the same that set exceptionOccurred to true
+          // exceptionOccurred=true getting the connection => it's a TimeoutException
+          if (exceptionOccurred != null || reset != null) {
+            return false;
+          }
           HttpClientStream stream;
           stream = conn.createStream();
           ContextImpl context = conn.getContext();
           context.executeFromIO(() -> {
             connected(stream, headersCompletionHandler);
           });
-        }
+          return true;
-
-        @Override
-        public boolean isCancelled() {
-          // No need to synchronize as the thread is the same that set exceptionOccurred to true
-          // exceptionOccurred=true getting the connection => it's a TimeoutException
-          return exceptionOccurred != null || reset != null;
         }
       };
 

diff --git a/src/main/java/io/vertx/core/http/impl/HttpConnectionPool.java b/src/main/java/io/vertx/core/http/impl/HttpConnectionPool.java
@@ -90,11 +90,11 @@ public int maxSize() {
   }
 
   @Override
-  public void initConnection(HttpClientConnection conn) {
+  public boolean initConnection(HttpClientConnection conn) {
     if (conn instanceof Http1xClientConnection && current instanceof Http2) {
       fallbackToHttp1x(((Http1xClientConnection) conn).version());
     }
-    ((ConnectionPool<HttpClientConnection>) current).initConnection(conn);
+    return ((ConnectionPool<HttpClientConnection>) current).initConnection(conn);
   }
 
   @Override
@@ -107,6 +107,10 @@ public void evictConnection(HttpClientConnection conn) {
     ((ConnectionPool<HttpClientConnection>) current).evictConnection(conn);
   }
 
+  @Override
+  public void close() {
+  }
+
   class Http2 implements ConnectionPool<Http2ClientConnection> {
 
     // Pools must locks on the queue object to keep a single lock
@@ -164,8 +168,14 @@ public void evictConnection(Http2ClientConnection conn) {
     }
 
     @Override
-    public void initConnection(Http2ClientConnection conn) {
+    public boolean initConnection(Http2ClientConnection conn) {
       allConnections.add(conn);
+      if (canReserveStream(conn)) {
+        conn.streamCount++;
+        return true;
+      } else {
+        return false;
+      }
     }
 
     @Override
@@ -182,6 +192,10 @@ public boolean isValid(Http2ClientConnection conn) {
     public ContextImpl getContext(Http2ClientConnection conn) {
       return conn.getContext();
     }
+
+    @Override
+    public void close() {
+    }
   }
 
   private class Http1x implements ConnectionPool<Http1xClientConnection> {
@@ -218,8 +232,9 @@ public boolean canCreateConnection(int connCount) {
     }
 
     @Override
-    public void initConnection(Http1xClientConnection conn) {
+    public boolean initConnection(Http1xClientConnection conn) {
       allConnections.add(conn);
+      return true;
     }
 
     @Override
@@ -255,5 +270,9 @@ public boolean isValid(Http1xClientConnection conn) {
     public ContextImpl getContext(Http1xClientConnection conn) {
       return conn.getContext();
     }
+
+    @Override
+    public void close() {
+    }
   }
 }
diff --git a/src/main/java/io/vertx/core/http/impl/pool/ConnectionListener.java b/src/main/java/io/vertx/core/http/impl/pool/ConnectionListener.java
@@ -20,8 +20,25 @@
  */
 public interface ConnectionListener<C> {
 
-  void onRecycle(C conn);
+  /**
+   * The connection concurrency changed.
+   *
+   * @param conn the connection
+   */
+  void concurrencyChanged(C conn);
 
-  void onClose(C conn);
+  /**
+   * The connection can be recycled.
+   *
+   * @param conn the connection
+   */
+  void recycle(C conn);
+
+  /**
+   * The connection is closed.
+   *
+   * @param conn the connection
+   */
+  void closed(C conn);
 
 }
diff --git a/src/main/java/io/vertx/core/http/impl/pool/ConnectionManager.java b/src/main/java/io/vertx/core/http/impl/pool/ConnectionManager.java
@@ -17,10 +17,8 @@
 package io.vertx.core.http.impl.pool;
 
 import io.netty.channel.Channel;
-import io.vertx.core.Vertx;
 import io.vertx.core.http.ConnectionPoolTooBusyException;
 import io.vertx.core.impl.ContextImpl;
-import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.logging.Logger;
 import io.vertx.core.logging.LoggerFactory;
 import io.vertx.core.net.SocketAddress;
@@ -171,27 +169,27 @@ private synchronized void checkPending() {
         if (metric != null) {
           metrics.dequeueRequest(metric, waiter.metric);
         }
-        if (waiter.isCancelled()) {
+        C conn = pool.pollConnection();
+        if (conn != null) {
           waiters.poll();
-        } else {
+          ContextImpl ctx = pool.getContext(conn);
-          C conn = pool.pollConnection();
+          ctx.nettyEventLoop().execute(() -> {
-          if (conn != null) {
+            if (pool.isValid(conn)) {
-            waiters.poll();
+              boolean handled = deliverInternal(conn, waiter);
-            ContextImpl ctx = pool.getContext(conn);
+              if (!handled) {
-            ctx.nettyEventLoop().execute(() -> {
+                pool.recycleConnection(conn);
-              if (pool.isValid(conn)) {
+                checkPending();
-                deliverInternal(conn, waiter);
-              } else {
-                onClose(conn);
-                getConnection(waiter);
               }
-            });
+            } else {
-          } else if (pool.canCreateConnection(connCount)) {
+              closed(conn);
-            waiters.poll();
+              getConnection(waiter);
-            createConnection(waiter);
+            }
-          } else {
+          });
-            break;
+        } else if (pool.canCreateConnection(connCount)) {
-          }
+          waiters.poll();
+          createConnection(waiter);
+        } else {
+          break;
         }
       }
     }
@@ -216,13 +214,18 @@ private void createConnection(Waiter<C> waiter) {
     }
 
     @Override
-    public synchronized void onRecycle(C conn) {
+    public void concurrencyChanged(C conn) {
+      checkPending();
+    }
+
+    @Override
+    public void recycle(C conn) {
       pool.recycleConnection(conn);
       checkPending();
     }
 
     @Override
-    public synchronized void onClose(C conn) {
+    public synchronized void closed(C conn) {
       Channel channel = connector.channel(conn);
       connectionMap.remove(channel);
       pool.evictConnection(conn);
@@ -231,34 +234,42 @@ public synchronized void onClose(C conn) {
 
     private synchronized void initConnection(Waiter<C> waiter, C conn) {
       connectionMap.put(connector.channel(conn), conn);
-      pool.initConnection(conn);
+      boolean ok = pool.initConnection(conn);
       waiter.initConnection(conn);
-      if (waiter.isCancelled()) {
+      if (ok) {
-        pool.recycleConnection(conn);
+        boolean handled = deliverInternal(conn, waiter);
+        if (!handled) {
+          pool.recycleConnection(conn);
+        }
+        checkPending();
       } else {
-        deliverInternal(conn, waiter);
+        getConnection(waiter);
       }
-      checkPending();
     }
 
-    private void deliverInternal(C conn, Waiter<C> waiter) {
+    private boolean deliverInternal(C conn, Waiter<C> waiter) {
       try {
-        waiter.handleConnection(conn);
+        return waiter.handleConnection(conn);
       } catch (Exception e) {
-        getConnection(waiter);
+        e.printStackTrace();
+        return true;
       }
     }
 
     // Called if the connection is actually closed OR the connection attempt failed
     synchronized void connectionClosed() {
       connCount--;
       checkPending();
+
+      // CHECK WAITERS
+
       if (connCount == 0) {
         // No waiters and no connections - remove the ConnQueue
         endpointMap.remove(key);
         if (metrics != null) {
           metrics.closeEndpoint(host, port, metric);
         }
+        pool.close();
       }
     }
   }