The WebSocket client pool should not starve on connect error.

Motivation:

The client WebSocket pool implementation (which is not really a pool but instead a semaphore + a wait queue) does not decrement the semaphore when the connection fails to be established.

Currently, no operation is done beside calling back the caller. As consequence, the wait queue fills with new requests and never satisfy them since.

Changes:

When the connection fails, the semaphore should be released and check the wait queue for pending requests.

src/main/java/io/vertx/core/http/impl/WebSocketEndpoint.java

@@ -47,34 +47,26 @@ private static class Waiter {
     }
   }
 
-  private void tryConnect(ContextInternal ctx, Handler<AsyncResult<HttpClientConnection>> handler) {
+  private void connect(ContextInternal ctx, Handler<AsyncResult<HttpClientConnection>> handler) {
 
     class Listener implements Handler<AsyncResult<HttpClientConnection>> {
 
-      private void onEvict() {
-        decRefCount();
-        Waiter h;
-        synchronized (WebSocketEndpoint.this) {
-          if (--inflightConnections > maxPoolSize || waiters.isEmpty()) {
-            return;
-          }
-          h = waiters.poll();
-        }
-        tryConnect(h.context, h.handler);
-      }
-
       @Override
       public void handle(AsyncResult<HttpClientConnection> ar) {
         if (ar.succeeded()) {
           HttpClientConnection c = ar.result();
           if (incRefCount()) {
-            c.evictionHandler(v -> onEvict());
+            c.evictionHandler(v -> {
+              decRefCount();
+              release();
+            });
             handler.handle(Future.succeededFuture(c));
           } else {
             c.close();
             handler.handle(Future.failedFuture("Connection closed"));
           }
         } else {
+          release();
           handler.handle(Future.failedFuture(ar.cause()));
         }
       }
@@ -91,16 +83,33 @@ public void handle(AsyncResult<HttpClientConnection> ar) {
       .httpConnect(eventLoopContext, new Listener());
   }
 
-  @Override
-  public void requestConnection2(ContextInternal ctx, long timeout, Handler<AsyncResult<HttpClientConnection>> handler) {
+  private void release() {
+    Waiter h;
+    synchronized (WebSocketEndpoint.this) {
+      if (--inflightConnections > maxPoolSize || waiters.isEmpty()) {
+        return;
+      }
+      h = waiters.poll();
+    }
+    connect(h.context, h.handler);
+  }
+
+  private boolean tryAcquire(ContextInternal ctx, Handler<AsyncResult<HttpClientConnection>> handler) {
     synchronized (this) {
       if (inflightConnections >= maxPoolSize) {
         waiters.add(new Waiter(handler, ctx));
-        return;
+        return false;
       }
       inflightConnections++;
     }
-    tryConnect(ctx, handler);
+    return true;
+  }
+
+  @Override
+  public void requestConnection2(ContextInternal ctx, long timeout, Handler<AsyncResult<HttpClientConnection>> handler) {
+    if (tryAcquire(ctx, handler)) {
+      connect(ctx, handler);
+    }
   }
 
   @Override
@@ -110,6 +119,7 @@ void checkExpired() {
 
   @Override
   public void close() {
+    System.out.println("CLOSING POOL " + waiters.size());
     super.close();
     synchronized (this) {
       waiters.forEach(waiter -> {

src/test/java/io/vertx/core/http/WebSocketTest.java

@@ -38,6 +38,7 @@
 import io.vertx.core.net.impl.NetSocketInternal;
 import io.vertx.core.streams.ReadStream;
 import io.vertx.test.core.CheckingSender;
+import io.vertx.test.core.Repeat;
 import io.vertx.test.core.TestUtils;
 import io.vertx.test.core.VertxTestBase;
 import io.vertx.test.proxy.HAProxy;
@@ -47,7 +48,13 @@
 import org.junit.Test;
 
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.io.UnsupportedEncodingException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
 import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
@@ -4026,4 +4033,88 @@ public void testCustomResponseHeadersBeforeUpgrade() {
       }));
       await();
   }
+
+  @Test
+  public void testPoolShouldNotStarveOnConnectError() throws Exception {
+
+    server = vertx.createHttpServer();
+
+    AtomicInteger accepted = new AtomicInteger();
+    server.webSocketHandler(ws -> {
+      assertTrue(accepted.getAndIncrement() == 0);
+    });
+
+    server.listen(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST).toCompletionStage().toCompletableFuture().get();
+
+    // This test requires a server socket to respond for the first connection
+    // Subsequent connections need to fail (connect error)
+    // Hence we need a custom proxy server in front of Vert.x HTTP server
+    InetAddress localhost = InetAddress.getByName("localhost");
+    ServerSocket proxy = new ServerSocket(0, 10, localhost);
+    int proxyPort = proxy.getLocalPort();
+
+    int maxConnections = 5;
+
+    client = vertx.createWebSocketClient(new WebSocketClientOptions()
+      .setMaxConnections(maxConnections)
+      .setConnectTimeout(4000));
+
+    Future<WebSocket> wsFut = client.connect(proxyPort, DEFAULT_HTTP_HOST, "/").andThen(onSuccess(v -> {
+    }));
+
+    Socket inbound = proxy.accept();
+    Socket outbound = new Socket();
+    outbound.connect(new InetSocketAddress(localhost, 8080));
+
+    class Pump extends Thread {
+      final InputStream src;
+      final OutputStream dst;
+      Pump(InputStream src, OutputStream dst) {
+        this.src = src;
+        this.dst = dst;
+      }
+      @Override
+      public void run() {
+        byte[] buffer = new byte[512];
+        while (true) {
+          try {
+            int l = src.read(buffer);
+            if (l == -1) {
+              break;
+            }
+            dst.write(buffer, 0, l);
+          } catch (IOException e) {
+            break;
+          }
+        }
+      }
+    }
+
+    Pump pump1 = new Pump(inbound.getInputStream(), outbound.getOutputStream());
+    Pump pump2 = new Pump(outbound.getInputStream(), inbound.getOutputStream());
+    pump1.start();
+    pump2.start();
+
+    // Finish handshake
+    wsFut.toCompletionStage().toCompletableFuture().get(10, TimeUnit.SECONDS);
+
+    proxy.close();
+
+    try {
+      int num = maxConnections + 10;
+      CountDownLatch latch = new CountDownLatch(num);
+      for (int i = 0;i < num;i++) {
+        client.connect(proxyPort, DEFAULT_HTTP_HOST, "/").onComplete(ar -> {
+          latch.countDown();
+        });
+      }
+
+      awaitLatch(latch, 10, TimeUnit.SECONDS);
+    } finally {
+      inbound.close();
+      outbound.close();
+      pump1.join(10_000);
+      pump2.join(10_000);
+    }
+  }
 }