Improve Http2Pool connection reuse for concurrent acquires

Signed-off-by: zimatars <11007575+zimatars@users.noreply.github.com>

Author: zimatars

reactor-netty-http/src/main/java/reactor/netty/http/client/Http2Pool.java

@@ -391,7 +391,8 @@ void drainLoop() {
 				// when cached connections are below minimum connections, then allocate a new connection
 				boolean belowMinConnections = minConnections > 0 &&
 						poolConfig.allocationStrategy().permitGranted() < minConnections;
-				Slot slot = belowMinConnections ? null : findConnection(resources);
+				int resourcesCount = idleSize;
+				Slot slot = belowMinConnections ? null : findConnection(resources, resourcesCount);
 				if (slot != null) {
 					Borrower borrower = pollPending(borrowers, true);
 					if (borrower == null || borrower.get()) {
@@ -406,19 +407,18 @@ void drainLoop() {
 						log.debug(format(slot.connection.channel(), "Channel activated"));
 					}
 					ACQUIRED.incrementAndGet(this);
-					// Reserve concurrency and re-offer the slot before async deliver so concurrent acquires can reuse the connection
-					slot.incrementConcurrencyAndGet();
-					slot.deactivate();
 					slot.connection.channel().eventLoop().execute(() -> {
-						borrower.deliver(new Http2PooledRef(slot), true);
+						borrower.deliver(new Http2PooledRef(slot));
 						drain();
 					});
 				}
 				else {
-					int resourcesCount = idleSize;
 					if (minConnections > 0 &&
 							poolConfig.allocationStrategy().permitGranted() >= minConnections &&
-							resourcesCount == 0) {
+							poolConfig.allocationStrategy().permitGranted() > resourcesCount) {
+						// connections allocations were triggered
+					}
+					else if (minConnections == 0 && poolConfig.allocationStrategy().permitGranted() > resourcesCount) {
 						// connections allocations were triggered
 					}
 					else {
@@ -542,8 +542,7 @@ void evictInBackground() {
 		scheduleEviction();
 	}
 
-	@Nullable Slot findConnection(ConcurrentLinkedQueue<Slot> resources) {
-		int resourcesCount = idleSize;
+	@Nullable Slot findConnection(ConcurrentLinkedQueue<Slot> resources, int resourcesCount) {
 		while (resourcesCount > 0) {
 			// There are connections in the queue
 
@@ -837,31 +836,17 @@ public String toString() {
 		}
 
 		void deliver(Http2PooledRef poolSlot) {
-			deliver(poolSlot, false);
-		}
-
-		void deliver(Http2PooledRef poolSlot, boolean alreadyReserved) {
 			assert poolSlot.slot.connection.channel().eventLoop().inEventLoop();
 			poolSlot.slot.updateMaxConcurrentStreams();
-
-			int effectiveConcurrency = poolSlot.slot.concurrency() - (alreadyReserved ? 1 : 0);
-			if (!poolSlot.slot.canOpenStream(effectiveConcurrency)) {
-				if (alreadyReserved) {
-					// Concurrency was reserved in drainLoop, rollback reservation
-					poolSlot.slot.decrementConcurrencyAndGet();
-				}
-				else {
-					poolSlot.slot.deactivate();
-				}
+			if (!poolSlot.slot.canOpenStream()) {
+				poolSlot.slot.deactivate();
 				pool.addPending(pool.pending, this, true);
 				return;
 			}
 			stopPendingCountdown(true);
-			if (!alreadyReserved) {
-				// Increment concurrency BEFORE deactivate so that canOpenStream() is correct for other threads
-				poolSlot.slot.incrementConcurrencyAndGet();
-				poolSlot.slot.deactivate();
-			}
+			// Increment concurrency BEFORE deactivate so that canOpenStream() is correct for other threads
+			poolSlot.slot.incrementConcurrencyAndGet();
+			poolSlot.slot.deactivate();
 			if (get()) {
 				//CANCELLED or timeout reached
 				poolSlot.invalidate().subscribe(aVoid -> {}, e -> Operators.onErrorDropped(e, Context.empty()));
@@ -1052,13 +1037,7 @@ private long computeMaxConcurrentStreams() {
 		}
 
 		boolean canOpenStream() {
-			return canOpenStream(this.concurrency);
-		}
-
-		boolean canOpenStream(int concurrency) {
-			if (concurrency < 0) {
-				return false;
-			}
+			int concurrency = this.concurrency;
 			long max = this.maxConcurrentStreams;
 			// For non-HTTP/2 connections (max == 0), allow opening a stream if concurrency is 0
 			// For HTTP/2 connections, check that we haven't reached max concurrent streams

reactor-netty-http/src/main/java/reactor/netty/http/client/Http3Pool.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2024-2026 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2024-2025 VMware, Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -88,11 +88,6 @@ boolean canOpenStream() {
 			return true;
 		}
 
-		@Override
-		boolean canOpenStream(int concurrency) {
-			return true;
-		}
-
 		@Override
 		boolean goAwayReceived() {
 			ChannelHandlerContext connectionHandlerCtx = http3ClientConnectionHandlerCtx();

reactor-netty-http/src/test/java/reactor/netty/http/client/Http2PooledConnectionProviderCustomMetricsTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2025 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2026 VMware, Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -42,6 +42,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.awaitility.Awaitility.await;
 import static reactor.netty.http.HttpProtocol.H2;
 
 class Http2PooledConnectionProviderCustomMetricsTest extends BaseHttpTest {
@@ -78,7 +79,7 @@ void measureActiveStreamsSize() throws InterruptedException {
 				                  .maxConnections(10)
 				                  .build();
 
-		CountDownLatch latch = new CountDownLatch(5);
+		CountDownLatch latch = new CountDownLatch(1);
 		HttpClient httpClient =
 				createClient(pool, disposableServer::address)
 				        .protocol(H2)
@@ -94,10 +95,14 @@ void measureActiveStreamsSize() throws InterruptedException {
 			                           .subscribe());
 
 			assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
-			assertThat(isRegistered.get()).isTrue();
-			HttpConnectionPoolMetrics actual = metrics.get();
-			assertThat(actual).isNotNull();
-			assertThat(actual.activeStreamSize()).isEqualTo(5);
+
+			await().atMost(5, TimeUnit.SECONDS)
+			       .untilAsserted(() -> {
+			           assertThat(isRegistered.get()).isTrue();
+			           HttpConnectionPoolMetrics actual = metrics.get();
+			           assertThat(actual).isNotNull();
+			           assertThat(actual.activeStreamSize()).isEqualTo(5);
+			       });
 		}
 		finally {
 			pool.disposeLater().block(Duration.ofSeconds(5));

reactor-netty-http/src/test/java/reactor/netty/http/server/HttpServerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2011-2025 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2011-2026 VMware, Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -1854,7 +1854,9 @@ void testGracefulShutdownH2(HttpProtocol[] serverProtocols, HttpProtocol[] clien
 	private void doTestGracefulShutdown(HttpServer server, HttpClient client) throws Exception {
 		CountDownLatch latch1 = new CountDownLatch(2);
 		CountDownLatch latch2 = new CountDownLatch(2);
-		CountDownLatch latchGoAway = new CountDownLatch(2);
+		// HTTP/2 GOAWAY is a connection-level frame. With multiplexing enabled, both concurrent requests
+		// can share the same connection, so we should only expect at least one GOAWAY to be observed.
+		CountDownLatch latchGoAway = new CountDownLatch(1);
 		CountDownLatch latch3 = new CountDownLatch(1);
 		LoopResources loop = LoopResources.create("testGracefulShutdown");
 		group = new DefaultChannelGroup(executor);
@@ -1906,7 +1908,7 @@ public void channelRead(@NotNull ChannelHandlerContext ctx, @NotNull Object msg)
 		// Stop accepting incoming requests, wait at most 3s for the active requests to finish
 		disposableServer.disposeNow();
 
-		assertThat(latchGoAway.await(30, TimeUnit.SECONDS)).as("2 GOAWAY should have been received").isTrue();
+		assertThat(latchGoAway.await(30, TimeUnit.SECONDS)).as("GOAWAY should have been received").isTrue();
 		assertThat(latch2.await(30, TimeUnit.SECONDS)).isTrue();
 
 		// Dispose the event loop