Improve Http2Pool connection reuse for concurrent acquires

Signed-off-by: zimatars <[email protected]>

Author: zimatars

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

@@ -406,8 +406,11 @@ 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));
+						borrower.deliver(new Http2PooledRef(slot), true);
 						drain();
 					});
 				}
@@ -827,17 +830,31 @@ 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();
-			if (!poolSlot.slot.canOpenStream()) {
-				poolSlot.slot.deactivate();
+
+			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();
+				}
 				pool.addPending(pool.pending, this, true);
 				return;
 			}
 			stopPendingCountdown(true);
-			// Increment concurrency BEFORE deactivate so that canOpenStream() is correct for other threads
-			poolSlot.slot.incrementConcurrencyAndGet();
-			poolSlot.slot.deactivate();
+			if (!alreadyReserved) {
+				// 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()));
@@ -1017,7 +1034,13 @@ private long computeMaxConcurrentStreams() {
 		}
 
 		boolean canOpenStream() {
-			int concurrency = this.concurrency;
+			return canOpenStream(this.concurrency);
+		}
+
+		boolean canOpenStream(int concurrency) {
+			if (concurrency < 0) {
+				return false;
+			}
 			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-2025 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2024-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.
@@ -88,6 +88,11 @@ 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/Http2PoolTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2021-2025 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2021-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.
@@ -144,6 +144,58 @@ void acquireRelease() {
 		}
 	}
 
+	@Test
+	void concurrentAcquireReusesConnection() {
+		AtomicInteger allocator = new AtomicInteger();
+		ConcurrentLinkedQueue<EmbeddedChannel> channels = new ConcurrentLinkedQueue<>();
+
+		PoolBuilder<Connection, PoolConfig<Connection>> poolBuilder =
+				PoolBuilder.from(Mono.fromSupplier(() -> {
+							allocator.incrementAndGet();
+							EmbeddedChannel channel = new EmbeddedChannel(
+									new TestChannelId(),
+									Http2FrameCodecBuilder.forClient().build(),
+									new Http2MultiplexHandler(new ChannelHandlerAdapter() {}));
+							channels.add(channel);
+							return Connection.from(channel);
+						}))
+						.idleResourceReuseLruOrder()
+						.maxPendingAcquireUnbounded()
+						.sizeBetween(0, 2);
+		Http2Pool http2Pool = poolBuilder.build(config -> new Http2Pool(config, null));
+
+		List<PooledRef<Connection>> acquired = new ArrayList<>();
+		try {
+			// Warm-up: allocate a connection and cache it.
+			PooledRef<Connection> warm = http2Pool.acquire().block(Duration.ofSeconds(1));
+			assertThat(warm).isNotNull();
+			warm.release().block(Duration.ofSeconds(1));
+
+			// Don't run pending tasks yet, simulate concurrent acquires before deliver is scheduled.
+			http2Pool.acquire().subscribe(acquired::add);
+			http2Pool.acquire().subscribe(acquired::add);
+
+			assertThat(allocator).as("allocator count").hasValue(1);
+
+			channels.forEach(EmbeddedChannel::runPendingTasks);
+
+			assertThat(acquired).hasSize(2);
+			assertThat(acquired.get(0).poolable())
+					.as("both acquires should share the same H2 connection")
+					.isSameAs(acquired.get(1).poolable());
+
+			for (PooledRef<Connection> ref : acquired) {
+				ref.release().block(Duration.ofSeconds(1));
+			}
+		}
+		finally {
+			for (EmbeddedChannel channel : channels) {
+				channel.finishAndReleaseAll();
+				Connection.from(channel).dispose();
+			}
+		}
+	}
+
 	@Test
 	void doAcquireNotCalledIfBorrowerInScopeCancelledEarly() {
 		AtomicInteger allocator = new AtomicInteger();