Http2Pool: when applying acquireTimeout, check for current pending Bo…

…rrowers (#3300)
reactor · Jun 14, 2024 · 10387d6 · 10387d6
1 parent 6e73be2
commit 10387d6
Show file tree

Hide file tree

Showing 2 changed files with 47 additions and 10 deletions.
diff --git a/reactor-netty-http/src/main/java/reactor/netty/http/client/Http2Pool.java b/reactor-netty-http/src/main/java/reactor/netty/http/client/Http2Pool.java
@@ -624,14 +624,24 @@ void pendingAcquireLimitReached(Borrower borrower, int maxPending) {
 	 * @param borrower a new {@link Borrower} to add to the queue and later either serve or consider pending
 	 */
 	void pendingOffer(Borrower borrower) {
-		int maxPending = poolConfig.maxPending();
 		ConcurrentLinkedDeque<Borrower> pendingQueue = pending;
 		if (pendingQueue == TERMINATED) {
 			return;
 		}
+
 		int postOffer = addPending(pendingQueue, borrower, false);
 
+		long estimateStreamsCount = totalMaxConcurrentStreams - acquired;
+		int permits = poolConfig.allocationStrategy().estimatePermitCount();
+		if (permits + estimateStreamsCount < postOffer) {
+			borrower.pendingAcquireStart = clock.millis();
+			if (!borrower.acquireTimeout.isZero()) {
+				borrower.timeoutTask = poolConfig.pendingAcquireTimer().apply(borrower, borrower.acquireTimeout);
+			}
+		}
+
 		if (WIP.getAndIncrement(this) == 0) {
+			int maxPending = poolConfig.maxPending();
 			ConcurrentLinkedQueue<Slot> ir = connections;
 			if (maxPending >= 0 && postOffer > maxPending && ir.isEmpty() && poolConfig.allocationStrategy().estimatePermitCount() == 0) {
 				Borrower toCull = pollPending(pendingQueue, false);
@@ -757,15 +767,6 @@ Context currentContext() {
 		@Override
 		public void request(long n) {
 			if (Operators.validate(n)) {
-				long estimateStreamsCount = pool.totalMaxConcurrentStreams - pool.acquired;
-				int permits = pool.poolConfig.allocationStrategy().estimatePermitCount();
-				int pending = pool.pendingSize;
-				if (permits + estimateStreamsCount <= pending) {
-					pendingAcquireStart = pool.clock.millis();
-					if (!acquireTimeout.isZero()) {
-						timeoutTask = pool.poolConfig.pendingAcquireTimer().apply(this, acquireTimeout);
-					}
-				}
 				pool.doAcquire(this);
 			}
 		}

diff --git a/reactor-netty-http/src/test/java/reactor/netty/http/client/Http2PoolTest.java b/reactor-netty-http/src/test/java/reactor/netty/http/client/Http2PoolTest.java
@@ -40,7 +40,10 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiFunction;
@@ -1206,6 +1209,39 @@ void nonHttp2ConnectionEmittedOnce() {
 		}
 	}
 
+	@Test
+	void pendingTimeout() throws Exception {
+		EmbeddedChannel channel = new EmbeddedChannel();
+		PoolBuilder<Connection, PoolConfig<Connection>> poolBuilder =
+				PoolBuilder.from(Mono.just(Connection.from(channel)))
+				           .maxPendingAcquire(10)
+				           .sizeBetween(0, 1);
+		Http2Pool http2Pool = poolBuilder.build(config -> new Http2Pool(config, null));
+
+		CountDownLatch latch = new CountDownLatch(3);
+		ExecutorService executorService = Executors.newFixedThreadPool(20);
+		try {
+			CompletableFuture<?>[] completableFutures = new CompletableFuture<?>[4];
+			for (int i = 0; i < completableFutures.length; i++) {
+				completableFutures[i] = CompletableFuture.runAsync(
+						() -> http2Pool.acquire(Duration.ofMillis(10))
+								.doOnEach(sig -> channel.runPendingTasks())
+								.doOnError(t -> latch.countDown())
+								.onErrorResume(PoolAcquireTimeoutException.class, t -> Mono.empty())
+								.block(),
+						executorService);
+			}
+
+			CompletableFuture.allOf(completableFutures).join();
+			assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+		}
+		finally {
+			channel.finishAndReleaseAll();
+			Connection.from(channel).dispose();
+			executorService.shutdown();
+		}
+	}
+
 	@Test
 	void recordsPendingCountAndLatencies() {
 		EmbeddedChannel channel = new EmbeddedChannel(new TestChannelId(),