Fix idle pool cleaner generation race

client/src/main/java/org/asynchttpclient/channel/ChannelPool.java

@@ -28,7 +28,7 @@ public interface ChannelPool {
      *
      * @param channel      an I/O channel
      * @param partitionKey a key used to retrieve the cached channel
-     * @return true if added.
+     * @return true if the offer was accepted; false if the channel was rejected
      */
     boolean offer(Channel channel, Object partitionKey);
 

client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java

@@ -37,7 +37,7 @@
 import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.function.Predicate;
 
 import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime;
@@ -52,10 +52,11 @@ public final class DefaultChannelPool implements ChannelPool {
     private static final AttributeKey<IdleState> IDLE_STATE_ATTRIBUTE_KEY = AttributeKey.valueOf("channelIdleState");
 
     // The partition deques hold the bare Channel; per-checkout idle state (start timestamp + the
-    // owned/tombstone CAS flag) lives on the channel's IDLE_STATE_ATTRIBUTE_KEY attribute, which is
-    // allocated once per physical connection and reused across every pool cycle (no per-offer holder).
+    // generation/ownership/reset CAS state) lives on the channel's IDLE_STATE_ATTRIBUTE_KEY attribute,
+    // which is allocated once per physical connection and reused across every pool cycle.
     private final ConcurrentHashMap<Object, ConcurrentLinkedDeque<Channel>> partitions = new ConcurrentHashMap<>();
     private final AtomicBoolean isClosed = new AtomicBoolean(false);
+    private final AtomicBoolean duplicateOfferLogged = new AtomicBoolean(false);
     private final Timer nettyTimer;
     private final long connectionTtl;
     private final boolean connectionTtlEnabled;
@@ -119,30 +120,32 @@ public boolean offer(Channel channel, Object partitionKey) {
             return false;
         }
 
-        boolean offered = offer0(channel, partitionKey, now);
-        if (connectionTtlEnabled && offered) {
-            registerChannelCreation(channel, partitionKey, now);
-        }
-
-        return offered;
+        return offer0(channel, partitionKey, now);
     }
 
     private boolean offer0(Channel channel, Object partitionKey, long now) {
-        ConcurrentLinkedDeque<Channel> partition = partitions.get(partitionKey);
-        if (partition == null) {
-            partition = partitions.computeIfAbsent(partitionKey, pk -> new ConcurrentLinkedDeque<>());
-        }
-        // Reuse the channel's IdleState instead of allocating a holder per offer; reset() stamps the
-        // idle start and clears the owned flag (must happen-before offerFirst publishes the channel,
-        // so any thread that observes it in the deque also observes owned == 0).
+        // Reuse the channel's IdleState instead of allocating a holder per offer; reset() transfers an
+        // owned generation to a new leasable generation before offerFirst publishes the channel.
         Attribute<IdleState> idleStateAttribute = channel.attr(IDLE_STATE_ATTRIBUTE_KEY);
         IdleState idleState = idleStateAttribute.get();
         if (idleState == null) {
-            idleState = new IdleState();
-            idleStateAttribute.set(idleState);
+            IdleState newIdleState = new IdleState();
+            IdleState existingIdleState = idleStateAttribute.setIfAbsent(newIdleState);
+            idleState = existingIdleState == null ? newIdleState : existingIdleState;
+        }
+        if (!idleState.reset(now)) {
+            if (LOGGER.isDebugEnabled() && duplicateOfferLogged.compareAndSet(false, true)) {
+                LOGGER.debug("Ignoring duplicate pool offer for channel {}", channel);
+            }
+            return true;
+        }
+        ConcurrentLinkedDeque<Channel> partition =
+                partitions.computeIfAbsent(partitionKey, pk -> new ConcurrentLinkedDeque<>());
+        boolean offered = partition.offerFirst(channel);
+        if (connectionTtlEnabled && offered) {
+            registerChannelCreation(channel, partitionKey, now);
         }
-        idleState.reset(now);
-        return partition.offerFirst(channel);
+        return offered;
     }
 
     private static void registerChannelCreation(Channel channel, Object partitionKey, long now) {
@@ -293,52 +296,79 @@ private static final class ChannelCreation {
      * {@link #IDLE_STATE_ATTRIBUTE_KEY} attribute, then reused across every pool checkout so no holder
      * is allocated per offer.
      *
-     * <p>{@code owned} is a single CAS flag with two roles, both meaning "this idle entry is claimed,
-     * do not lease it": a successful {@code poll()} lease, or a {@code removeAll()} tombstone. The pool
-     * upholds the invariant that a channel sitting in a partition deque has {@code owned == 0} unless it
-     * was tombstoned, because {@link #reset(long)} clears the flag before {@code offerFirst} publishes
-     * the channel and {@code poll()} unlinks a channel from the deque before claiming it. {@code start}
-     * doubles as a generation token: it changes on every offer, letting the cleaner detect a channel
-     * that was leased and re-offered between its expiry check and its claim.
+     * <p>The low bit of {@code state} is the ownership flag and the next bit marks a reset in progress;
+     * the remaining bits advance by four on every offer. A successful {@code poll()} lease, a
+     * {@code removeAll()} tombstone, and the cleaner's pre-close claim all own the current generation.
+     * The cleaner can therefore claim only the exact generation whose expiry it evaluated, without
+     * temporarily claiming a newer entry while checking whether {@code start} changed.
+     *
+     * <p>A generation can be reset only while owned and not already being reset. New state starts owned,
+     * and reset first claims the generation's reset bit before publishing the timestamp and next leasable
+     * generation. Concurrent or delayed resets therefore cannot transfer a generation they did not claim.
      */
     static final class IdleState {
 
-        private static final AtomicIntegerFieldUpdater<IdleState> OWNED_UPDATER =
-                AtomicIntegerFieldUpdater.newUpdater(IdleState.class, "owned");
+        private static final long OWNED_MASK = 1L;
+        private static final long RESETTING_MASK = 2L;
+        private static final long GENERATION_INCREMENT = 4L;
+        private static final AtomicLongFieldUpdater<IdleState> STATE_UPDATER =
+                AtomicLongFieldUpdater.newUpdater(IdleState.class, "state");
 
         private volatile long start;
-        @SuppressWarnings("unused")
-        private volatile int owned;
+        private volatile long state = OWNED_MASK;
 
         long start() {
             return start;
         }
 
+        long snapshot() {
+            return state;
+        }
+
         boolean isOwned() {
-            return owned != 0;
+            return isOwned(state);
+        }
+
+        static boolean isOwned(long stateSnapshot) {
+            return (stateSnapshot & OWNED_MASK) != 0;
         }
 
-        /** Atomically claim this entry; returns true only for the caller that transitions 0 -> 1. */
+        private static boolean isResetting(long stateSnapshot) {
+            return (stateSnapshot & RESETTING_MASK) != 0;
+        }
+
+        /** Atomically claim the current generation. */
         boolean takeOwnership() {
-            return OWNED_UPDATER.getAndSet(this, 1) == 0;
+            return tryTakeOwnership(state);
+        }
+
+        /** Atomically claim {@code stateSnapshot}, provided it is still the current generation. */
+        boolean tryTakeOwnership(long stateSnapshot) {
+            return !isOwned(stateSnapshot)
+                    && STATE_UPDATER.compareAndSet(this, stateSnapshot, stateSnapshot | OWNED_MASK);
         }
 
-        /** Undo a claim taken via {@link #takeOwnership()} (used only on the cleaner re-offer race). */
-        void releaseOwnership() {
-            owned = 0;
+        /** Attempt to transfer the current owned generation to a new leasable generation. */
+        boolean reset(long now) {
+            return tryReset(state, now);
         }
 
-        /** Stamp the idle start and mark the channel leasable again. Called on every offer. */
-        void reset(long now) {
+        /** Transfer {@code stateSnapshot} if it is still owned, current, and not already being reset. */
+        boolean tryReset(long stateSnapshot, long now) {
+            if (!isOwned(stateSnapshot) || isResetting(stateSnapshot)
+                    || !STATE_UPDATER.compareAndSet(this, stateSnapshot, stateSnapshot | RESETTING_MASK)) {
+                return false;
+            }
             start = now;
-            owned = 0;
+            state = (stateSnapshot & ~(OWNED_MASK | RESETTING_MASK)) + GENERATION_INCREMENT;
+            return true;
         }
     }
 
     private final class IdleChannelDetector implements TimerTask {
 
-        private boolean isIdleTimeoutExpired(IdleState idleState, long now) {
-            return maxIdleTimeEnabled && now - idleState.start() >= maxIdleTime;
+        private boolean isIdleTimeoutExpired(long idleStart, long now) {
+            return maxIdleTimeEnabled && now - idleStart >= maxIdleTime;
         }
 
         @Override
@@ -410,31 +440,27 @@ private int reapPartition(ConcurrentLinkedDeque<Channel> partition, long now) {
                     continue;
                 }
 
-                if (idleState.isOwned()) {
-                    // In-deque + owned ==> a removeAll(Channel) tombstone, or a node a concurrent poll()
-                    // has already leased and unlinked. Either way: unlink, never close — the owner of the
-                    // claim is responsible for closing it. Unlinking an already-unlinked node through the
-                    // iterator is a harmless no-op.
+                long stateSnapshot = idleState.snapshot();
+                if (IdleState.isOwned(stateSnapshot)) {
+                    // In-deque + owned ==> a removeAll(Channel) tombstone, a node a concurrent poll()
+                    // already leased and unlinked, or an old node whose generation is being reset.
+                    // Either way: unlink, never close — the owner of the claim handles the channel.
+                    // Unlinking an already-unlinked node through the iterator is a harmless no-op.
                     it.remove();
                     continue;
                 }
 
-                boolean isIdleTimeoutExpired = isIdleTimeoutExpired(idleState, now);
+                long idleStart = idleState.start();
+                boolean isIdleTimeoutExpired = isIdleTimeoutExpired(idleStart, now);
                 boolean isRemotelyClosed = !Channels.isChannelActive(channel);
                 boolean isTtlExpired = isTtlExpired(channel, now);
                 if (!isIdleTimeoutExpired && !isRemotelyClosed && !isTtlExpired) {
                     continue; // healthy idle channel, leave it for poll()
                 }
 
-                long startSnapshot = idleState.start();
-                // Claim before closing so we never close a channel poll() is leasing concurrently.
-                if (!idleState.takeOwnership()) {
-                    continue; // poll() (or removeAll(Channel)) won the claim; that owner now handles the channel
-                }
-                if (idleState.start() != startSnapshot) {
-                    // The channel was leased and re-offered (fresh start) between the expiry check and
-                    // the claim, so it is leasable again — release it instead of closing it.
-                    idleState.releaseOwnership();
+                // Claim the generation evaluated above. A lease and re-offer changes the generation,
+                // so this CAS cannot claim the fresh entry or interfere with a concurrent poll of it.
+                if (!idleState.tryTakeOwnership(stateSnapshot)) {
                     continue;
                 }