Guard per-host semaphore pruning against unbalanced releases

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;

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;
     }
 
     /**

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<Future<?>> 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,