diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/PerHostConnectionSemaphore.java b/client/src/main/java/org/asynchttpclient/netty/channel/PerHostConnectionSemaphore.java index 041b45ac4..75c32a01b 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/PerHostConnectionSemaphore.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/PerHostConnectionSemaphore.java @@ -16,6 +16,8 @@ package org.asynchttpclient.netty.channel; import org.asynchttpclient.exception.TooManyConnectionsPerHostException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.concurrent.ConcurrentHashMap; @@ -29,6 +31,8 @@ */ public class PerHostConnectionSemaphore implements ConnectionSemaphore { + private static final Logger LOGGER = LoggerFactory.getLogger(PerHostConnectionSemaphore.class); + private static final class TrackedSemaphore extends Semaphore { private static final long serialVersionUID = 1L; @@ -89,15 +93,12 @@ public void releaseChannelLock(Object partitionKey) { if (freeConnections != null) { freeConnections.release(); releaseHostReservation(partitionKey, freeConnections); + } else { + LOGGER.debug("releaseChannelLock called for partition key {} with no matching reservation; ignoring " + + "(likely a duplicate release)", partitionKey); } } - protected Semaphore getFreeConnectionsForHost(Object partitionKey) { - return maxConnectionsPerHost > 0 ? - freeChannelsPerHost.computeIfAbsent(partitionKey, pk -> new TrackedSemaphore(maxConnectionsPerHost)) : - InfiniteSemaphore.INSTANCE; - } - final Semaphore reserveFreeConnectionsForHost(Object partitionKey) { if (maxConnectionsPerHost <= 0) { return InfiniteSemaphore.INSTANCE; @@ -117,6 +118,8 @@ final void releaseHostReservation(Object partitionKey, Semaphore expected) { } freeChannelsPerHost.computeIfPresent(partitionKey, (pk, current) -> { if (current != expected) { + LOGGER.debug("releaseHostReservation for partition key {} found a different semaphore instance " + + "than expected; leaving it untouched (likely a duplicate release racing a prune)", pk); return current; } TrackedSemaphore semaphore = (TrackedSemaphore) current; diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2DrainPermitTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2DrainPermitTest.java index f9dc9d467..53fd0cc5a 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2DrainPermitTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2DrainPermitTest.java @@ -26,6 +26,7 @@ import org.junit.jupiter.api.Test; import java.net.InetAddress; +import java.util.concurrent.Semaphore; import static org.asynchttpclient.Dsl.config; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; @@ -86,9 +87,13 @@ private EmbeddedChannel registerRrConnectionHoldingPermit(ConnectionSemaphore se return channel; } - /** Available per-host permits for {@code baseKey} under a {@link PerHostConnectionSemaphore}. */ + /** + * Available per-host permits for {@code baseKey} under a {@link PerHostConnectionSemaphore}. A pruned + * (absent) entry means nobody holds or awaits a permit for that host, so the full capacity is free. + */ private static int availablePerHost(PerHostConnectionSemaphore semaphore, Object baseKey) { - return semaphore.getFreeConnectionsForHost(baseKey).availablePermits(); + Semaphore freeConnections = semaphore.freeChannelsPerHost.get(baseKey); + return freeConnections != null ? freeConnections.availablePermits() : semaphore.maxConnectionsPerHost; } /** diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreTest.java index 2cd85a01b..e56755aac 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/SemaphoreTest.java @@ -32,7 +32,10 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.Semaphore; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicIntegerArray; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -228,6 +231,71 @@ public void perHostRetainsEntryDuringWaitingAcquire() throws Exception { } } + // ---- concurrent stress: the reference-counted map entry must never be pruned while a permit is + // held or awaited (which would split a host's permits across two semaphore instances and let the + // per-host cap be bypassed), and must always be pruned back to empty once nothing is outstanding ---- + + static final int STRESS__KEY_COUNT = 4; + static final int STRESS__MAX_PER_HOST = 2; + static final int STRESS__THREAD_COUNT = 16; + static final int STRESS__ITERATIONS_PER_THREAD = 500; + + @Test + @Timeout(unit = TimeUnit.SECONDS, value = 20) + public void perHostConcurrentStressNeverExceedsCapAndPrunesToEmpty() throws Exception { + concurrentStress(new PerHostConnectionSemaphore(STRESS__MAX_PER_HOST, 0)); + } + + @Test + @Timeout(unit = TimeUnit.SECONDS, value = 20) + public void combinedConcurrentStressNeverExceedsCapAndPrunesToEmpty() throws Exception { + // global limit wider than the per-host limit so both layers see real contention + concurrentStress(new CombinedConnectionSemaphore(STRESS__THREAD_COUNT, STRESS__MAX_PER_HOST, 0)); + } + + private static void concurrentStress(PerHostConnectionSemaphore semaphore) throws Exception { + Object[] keys = IntStream.range(0, STRESS__KEY_COUNT) + .mapToObj(i -> "stress-host-" + i) + .toArray(); + AtomicIntegerArray currentlyHeld = new AtomicIntegerArray(STRESS__KEY_COUNT); + AtomicBoolean capExceeded = new AtomicBoolean(false); + + ExecutorService executor = Executors.newFixedThreadPool(STRESS__THREAD_COUNT); + try { + List> futures = IntStream.range(0, STRESS__THREAD_COUNT) + .mapToObj(t -> executor.submit(() -> { + ThreadLocalRandom random = ThreadLocalRandom.current(); + for (int i = 0; i < STRESS__ITERATIONS_PER_THREAD; i++) { + int keyIndex = random.nextInt(STRESS__KEY_COUNT); + Object key = keys[keyIndex]; + try { + // non-blocking: cheap enough to run thousands of iterations quickly + semaphore.acquireChannelLock(key, true); + } catch (IOException e) { + continue; + } + if (currentlyHeld.incrementAndGet(keyIndex) > STRESS__MAX_PER_HOST) { + capExceeded.set(true); + } + Thread.onSpinWait(); + currentlyHeld.decrementAndGet(keyIndex); + semaphore.releaseChannelLock(key); + } + })) + .collect(Collectors.toList()); + for (Future future : futures) { + future.get(); + } + } finally { + executor.shutdownNow(); + } + + assertFalse(capExceeded.get(), + "observed more than " + STRESS__MAX_PER_HOST + " concurrently held permits for a single host"); + assertTrue(semaphore.freeChannelsPerHost.isEmpty(), + "map must be pruned back to empty once every permit has been released"); + } + // ---- non-blocking acquire (event-loop path): must fail fast, never wait for acquireTimeout ---- // Deliberately large: a blocking acquire on an exhausted semaphore would blow the 1s @Timeout below,