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 bfa79935f..56ee13197 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,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 - // 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> 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 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 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 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. * - *

{@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 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 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 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 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 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; } 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..198e99ab2 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; @@ -86,6 +87,39 @@ public void offerThenPollReturnsSameChannelThenEmpty() { pool.destroy(); } + @Test + public void duplicateOfferIsHarmlessNoOp() throws Exception { + DefaultChannelPool pool = noReaperPool(); + Channel channel = new EmbeddedChannel(); + + assertTrue(pool.offer(channel, KEY)); + long generation = idleState(channel).snapshot(); + + 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(); @@ -245,6 +279,53 @@ public void channelReofferedAfterExpiryIsNotReaped() throws Exception { pool.destroy(); } + @Test + public void cleanerDoesNotCloseChannelReofferedAfterExpiryCheck() throws Exception { + CapturingTimer timer = new CapturingTimer(); + DefaultChannelPool pool = new DefaultChannelPool(Duration.ofMillis(1), Duration.ZERO, + PoolLeaseStrategy.FIFO, timer, Duration.ofMillis(1)); + BlockingActiveChannel channel = new BlockingActiveChannel(); + Channel untouchedChannel = new EmbeddedChannel(); + + 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); + 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"); + 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(); + } + + @Test + public void idleGenerationChangesWhenTimestampIsReused() { + DefaultChannelPool.IdleState idleState = new DefaultChannelPool.IdleState(); + assertTrue(idleState.reset(1)); + long firstGeneration = idleState.snapshot(); + + assertTrue(idleState.takeOwnership()); + assertTrue(idleState.reset(1)); + + assertFalse(idleState.tryTakeOwnership(firstGeneration)); + assertFalse(idleState.isOwned()); + } + // ---- reap pass unlinks many channels in a single tick (O(n) iterator-remove) ---- @Test @@ -415,11 +496,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]; @@ -473,6 +558,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: @@ -490,14 +577,52 @@ 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() { + // EmbeddedChannel calls this from its constructor before subclass fields are initialized. + 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() {