From 6bd5d096a27c253a29448d5f5aacf7a9f59d398f Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 20 Jul 2026 19:08:30 +0200 Subject: [PATCH 1/3] Run HTTP/2 openers outside lock Reserve HTTP/2 stream slots and dequeue pending openers while holding pendingLock, then invoke opening callbacks after releasing it. This prevents a slow opener or onRequestSend callback from serializing other request submissions while preserving queue and slot accounting. Add deterministic concurrency tests for immediate and queued openers. Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../netty/channel/Http2ConnectionState.java | 27 +++++-- .../channel/Http2ConnectionStateTest.java | 72 +++++++++++++++++++ 2 files changed, 93 insertions(+), 6 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java b/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java index 580ab0799..8a7c49064 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java @@ -129,18 +129,20 @@ public boolean offerPendingOpener(Runnable opener) { * Race-free against {@link #failPendingOpeners}: that method sets {@code closed} and drains the queue under * {@code pendingLock}. An opener enqueued before the drain runs is caught by the drain; an enqueue attempt * sequenced after it observes {@code closed} here (the lock provides the happens-before) and is rejected. - * Either way no opener is left stranded. + * Either way no opener is left stranded. Opener callbacks always run after releasing {@code pendingLock}, + * because opening a stream can invoke user code and must not serialize other request submissions. * * @return {@code true} if the opener was run inline or queued; {@code false} if rejected because the * connection is draining/closed or the pending queue is full (caller must fail the request) */ public boolean offerPendingOpener(NettyResponseFuture future, Runnable opener) { + boolean runOpener = false; synchronized (pendingLock) { if (draining.get() || closed.get()) { return false; } if (tryAcquireStream()) { - opener.run(); + runOpener = true; } else { if (pendingCount >= MAX_PENDING_OPENERS) { return false; @@ -148,22 +150,35 @@ public boolean offerPendingOpener(NettyResponseFuture future, Runnable opener pendingOpeners.add(new PendingOpener(future, opener)); pendingCount++; } - return true; } + if (runOpener) { + opener.run(); + } + return true; } private void drainPendingOpeners() { + List ready = null; synchronized (pendingLock) { // Open as many queued requests as there are now-free stream slots. A single stream completion // frees exactly one slot (so this usually runs one opener), but a SETTINGS frame that RAISES // SETTINGS_MAX_CONCURRENT_STREAMS frees several at once — drain them all here rather than waking // only one and stalling the rest until the next completion (a missed-wakeup; the Issue #2160 // silent-timeout class). tryAcquireStream() enforces the cap and the draining/closed gate, so - // this never over-opens; every poll is under pendingLock, so a non-empty queue always yields a - // non-null opener. + // this never over-opens; reserve each slot and dequeue its opener atomically under pendingLock. while (!pendingOpeners.isEmpty() && tryAcquireStream()) { pendingCount--; - pendingOpeners.poll().opener.run(); + if (ready == null) { + ready = new ArrayList<>(); + } + ready.add(pendingOpeners.poll()); + } + } + // Stream opening can invoke user callbacks such as onRequestSend. Run after releasing pendingLock so + // a slow callback does not serialize unrelated submissions to this HTTP/2 connection. + if (ready != null) { + for (PendingOpener pending : ready) { + pending.opener.run(); } } } diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java index ed7a051c2..0c2e6d0b2 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java @@ -24,6 +24,7 @@ import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -270,6 +271,63 @@ public void pendingOpenerRunsOnRelease() { assertEquals(1, executionCount.get(), "Pending opener should have been executed on release"); } + @Test + public void immediateOpenerRunsOutsidePendingLock() throws Exception { + Http2ConnectionState state = new Http2ConnectionState(); + state.updateMaxConcurrentStreams(2); + CountDownLatch openerStarted = new CountDownLatch(1); + CountDownLatch releaseOpener = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(2); + + try { + Future blockingOffer = executor.submit(() -> + state.offerPendingOpener(blockingOpener(openerStarted, releaseOpener))); + assertTrue(openerStarted.await(5, TimeUnit.SECONDS), "first opener should start"); + + Future competingOffer = executor.submit(() -> state.offerPendingOpener(() -> { })); + assertTrue(competingOffer.get(5, TimeUnit.SECONDS), + "a running opener must not hold pendingLock"); + + releaseOpener.countDown(); + assertTrue(blockingOffer.get(5, TimeUnit.SECONDS)); + state.releaseStream(); + state.releaseStream(); + } finally { + releaseOpener.countDown(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + public void queuedOpenerRunsOutsidePendingLock() throws Exception { + Http2ConnectionState state = new Http2ConnectionState(); + state.updateMaxConcurrentStreams(1); + assertTrue(state.tryAcquireStream()); + CountDownLatch openerStarted = new CountDownLatch(1); + CountDownLatch releaseOpener = new CountDownLatch(1); + state.addPendingOpener(blockingOpener(openerStarted, releaseOpener)); + ExecutorService executor = Executors.newFixedThreadPool(2); + + try { + Future drain = executor.submit(state::releaseStream); + assertTrue(openerStarted.await(5, TimeUnit.SECONDS), "queued opener should start"); + + Future competingOffer = executor.submit(() -> state.offerPendingOpener(() -> { })); + assertTrue(competingOffer.get(5, TimeUnit.SECONDS), + "a drained opener must not hold pendingLock"); + + releaseOpener.countDown(); + drain.get(5, TimeUnit.SECONDS); + state.releaseStream(); + state.releaseStream(); + } finally { + releaseOpener.countDown(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + @Test public void multiplePendingOpenersExecuteInOrder() { Http2ConnectionState state = new Http2ConnectionState(); @@ -897,4 +955,18 @@ public void releasePermitOnceIsAtomicUnderConcurrency() throws InterruptedExcept } assertEquals(rounds, totalReleases.get(), "exactly one release per round"); } + + private static Runnable blockingOpener(CountDownLatch started, CountDownLatch release) { + return () -> { + started.countDown(); + try { + if (!release.await(10, TimeUnit.SECONDS)) { + throw new AssertionError("timed out waiting to release opener"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError("interrupted while waiting to release opener", e); + } + }; + } } From 53bb560af560597d9b6561b823d26799b0a342ac Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Sat, 25 Jul 2026 20:42:20 +0200 Subject: [PATCH 2/3] Preserve queued HTTP/2 openers Preserve queued openers when one callback fails during a SETTINGS-driven drain. Dequeueing and reserving the whole batch up front could strand the remaining requests and leak their stream slots. Keep each callback outside pendingLock while reserving only one opener at a time, and cover successful and exceptional multi-opener drains. Refs #2160 Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../netty/channel/Http2ConnectionState.java | 31 ++++++--------- .../channel/Http2ConnectionStateTest.java | 39 +++++++++++++++++++ 2 files changed, 50 insertions(+), 20 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java b/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java index 8a7c49064..9fb93e625 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java @@ -158,28 +158,19 @@ public boolean offerPendingOpener(NettyResponseFuture future, Runnable opener } private void drainPendingOpeners() { - List ready = null; - synchronized (pendingLock) { - // Open as many queued requests as there are now-free stream slots. A single stream completion - // frees exactly one slot (so this usually runs one opener), but a SETTINGS frame that RAISES - // SETTINGS_MAX_CONCURRENT_STREAMS frees several at once — drain them all here rather than waking - // only one and stalling the rest until the next completion (a missed-wakeup; the Issue #2160 - // silent-timeout class). tryAcquireStream() enforces the cap and the draining/closed gate, so - // this never over-opens; reserve each slot and dequeue its opener atomically under pendingLock. - while (!pendingOpeners.isEmpty() && tryAcquireStream()) { - pendingCount--; - if (ready == null) { - ready = new ArrayList<>(); + while (true) { + PendingOpener pending; + synchronized (pendingLock) { + // A SETTINGS increase can free several slots at once. Reserve and dequeue one opener + // atomically, then repeat so an exception cannot strand an already-dequeued batch. + if (pendingOpeners.isEmpty() || !tryAcquireStream()) { + return; } - ready.add(pendingOpeners.poll()); - } - } - // Stream opening can invoke user callbacks such as onRequestSend. Run after releasing pendingLock so - // a slow callback does not serialize unrelated submissions to this HTTP/2 connection. - if (ready != null) { - for (PendingOpener pending : ready) { - pending.opener.run(); + pendingCount--; + pending = pendingOpeners.poll(); } + // Stream opening can invoke user callbacks such as onRequestSend. + pending.opener.run(); } } diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java index 0c2e6d0b2..b6e4ad8d3 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java @@ -328,6 +328,45 @@ public void queuedOpenerRunsOutsidePendingLock() throws Exception { } } + @Test + public void raisedLimitDrainsMultiplePendingOpenersInOrder() { + Http2ConnectionState state = new Http2ConnectionState(); + state.updateMaxConcurrentStreams(1); + assertTrue(state.tryAcquireStream()); + List executionOrder = new ArrayList<>(); + state.addPendingOpener(() -> executionOrder.add(1)); + state.addPendingOpener(() -> executionOrder.add(2)); + state.addPendingOpener(() -> executionOrder.add(3)); + + state.updateMaxConcurrentStreams(4); + + assertEquals(List.of(1, 2, 3), executionOrder); + assertEquals(4, state.getActiveStreams()); + } + + @Test + public void throwingBatchOpenerLeavesRemainingQueueDrainable() { + Http2ConnectionState state = new Http2ConnectionState(); + state.updateMaxConcurrentStreams(1); + assertTrue(state.tryAcquireStream()); + List executionOrder = new ArrayList<>(); + state.addPendingOpener(() -> { + executionOrder.add(1); + throw new IllegalStateException("boom"); + }); + state.addPendingOpener(() -> executionOrder.add(2)); + state.addPendingOpener(() -> executionOrder.add(3)); + + assertThrows(IllegalStateException.class, () -> state.updateMaxConcurrentStreams(4)); + assertEquals(List.of(1), executionOrder); + assertEquals(2, state.getActiveStreams()); + + state.releaseStream(); + + assertEquals(List.of(1, 2, 3), executionOrder); + assertEquals(3, state.getActiveStreams()); + } + @Test public void multiplePendingOpenersExecuteInOrder() { Http2ConnectionState state = new Http2ConnectionState(); From b63975d4594909d2edfa37617f34eae75a795f1f Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 27 Jul 2026 22:11:06 +0200 Subject: [PATCH 3/3] Clarify HTTP/2 opener invariants Explain why the pending-opener drain must reserve one stream at a time while continuing across all available slots. Document that the exceptional test preserves the prior public-Runnable slot accounting behavior. Refs #2160 Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../netty/channel/Http2ConnectionState.java | 9 ++++++--- .../netty/channel/Http2ConnectionStateTest.java | 1 + 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java b/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java index 9fb93e625..c6f5ce60d 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/Http2ConnectionState.java @@ -158,18 +158,21 @@ public boolean offerPendingOpener(NettyResponseFuture future, Runnable opener } private void drainPendingOpeners() { + // Drain every slot exposed by a SETTINGS increase; stopping after one can strand requests until another + // stream completes (Issue #2160). Reserve and dequeue one opener per iteration under pendingLock: + // tryAcquireStream() enforces capacity and draining/closed gates, the lock makes poll() non-null after + // the emptiness check, and a throwing opener cannot strand a pre-reserved batch. while (true) { PendingOpener pending; synchronized (pendingLock) { - // A SETTINGS increase can free several slots at once. Reserve and dequeue one opener - // atomically, then repeat so an exception cannot strand an already-dequeued batch. if (pendingOpeners.isEmpty() || !tryAcquireStream()) { return; } pendingCount--; pending = pendingOpeners.poll(); } - // Stream opening can invoke user callbacks such as onRequestSend. + // Opening a stream can invoke user callbacks. Run without pendingLock so a slow callback does not + // serialize unrelated submissions to this connection. pending.opener.run(); } } diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java index b6e4ad8d3..7873ab7f7 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/Http2ConnectionStateTest.java @@ -359,6 +359,7 @@ public void throwingBatchOpenerLeavesRemainingQueueDrainable() { assertThrows(IllegalStateException.class, () -> state.updateMaxConcurrentStreams(4)); assertEquals(List.of(1), executionOrder); + // A throwing public Runnable leaks its reserved slot, matching the behavior before this change. assertEquals(2, state.getActiveStreams()); state.releaseStream();