From 7ac8bc6479c57ef298a51a49b129918b7a50fd80 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Sat, 25 Jul 2026 15:32:10 +0200 Subject: [PATCH 1/3] Fix idle pool generation race Claim the exact idle generation evaluated by the cleaner so a lease and re-offer cannot turn a stale expiry decision into a close of the fresh entry. Keep generation and ownership in one CAS-able state to preserve the per-channel allocation model and avoid hiding a newer entry from poll. Fixes #2284 Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../netty/channel/DefaultChannelPool.java | 81 ++++++++------- .../netty/channel/DefaultChannelPoolTest.java | 99 ++++++++++++++++++- 2 files changed, 140 insertions(+), 40 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java b/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java index bfa79935f..48d3ae505 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java +++ b/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,8 +52,8 @@ public final class DefaultChannelPool implements ChannelPool { private static final AttributeKey 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 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> partitions = new ConcurrentHashMap<>(); private final AtomicBoolean isClosed = new AtomicBoolean(false); private final Timer nettyTimer; @@ -132,9 +132,8 @@ private boolean offer0(Channel channel, Object partitionKey, long now) { 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() publishes a + // new idle generation before offerFirst publishes the channel. Attribute idleStateAttribute = channel.attr(IDLE_STATE_ATTRIBUTE_KEY); IdleState idleState = idleStateAttribute.get(); if (idleState == null) { @@ -293,52 +292,62 @@ private static final class ChannelCreation { * {@link #IDLE_STATE_ATTRIBUTE_KEY} attribute, then reused across every pool checkout so no holder * is allocated per offer. * - *

{@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. + *

The low bit of {@code state} is the ownership flag; the remaining bits are incremented on + * every offer. A successful {@code poll()} lease and a {@code removeAll()} tombstone both claim 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. */ static final class IdleState { - private static final AtomicIntegerFieldUpdater OWNED_UPDATER = - AtomicIntegerFieldUpdater.newUpdater(IdleState.class, "owned"); + private static final long OWNED_MASK = 1L; + private static final AtomicLongFieldUpdater STATE_UPDATER = + AtomicLongFieldUpdater.newUpdater(IdleState.class, "state"); private volatile long start; - @SuppressWarnings("unused") - private volatile int owned; + private volatile long state; long start() { return start; } + long snapshot() { + return state; + } + boolean isOwned() { - return owned != 0; + return isOwned(state); } - /** Atomically claim this entry; returns true only for the caller that transitions 0 -> 1. */ + static boolean isOwned(long stateSnapshot) { + return (stateSnapshot & OWNED_MASK) != 0; + } + + /** Atomically claim the current generation. */ boolean takeOwnership() { - return OWNED_UPDATER.getAndSet(this, 1) == 0; + return takeOwnership(state); } - /** Undo a claim taken via {@link #takeOwnership()} (used only on the cleaner re-offer race). */ - void releaseOwnership() { - owned = 0; + /** Atomically claim {@code stateSnapshot}, provided it is still the current generation. */ + boolean takeOwnership(long stateSnapshot) { + return !isOwned(stateSnapshot) + && STATE_UPDATER.compareAndSet(this, stateSnapshot, stateSnapshot | OWNED_MASK); } - /** Stamp the idle start and mark the channel leasable again. Called on every offer. */ + /** Stamp a new idle generation and mark the channel leasable again. Called on every offer. */ void reset(long now) { + long nextGeneration = (state & ~OWNED_MASK) + 2; + // Keep the new generation unavailable until its timestamp is published. + state = nextGeneration | OWNED_MASK; start = now; - owned = 0; + state = nextGeneration; } } 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,7 +419,8 @@ private int reapPartition(ConcurrentLinkedDeque partition, long now) { continue; } - if (idleState.isOwned()) { + long stateSnapshot = idleState.snapshot(); + if (IdleState.isOwned(stateSnapshot)) { // 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 @@ -419,22 +429,17 @@ private int reapPartition(ConcurrentLinkedDeque partition, long now) { 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.takeOwnership(stateSnapshot)) { continue; } diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java index 9795b4155..be3c65e25 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java @@ -31,6 +31,7 @@ import java.util.Collections; import java.util.Map; import java.util.Set; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.ConcurrentLinkedQueue; @@ -245,6 +246,63 @@ public void channelReofferedAfterExpiryIsNotReaped() throws Exception { pool.destroy(); } + @Test + public void cleanerDoesNotCloseChannelReofferedAfterExpiryCheck() throws Exception { + CapturingTimer timer = new CapturingTimer(); + DefaultChannelPool pool = idlePool(timer, Duration.ofMillis(1)); + BlockingActiveChannel channel = new BlockingActiveChannel(); + + pool.offer(channel, KEY); + idleState(channel).reset(0); + channel.blockNextActiveCheck(); + + CompletableFuture cleanerRun = CompletableFuture.runAsync(timer::fire); + try { + assertTrue(channel.awaitActiveCheck(), "cleaner must reach the active-channel check"); + assertSame(channel, pool.poll(KEY)); + assertTrue(pool.offer(channel, KEY)); + } finally { + channel.resumeActiveCheck(); + } + cleanerRun.get(5, TimeUnit.SECONDS); + + assertTrue(channel.isActive(), "cleaner must not close a freshly re-offered channel"); + assertSame(channel, pool.poll(KEY), "freshly re-offered channel must remain leasable"); + + pool.destroy(); + } + + @Test + public void staleCleanerClaimDoesNotHideReofferedChannelFromPoll() throws Exception { + DefaultChannelPool pool = noReaperPool(); + Channel channel = new EmbeddedChannel(); + + assertTrue(pool.offer(channel, KEY)); + DefaultChannelPool.IdleState idleState = idleState(channel); + long staleSnapshot = idleState.snapshot(); + + assertSame(channel, pool.poll(KEY)); + assertTrue(pool.offer(channel, KEY)); + + assertFalse(idleState.takeOwnership(staleSnapshot), "cleaner must not claim a newer idle generation"); + assertSame(channel, pool.poll(KEY), "a failed stale claim must not hide the fresh generation"); + + pool.destroy(); + } + + @Test + public void idleGenerationChangesWhenTimestampIsReused() { + DefaultChannelPool.IdleState idleState = new DefaultChannelPool.IdleState(); + idleState.reset(1); + long firstGeneration = idleState.snapshot(); + + assertTrue(idleState.takeOwnership()); + idleState.reset(1); + + assertFalse(idleState.takeOwnership(firstGeneration)); + assertFalse(idleState.isOwned()); + } + // ---- reap pass unlinks many channels in a single tick (O(n) iterator-remove) ---- @Test @@ -490,14 +548,51 @@ public void concurrentOfferPollRemoveAllIsConsistent() throws Exception { // ---- helpers ---- - private static Object idleState(Channel channel) throws Exception { + private static DefaultChannelPool.IdleState idleState(Channel channel) throws Exception { Field keyField = DefaultChannelPool.class.getDeclaredField("IDLE_STATE_ATTRIBUTE_KEY"); keyField.setAccessible(true); @SuppressWarnings("unchecked") - io.netty.util.AttributeKey key = (io.netty.util.AttributeKey) keyField.get(null); + io.netty.util.AttributeKey key = + (io.netty.util.AttributeKey) keyField.get(null); return channel.attr(key).get(); } + private static final class BlockingActiveChannel extends EmbeddedChannel { + + private final AtomicBoolean blockNextActiveCheck = new AtomicBoolean(); + private final CountDownLatch activeCheck = new CountDownLatch(1); + private final CountDownLatch resumeActiveCheck = new CountDownLatch(1); + + @Override + public boolean isActive() { + AtomicBoolean block = blockNextActiveCheck; + if (block != null && block.compareAndSet(true, false)) { + activeCheck.countDown(); + try { + if (!resumeActiveCheck.await(5, TimeUnit.SECONDS)) { + throw new AssertionError("active-channel check was not resumed"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError("active-channel check was interrupted", e); + } + } + return super.isActive(); + } + + void blockNextActiveCheck() { + blockNextActiveCheck.set(true); + } + + boolean awaitActiveCheck() throws InterruptedException { + return activeCheck.await(5, TimeUnit.SECONDS); + } + + void resumeActiveCheck() { + resumeActiveCheck.countDown(); + } + } + private static Channel channelWithRemoteAddress(String host) { return new EmbeddedChannel() { From f1cc0e05b6538437cf9d75e358ea4de8b0998707 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Sat, 25 Jul 2026 20:29:19 +0200 Subject: [PATCH 2/3] Harden idle generation transfer Make pool offers transfer an owned idle generation with CAS so duplicate or concurrent offers cannot reset a leasable generation with a non-atomic update. Clarify generation encoding and snapshot claim naming, and exercise the cleaner retention and close paths end to end. Refs #2284 Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../netty/channel/DefaultChannelPool.java | 49 ++++++++++------- .../netty/channel/DefaultChannelPoolTest.java | 54 ++++++++++--------- 2 files changed, 59 insertions(+), 44 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java b/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java index 48d3ae505..750ce7e8b 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java @@ -132,15 +132,18 @@ private boolean offer0(Channel channel, Object partitionKey, long now) { if (partition == null) { partition = partitions.computeIfAbsent(partitionKey, pk -> new ConcurrentLinkedDeque<>()); } - // Reuse the channel's IdleState instead of allocating a holder per offer; reset() publishes a - // new idle generation before offerFirst publishes the channel. + // 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 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)) { + return false; } - idleState.reset(now); return partition.offerFirst(channel); } @@ -292,11 +295,15 @@ private static final class ChannelCreation { * {@link #IDLE_STATE_ATTRIBUTE_KEY} attribute, then reused across every pool checkout so no holder * is allocated per offer. * - *

The low bit of {@code state} is the ownership flag; the remaining bits are incremented on - * every offer. A successful {@code poll()} lease and a {@code removeAll()} tombstone both claim 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. + *

The low bit of {@code state} is the ownership flag; the remaining bits advance by two on every + * offer because bit zero is reserved. A successful {@code poll()} lease and a {@code removeAll()} + * tombstone both claim 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. + * + *

A generation can be reset only while owned. New state starts owned, and {@code poll()} claims + * a generation before returning it, so {@code offer()} transfers ownership instead of performing a + * non-atomic update from a leasable generation. */ static final class IdleState { @@ -305,7 +312,7 @@ static final class IdleState { AtomicLongFieldUpdater.newUpdater(IdleState.class, "state"); private volatile long start; - private volatile long state; + private volatile long state = OWNED_MASK; long start() { return start; @@ -325,22 +332,24 @@ static boolean isOwned(long stateSnapshot) { /** Atomically claim the current generation. */ boolean takeOwnership() { - return takeOwnership(state); + return tryTakeOwnership(state); } /** Atomically claim {@code stateSnapshot}, provided it is still the current generation. */ - boolean takeOwnership(long stateSnapshot) { + boolean tryTakeOwnership(long stateSnapshot) { return !isOwned(stateSnapshot) && STATE_UPDATER.compareAndSet(this, stateSnapshot, stateSnapshot | OWNED_MASK); } - /** Stamp a new idle generation and mark the channel leasable again. Called on every offer. */ - void reset(long now) { - long nextGeneration = (state & ~OWNED_MASK) + 2; - // Keep the new generation unavailable until its timestamp is published. - state = nextGeneration | OWNED_MASK; + /** Transfer an owned generation to a new leasable generation. Called on every offer. */ + synchronized boolean reset(long now) { + long stateSnapshot = state; + if (!isOwned(stateSnapshot)) { + return false; + } + long nextGeneration = (stateSnapshot & ~OWNED_MASK) + 2; start = now; - state = nextGeneration; + return STATE_UPDATER.compareAndSet(this, stateSnapshot, nextGeneration); } } @@ -439,7 +448,7 @@ private int reapPartition(ConcurrentLinkedDeque partition, long now) { // 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.takeOwnership(stateSnapshot)) { + if (!idleState.tryTakeOwnership(stateSnapshot)) { continue; } diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java index be3c65e25..5d41d7c17 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java @@ -87,6 +87,22 @@ public void offerThenPollReturnsSameChannelThenEmpty() { pool.destroy(); } + @Test + public void duplicateOfferDoesNotResetLeasableGeneration() throws Exception { + DefaultChannelPool pool = noReaperPool(); + Channel channel = new EmbeddedChannel(); + + assertTrue(pool.offer(channel, KEY)); + long generation = idleState(channel).snapshot(); + + assertFalse(pool.offer(channel, KEY), "an already leasable channel must not be offered again"); + assertEquals(generation, idleState(channel).snapshot()); + assertEquals(1, partitionSize(pool, KEY)); + assertSame(channel, pool.poll(KEY)); + + pool.destroy(); + } + @Test public void reofferingReusesTheSameIdleStateInstance() throws Exception { DefaultChannelPool pool = noReaperPool(); @@ -267,39 +283,22 @@ public void cleanerDoesNotCloseChannelReofferedAfterExpiryCheck() throws Excepti cleanerRun.get(5, TimeUnit.SECONDS); assertTrue(channel.isActive(), "cleaner must not close a freshly re-offered channel"); + assertEquals(1, partitionSize(pool, KEY), "cleaner must not unlink the fresh generation"); assertSame(channel, pool.poll(KEY), "freshly re-offered channel must remain leasable"); pool.destroy(); } - @Test - public void staleCleanerClaimDoesNotHideReofferedChannelFromPoll() throws Exception { - DefaultChannelPool pool = noReaperPool(); - Channel channel = new EmbeddedChannel(); - - assertTrue(pool.offer(channel, KEY)); - DefaultChannelPool.IdleState idleState = idleState(channel); - long staleSnapshot = idleState.snapshot(); - - assertSame(channel, pool.poll(KEY)); - assertTrue(pool.offer(channel, KEY)); - - assertFalse(idleState.takeOwnership(staleSnapshot), "cleaner must not claim a newer idle generation"); - assertSame(channel, pool.poll(KEY), "a failed stale claim must not hide the fresh generation"); - - pool.destroy(); - } - @Test public void idleGenerationChangesWhenTimestampIsReused() { DefaultChannelPool.IdleState idleState = new DefaultChannelPool.IdleState(); - idleState.reset(1); + assertTrue(idleState.reset(1)); long firstGeneration = idleState.snapshot(); assertTrue(idleState.takeOwnership()); - idleState.reset(1); + assertTrue(idleState.reset(1)); - assertFalse(idleState.takeOwnership(firstGeneration)); + assertFalse(idleState.tryTakeOwnership(firstGeneration)); assertFalse(idleState.isOwned()); } @@ -473,11 +472,15 @@ public void idleCountPerHostCountsOnlyLeasableChannels() { @Test public void concurrentOfferPollRemoveAllIsConsistent() throws Exception { - // Real timer so the cleaner reaps tombstones concurrently with offer/poll/removeAll. - // TTL only (idle disabled) so the cleaner never closes our shared EmbeddedChannels cross-thread. + // Real timer so the cleaner claims expired entries and reaps tombstones concurrently with + // offer/poll/removeAll. HashedWheelTimer timer = new HashedWheelTimer(10, TimeUnit.MILLISECONDS); - DefaultChannelPool pool = new DefaultChannelPool(Duration.ZERO, Duration.ofHours(1), + DefaultChannelPool pool = new DefaultChannelPool(Duration.ofMillis(20), Duration.ofHours(1), PoolLeaseStrategy.LIFO, timer, Duration.ofMillis(10)); + CountDownLatch cleanerClosedChannel = new CountDownLatch(1); + Channel expiringChannel = new EmbeddedChannel(); + expiringChannel.closeFuture().addListener(future -> cleanerClosedChannel.countDown()); + assertTrue(pool.offer(expiringChannel, "expiring-partition")); final int channelCount = 16; Channel[] channels = new Channel[channelCount]; @@ -531,6 +534,8 @@ public void concurrentOfferPollRemoveAllIsConsistent() throws Exception { if (failure.get() != null) { fail("worker threw: " + failure.get(), failure.get()); } + assertTrue(cleanerClosedChannel.await(5, TimeUnit.SECONDS), + "soak must exercise the cleaner ownership and close path"); assertTrue(leasedInactive.isEmpty(), "poll must never lease an inactive channel"); // Drain leases, then let the cleaner run a couple of ticks and confirm no tombstone leak: @@ -565,6 +570,7 @@ private static final class BlockingActiveChannel extends EmbeddedChannel { @Override public boolean isActive() { + // EmbeddedChannel calls this from its constructor before subclass fields are initialized. AtomicBoolean block = blockNextActiveCheck; if (block != null && block.compareAndSet(true, false)) { activeCheck.countDown(); From 94fc757c02e0ae0ef87f67de28f6afae21045472 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Sun, 26 Jul 2026 10:33:21 +0200 Subject: [PATCH 3/3] Harden concurrent idle resets Prevent duplicate or delayed offers from transferring a newer idle generation. A reset-in-progress state now binds the timestamp update to the exact owned snapshot without serializing reset calls. Treat duplicate offers as accepted no-ops, defer partition creation until a real insertion, and exercise the cleaner with a neighboring live entry. Refs #2284 Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../asynchttpclient/channel/ChannelPool.java | 2 +- .../netty/channel/DefaultChannelPool.java | 74 +++++++++++-------- .../netty/channel/DefaultChannelPoolTest.java | 36 +++++++-- 3 files changed, 74 insertions(+), 38 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/channel/ChannelPool.java b/client/src/main/java/org/asynchttpclient/channel/ChannelPool.java index 4f2bc3b9b..5d0cb2ccc 100755 --- a/client/src/main/java/org/asynchttpclient/channel/ChannelPool.java +++ b/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); diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java b/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java index 750ce7e8b..56ee13197 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java @@ -52,10 +52,11 @@ public final class DefaultChannelPool implements ChannelPool { private static final AttributeKey IDLE_STATE_ATTRIBUTE_KEY = AttributeKey.valueOf("channelIdleState"); // The partition deques hold the bare Channel; per-checkout idle state (start timestamp + the - // generation/ownership CAS state) lives on the channel's IDLE_STATE_ATTRIBUTE_KEY attribute, + // 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> 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,19 +120,10 @@ 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 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() transfers an // owned generation to a new leasable generation before offerFirst publishes the channel. Attribute idleStateAttribute = channel.attr(IDLE_STATE_ATTRIBUTE_KEY); @@ -142,9 +134,18 @@ private boolean offer0(Channel channel, Object partitionKey, long now) { idleState = existingIdleState == null ? newIdleState : existingIdleState; } if (!idleState.reset(now)) { - return false; + if (LOGGER.isDebugEnabled() && duplicateOfferLogged.compareAndSet(false, true)) { + LOGGER.debug("Ignoring duplicate pool offer for channel {}", channel); + } + return true; } - return partition.offerFirst(channel); + ConcurrentLinkedDeque partition = + partitions.computeIfAbsent(partitionKey, pk -> new ConcurrentLinkedDeque<>()); + boolean offered = partition.offerFirst(channel); + if (connectionTtlEnabled && offered) { + registerChannelCreation(channel, partitionKey, now); + } + return offered; } private static void registerChannelCreation(Channel channel, Object partitionKey, long now) { @@ -295,19 +296,21 @@ private static final class ChannelCreation { * {@link #IDLE_STATE_ATTRIBUTE_KEY} attribute, then reused across every pool checkout so no holder * is allocated per offer. * - *

The low bit of {@code state} is the ownership flag; the remaining bits advance by two on every - * offer because bit zero is reserved. A successful {@code poll()} lease and a {@code removeAll()} - * tombstone both claim 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. + *

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. * - *

A generation can be reset only while owned. New state starts owned, and {@code poll()} claims - * a generation before returning it, so {@code offer()} transfers ownership instead of performing a - * non-atomic update from a leasable generation. + *

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 long OWNED_MASK = 1L; + private static final long RESETTING_MASK = 2L; + private static final long GENERATION_INCREMENT = 4L; private static final AtomicLongFieldUpdater STATE_UPDATER = AtomicLongFieldUpdater.newUpdater(IdleState.class, "state"); @@ -330,6 +333,10 @@ static boolean isOwned(long stateSnapshot) { return (stateSnapshot & OWNED_MASK) != 0; } + private static boolean isResetting(long stateSnapshot) { + return (stateSnapshot & RESETTING_MASK) != 0; + } + /** Atomically claim the current generation. */ boolean takeOwnership() { return tryTakeOwnership(state); @@ -341,15 +348,20 @@ boolean tryTakeOwnership(long stateSnapshot) { && STATE_UPDATER.compareAndSet(this, stateSnapshot, stateSnapshot | OWNED_MASK); } - /** Transfer an owned generation to a new leasable generation. Called on every offer. */ - synchronized boolean reset(long now) { - long stateSnapshot = state; - if (!isOwned(stateSnapshot)) { + /** Attempt to transfer the current owned generation to a new leasable generation. */ + boolean reset(long now) { + return tryReset(state, 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; } - long nextGeneration = (stateSnapshot & ~OWNED_MASK) + 2; start = now; - return STATE_UPDATER.compareAndSet(this, stateSnapshot, nextGeneration); + state = (stateSnapshot & ~(OWNED_MASK | RESETTING_MASK)) + GENERATION_INCREMENT; + return true; } } @@ -430,10 +442,10 @@ private int reapPartition(ConcurrentLinkedDeque partition, long now) { long stateSnapshot = idleState.snapshot(); if (IdleState.isOwned(stateSnapshot)) { - // 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. + // 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; } diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java index 5d41d7c17..198e99ab2 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java @@ -88,21 +88,38 @@ public void offerThenPollReturnsSameChannelThenEmpty() { } @Test - public void duplicateOfferDoesNotResetLeasableGeneration() throws Exception { + public void duplicateOfferIsHarmlessNoOp() throws Exception { DefaultChannelPool pool = noReaperPool(); Channel channel = new EmbeddedChannel(); assertTrue(pool.offer(channel, KEY)); long generation = idleState(channel).snapshot(); - assertFalse(pool.offer(channel, KEY), "an already leasable channel must not be offered again"); + assertTrue(pool.offer(channel, "other-partition")); assertEquals(generation, idleState(channel).snapshot()); assertEquals(1, partitionSize(pool, KEY)); + assertEquals(0, partitionSize(pool, "other-partition")); + assertTrue(channel.isActive()); assertSame(channel, pool.poll(KEY)); pool.destroy(); } + @Test + public void staleResetCannotTransferNewerOwnedGeneration() { + DefaultChannelPool.IdleState idleState = new DefaultChannelPool.IdleState(); + assertTrue(idleState.reset(1000)); + assertTrue(idleState.takeOwnership()); + long staleSnapshot = idleState.snapshot(); + + assertTrue(idleState.tryReset(staleSnapshot, 5000)); + assertTrue(idleState.takeOwnership()); + + assertFalse(idleState.tryReset(staleSnapshot, 1000)); + assertEquals(5000, idleState.start()); + assertTrue(idleState.isOwned()); + } + @Test public void reofferingReusesTheSameIdleStateInstance() throws Exception { DefaultChannelPool pool = noReaperPool(); @@ -265,11 +282,17 @@ public void channelReofferedAfterExpiryIsNotReaped() throws Exception { @Test public void cleanerDoesNotCloseChannelReofferedAfterExpiryCheck() throws Exception { CapturingTimer timer = new CapturingTimer(); - DefaultChannelPool pool = idlePool(timer, Duration.ofMillis(1)); + DefaultChannelPool pool = new DefaultChannelPool(Duration.ofMillis(1), Duration.ZERO, + PoolLeaseStrategy.FIFO, timer, Duration.ofMillis(1)); BlockingActiveChannel channel = new BlockingActiveChannel(); + Channel untouchedChannel = new EmbeddedChannel(); - pool.offer(channel, KEY); - idleState(channel).reset(0); + assertTrue(pool.offer(channel, KEY)); + assertTrue(idleState(channel).takeOwnership()); + assertTrue(idleState(channel).reset(0)); + assertTrue(pool.offer(untouchedChannel, KEY)); + assertTrue(idleState(untouchedChannel).takeOwnership()); + assertTrue(idleState(untouchedChannel).reset(Long.MAX_VALUE)); channel.blockNextActiveCheck(); CompletableFuture cleanerRun = CompletableFuture.runAsync(timer::fire); @@ -283,7 +306,8 @@ public void cleanerDoesNotCloseChannelReofferedAfterExpiryCheck() throws Excepti cleanerRun.get(5, TimeUnit.SECONDS); assertTrue(channel.isActive(), "cleaner must not close a freshly re-offered channel"); - assertEquals(1, partitionSize(pool, KEY), "cleaner must not unlink the fresh generation"); + assertEquals(2, partitionSize(pool, KEY), "cleaner must not unlink the fresh generation"); + assertSame(untouchedChannel, pool.poll(KEY)); assertSame(channel, pool.poll(KEY), "freshly re-offered channel must remain leasable"); pool.destroy();