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] 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); + } + }; + } }