diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java index aaaa93f40..858d77cc4 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java @@ -155,6 +155,11 @@ public class ChannelManager { // rather than hang (its request-timeout is not scheduled yet at this point). See NettyRequestSender's // HTTP/2 deferral. private final ConcurrentHashMap>> http2ConnectionWaiters = new ConcurrentHashMap<>(); + // Set once, permanently, when the client closes and sweeps its waiters (failHttp2ConnectionWaiters). Read + // by addHttp2ConnectionWaiter to fail-closed against a request that arms a waiter in the window between the + // sweep and nettyTimer.stop() — such a waiter would otherwise be neither woken nor timed out, hanging its + // request forever. See addHttp2ConnectionWaiter and NettyRequestSender's Http2ConnectionWaiter.arm(). + private volatile boolean waitersClosed; private AsyncHttpClientHandler wsHandler; private Http2Handler http2Handler; @@ -504,38 +509,73 @@ public void registerHttp2Connection(Object partitionKey, Channel channel) { * (issue #2214) — the per-IP registration key on its own would never wake it. The waiter must be * idempotent — it may be invoked by a registration, by the client-close sweep, or removed and invoked by * its own timeout concurrently. + * + * @return {@code true} if the waiter was registered; {@code false} if the client is already closing, in + * which case it is NOT registered and the caller must fail its request immediately rather than + * arm a timeout (no connection will register and the timer that would fire the deadline is being + * stopped). This closes the window where a request arms a waiter between the + * {@link #failHttp2ConnectionWaiters()} sweep and {@code nettyTimer.stop()} and then hangs forever. */ - public void addHttp2ConnectionWaiter(Object partitionKey, Consumer onConnection) { + public boolean addHttp2ConnectionWaiter(Object partitionKey, Consumer onConnection) { http2ConnectionWaiters.computeIfAbsent(baseKeyOf(partitionKey), k -> ConcurrentHashMap.newKeySet()).add(onConnection); + // Recheck after adding: failHttp2ConnectionWaiters sets waitersClosed BEFORE sweeping, so a waiter that + // is not caught by the sweep (added into a bucket the sweep already passed, or a bucket it recreated) + // observes the flag here and unregisters itself. Every waiter is therefore either swept or fails-closed. + if (waitersClosed) { + removeHttp2ConnectionWaiter(partitionKey, onConnection); + return false; + } + return true; } public void removeHttp2ConnectionWaiter(Object partitionKey, Consumer onConnection) { - Set> waiters = http2ConnectionWaiters.get(baseKeyOf(partitionKey)); - if (waiters != null) { + // Prune an emptied set under the bucket lock so an origin whose deferred requests only ever time out + // (a saturated or unreachable host that never registers an H2 connection) does not retain one empty + // Set per host for the client's lifetime. computeIfPresent serializes with a concurrent + // addHttp2ConnectionWaiter on the same key; a racing add that recreates the set is re-covered by the + // waiter's own re-poll in arm(), so no wakeup is lost. + http2ConnectionWaiters.computeIfPresent(baseKeyOf(partitionKey), (k, waiters) -> { waiters.remove(onConnection); - } + return waiters.isEmpty() ? null : waiters; + }); } private void wakeHttp2ConnectionWaiters(Object partitionKey, Channel channel) { Set> waiters = http2ConnectionWaiters.remove(baseKeyOf(partitionKey)); if (waiters != null) { for (Consumer waiter : waiters) { - waiter.accept(channel); + notifyHttp2ConnectionWaiter(waiter, channel); } } } private void failHttp2ConnectionWaiters() { + // Mark closed before the sweep so a waiter arming concurrently is either swept below or fails-closed in + // addHttp2ConnectionWaiter — it can never remain armed-but-never-notified. + waitersClosed = true; for (Object key : http2ConnectionWaiters.keySet()) { Set> waiters = http2ConnectionWaiters.remove(key); if (waiters != null) { for (Consumer waiter : waiters) { - waiter.accept(null); + notifyHttp2ConnectionWaiter(waiter, null); } } } } + // Invoking a waiter drives a full request send (writeRequest). Isolate failures: one waiter throwing must + // not starve the remaining waiters, and — since wakeHttp2ConnectionWaiters runs inside + // registerHttp2Connection, on the establishing connection's onSuccess path — must not propagate out and + // abort that request's own semaphore release and writeRequest (which would silently leak a connection + // permit and hang the establishing request). + private static void notifyHttp2ConnectionWaiter(Consumer waiter, Channel channel) { + try { + waiter.accept(channel); + } catch (Throwable t) { + LOGGER.warn("HTTP/2 connection waiter failed", t); + } + } + /** * Fails a request that was queued in {@link Http2ConnectionState} waiting for a stream slot but can * never get one (the connection dropped or started draining). Releases its request body first — diff --git a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java index 79c445628..1ca9af09d 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -1204,7 +1204,14 @@ private final class Http2ConnectionWaiter implements Consumer { } void arm() { - channelManager.addHttp2ConnectionWaiter(waitKey, this); + if (!channelManager.addHttp2ConnectionWaiter(waitKey, this)) { + // The client is closing (the waiter sweep already ran): no connection will register and the + // nettyTimer that would fire our deadline is being stopped, so arming would hang the request. + // Fail now with the original permit exception instead — and before touching nettyTimer, whose + // newTimeout throws once stopped. + accept(null); + return; + } // Assign the deadline before the recheck so it exists (and is cancellable) if a wake races in. deadline = nettyTimer.newTimeout(t -> fireTimeout(), config.getConnectTimeout().toMillis(), TimeUnit.MILLISECONDS); diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2WaiterTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2WaiterTest.java index ebd6a9b98..e2d4815f1 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2WaiterTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2WaiterTest.java @@ -23,6 +23,8 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; +import java.lang.reflect.Field; +import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -133,4 +135,94 @@ public void waiterFailedWithNullOnClose() { timer.stop(); } } + + @Test + @Timeout(unit = TimeUnit.SECONDS, value = 30) + public void throwingWaiterDoesNotStarveOthersOnRegistration() { + Timer timer = new HashedWheelTimer(); + ChannelManager cm = newChannelManager(timer); + Channel channel = new EmbeddedChannel(); + try { + AtomicBoolean survivorWoken = new AtomicBoolean(); + cm.addHttp2ConnectionWaiter(KEY, c -> { + throw new RuntimeException("waiter blew up"); + }); + cm.addHttp2ConnectionWaiter(KEY, c -> survivorWoken.set(true)); + + // Must not propagate the throwing waiter's exception: registerHttp2Connection runs on the + // establishing connection's onSuccess path, which still has to release its permit and write. + cm.registerHttp2Connection(KEY, channel); + + assertTrue(survivorWoken.get(), "a throwing waiter must not starve the other waiters for the key"); + } finally { + channel.close(); + cm.close(); + timer.stop(); + } + } + + @Test + @Timeout(unit = TimeUnit.SECONDS, value = 30) + public void throwingWaiterDoesNotStarveOthersOnClose() { + Timer timer = new HashedWheelTimer(); + ChannelManager cm = newChannelManager(timer); + try { + AtomicBoolean survivorFailed = new AtomicBoolean(); + cm.addHttp2ConnectionWaiter(KEY, c -> { + throw new RuntimeException("waiter blew up"); + }); + cm.addHttp2ConnectionWaiter(KEY, c -> survivorFailed.set(true)); + + cm.close(); // must isolate the throwing waiter and still fail the rest + + assertTrue(survivorFailed.get(), "a throwing waiter must not starve the other waiters on close"); + } finally { + timer.stop(); + } + } + + @Test + @Timeout(unit = TimeUnit.SECONDS, value = 30) + public void waiterAddedAfterCloseFailsClosed() { + Timer timer = new HashedWheelTimer(); + ChannelManager cm = newChannelManager(timer); + try { + cm.close(); // sweeps pending waiters and latches the closed state + + AtomicBoolean woken = new AtomicBoolean(); + boolean registered = cm.addHttp2ConnectionWaiter(KEY, c -> woken.set(true)); + + assertFalse(registered, "adding a waiter after close must fail-closed so the caller fails its request " + + "instead of arming a timeout that never fires"); + assertFalse(woken.get(), "a fail-closed add must not itself invoke the waiter"); + } finally { + timer.stop(); + } + } + + @Test + @Timeout(unit = TimeUnit.SECONDS, value = 30) + public void removingLastWaiterPrunesEmptyEntry() throws Exception { + Timer timer = new HashedWheelTimer(); + ChannelManager cm = newChannelManager(timer); + try { + Consumer waiter = c -> { + }; + cm.addHttp2ConnectionWaiter(KEY, waiter); + cm.removeHttp2ConnectionWaiter(KEY, waiter); + + assertTrue(waiterMap(cm).isEmpty(), + "removing the last waiter for a key must prune the empty set, not retain it for the client's lifetime"); + } finally { + cm.close(); + timer.stop(); + } + } + + @SuppressWarnings("unchecked") + private static Map waiterMap(ChannelManager cm) throws Exception { + Field field = ChannelManager.class.getDeclaredField("http2ConnectionWaiters"); + field.setAccessible(true); + return (Map) field.get(cm); + } }