diff --git a/client/src/main/java/org/asynchttpclient/channel/ChannelPool.java b/client/src/main/java/org/asynchttpclient/channel/ChannelPool.java index 4f2bc3b9b..4c333ec8e 100755 --- a/client/src/main/java/org/asynchttpclient/channel/ChannelPool.java +++ b/client/src/main/java/org/asynchttpclient/channel/ChannelPool.java @@ -28,7 +28,9 @@ 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 pool accepted the channel, false if it rejected it, in which case the caller + * owns the channel and must close it. An implementation may accept a channel without pooling it, + * for instance when the very same channel is already pooled. */ 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..79344f31d 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 + // ownership/generation word) 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). private final ConcurrentHashMap> partitions = new ConcurrentHashMap<>(); private final AtomicBoolean isClosed = new AtomicBoolean(false); + private final AtomicBoolean rejectedOfferLogged = new AtomicBoolean(false); private final Timer nettyTimer; private final long connectionTtl; private final boolean connectionTtlEnabled; @@ -119,29 +120,54 @@ public boolean offer(Channel channel, Object partitionKey) { return false; } - boolean offered = offer0(channel, partitionKey, now); - if (connectionTtlEnabled && offered) { + // "accepted", not "offered": offer0 also returns true for a channel that was already pooled, so + // this flag means the pool took responsibility for the channel, not that this call linked it. + // registerChannelCreation is safe under that weaker meaning because it only sets when absent. + boolean accepted = offer0(channel, partitionKey, now); + if (connectionTtlEnabled && accepted) { registerChannelCreation(channel, partitionKey, now); } - return offered; + return accepted; } 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). + // idle start and publishes the next, leasable generation (which happens-before offerFirst + // publishes the channel, so any thread that observes it in the deque also observes it unowned). + // setIfAbsent, not set: two concurrent first offers of the same channel would otherwise each + // install their own IdleState, the second clobbering the first, after which both would transfer + // their own generation and link the channel twice. Both must end up on one shared state so that + // exactly one of them wins the transfer below. Attribute idleStateAttribute = channel.attr(IDLE_STATE_ATTRIBUTE_KEY); IdleState idleState = idleStateAttribute.get(); if (idleState == null) { - idleState = new IdleState(); - idleStateAttribute.set(idleState); + IdleState created = new IdleState(); + IdleState raced = idleStateAttribute.setIfAbsent(created); + idleState = raced == null ? created : raced; + } + + if (!idleState.reset(now)) { + // No generation to transfer: the channel is already pooled (so this is a duplicate offer), or + // a concurrent offer of the same channel won the reservation. Adding it would double-link it + // in a deque, so treat this as an accepted no-op and leave the existing generation and + // placement alone. Returning false would be worse than useless: ChannelManager closes a + // rejected channel, so a duplicate offer would kill a live pooled keep-alive. + if (LOGGER.isDebugEnabled() && rejectedOfferLogged.compareAndSet(false, true)) { + LOGGER.debug("Ignoring offer of channel {} for partition {}: it is already pooled, or a " + + "concurrent offer is pooling it. Further occurrences are not logged.", channel, partitionKey); + } + return true; + } + + // Resolve the partition only once the generation transfer succeeded, so a rejected offer cannot + // leave an empty deque behind for the cleaner to walk on every tick. Keep the plain get() fast + // path: computeIfAbsent only returns lock-free when the key is its bin's head node, and otherwise + // locks the bin (or joins a resize), which get() never does. + ConcurrentLinkedDeque partition = partitions.get(partitionKey); + if (partition == null) { + partition = partitions.computeIfAbsent(partitionKey, pk -> new ConcurrentLinkedDeque<>()); } - idleState.reset(now); return partition.offerFirst(channel); } @@ -221,7 +247,18 @@ private void flushPartition(Object partitionKey, ConcurrentLinkedDeque if (partition != null) { partitions.remove(partitionKey); for (Channel channel : partition) { - close(channel); + // Claim before closing, the same rule the idle cleaner follows. Removing the partition + // from the map does not stop a concurrent poll(): a caller that read the deque reference + // first can still lease out of it, and this iterator can still see a node that poll has + // unlinked. Closing unconditionally would then kill a connection a request is already + // using. Losing the claim means somebody else owns the channel -- a lessee, a removeAll + // tombstone, or the cleaner -- and closing it is their responsibility, not ours. + // Winning the claim also leaves the channel owned, so the pool never holds a channel that + // is closed yet still reads as leasable. + IdleState idleState = channel.attr(IDLE_STATE_ATTRIBUTE_KEY).get(); + if (idleState == null || idleState.takeOwnership()) { + close(channel); + } } } } @@ -241,8 +278,9 @@ public Map getIdleChannelCountPerHost() { Map idleChannelsPerHost = new HashMap<>(); for (ConcurrentLinkedDeque partition : partitions.values()) { for (Channel channel : partition) { - // Skip channels that have been claimed (removeAll tombstone, or a node a concurrent - // poll already leased) but not yet unlinked, so the count reflects leasable channels. + // Skip channels that have been claimed (removeAll tombstone, a node a concurrent poll + // already leased, or an offer mid-transfer) but not yet unlinked, so the count reflects + // leasable channels. if (isLeasable(channel)) { SocketAddress remoteAddress = channel.remoteAddress(); if (remoteAddress.getClass() == InetSocketAddress.class) { @@ -293,52 +331,110 @@ 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. + *

{@code state} packs an ownership flag ({@link #OWNED}, bit 0), a transfer-in-progress flag + * ({@link #RESETTING}, bit 1) and a generation counter (all remaining bits). Because the two flags sit + * in the low bits, the counter advances by {@link #GENERATION_INCREMENT} = 4 on every offer rather + * than by one. A generation identifies one idle checkout and {@code start} is the millisecond + * timestamp at which that checkout became idle. The counter, not the timestamp, is the identity: + * {@code unpreciseMillisTime()} is millisecond-grained, so two consecutive checkouts can carry the + * same {@code start}. + * + *

Owned means "claimed, do not lease". It has five producers: the initial state (the thread about + * to make the first offer owns the channel, so that offer is a legal transfer), a successful + * {@code poll()} lease, a {@code removeAll(Channel)} tombstone, the idle cleaner's pre-close claim, + * and a transfer in progress ({@link #RESETTING} is only ever set on top of {@link #OWNED}, so a + * mid-reset entry reads as owned and is therefore neither leased, counted as idle, nor closed). The + * pool upholds the invariant that a channel sitting in a partition deque is unowned unless it was + * tombstoned, is being leased by a concurrent {@code poll()}, or is being closed by the cleaner: + * {@link #reset(long)} publishes the next, unowned generation before {@code offerFirst} publishes the + * channel, and {@code poll()} unlinks a channel from the deque before claiming it. */ static final class IdleState { - private static final AtomicIntegerFieldUpdater OWNED_UPDATER = - AtomicIntegerFieldUpdater.newUpdater(IdleState.class, "owned"); + private static final long OWNED = 1L; + private static final long RESETTING = 1L << 1; + private static final long GENERATION_INCREMENT = 1L << 2; + + 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; long start() { return start; } + /** The current state word, to be passed back to {@link #tryTakeOwnership(long)}. */ + long snapshot() { + return state; + } + + static boolean isOwned(long stateSnapshot) { + return (stateSnapshot & OWNED) != 0; + } + boolean isOwned() { - return owned != 0; + return isOwned(state); } - /** Atomically claim this entry; returns true only for the caller that transitions 0 -> 1. */ + /** + * Claim whichever generation is current, for a caller that already holds the channel (a + * {@code poll()} that unlinked it) or wants to tombstone it ({@code removeAll}). Returns true + * only for the caller that transitions an unowned generation to owned; a lost CAS is reported as + * failure rather than retried, since the only transition out of an unowned generation is to + * owned, so a lost CAS means somebody else claimed it. + * + *

Reading {@code state} at call time can observe a generation published by a re-offer, which + * has linked a fresh node. That is benign rather than a double lease: a {@code poll()} claim has + * already unlinked its own node, so the only claimer that can be followed by a re-offer is + * {@code removeAll}, and {@link #reset(long)} refuses a second transfer of the same generation. + * The worst case is a channel leased while a stale node lingers, which the next {@code poll()} or + * cleaner tick unlinks. + */ boolean takeOwnership() { - return OWNED_UPDATER.getAndSet(this, 1) == 0; + return tryTakeOwnership(state); } - /** Undo a claim taken via {@link #takeOwnership()} (used only on the cleaner re-offer race). */ - void releaseOwnership() { - owned = 0; + /** + * Claim exactly the generation {@code stateSnapshot} was taken from, in one atomic step. Fails, + * without ever owning the channel even transiently, if the channel was claimed or leased and + * re-offered since the snapshot was taken. + */ + boolean tryTakeOwnership(long stateSnapshot) { + return !isOwned(stateSnapshot) && STATE_UPDATER.compareAndSet(this, stateSnapshot, stateSnapshot | OWNED); } - /** Stamp the idle start and mark the channel leasable again. Called on every offer. */ - void reset(long now) { + /** + * Stamp a fresh idle start and publish the next generation, transferring ownership to the pool. + * Called on every offer. The guard is on the generation, not on the caller's identity: an unowned + * generation is already pooled, so it is refused rather than stamped with a second idle start and + * linked into a deque twice. A generation owned by somebody else is still transferable, which is + * what lets a lessee hand the channel back. Reserving the transfer with {@link #RESETTING} keeps the entry unleasable while + * {@code start} is written and makes a second transfer of the same generation fail rather than + * rewind the idle clock. Publishing {@code state} last is sufficient for a reader that observes + * the new generation to also observe the new {@code start}, since volatile accesses are totally + * ordered in the synchronization order (JLS 17.4.4, 17.4.7). + * + * @return false if there was no owned generation to transfer, or a concurrent transfer of it won + * the reservation, in which case nothing changed + */ + boolean reset(long now) { + long stateSnapshot = state; + if (!isOwned(stateSnapshot) || (stateSnapshot & RESETTING) != 0 + || !STATE_UPDATER.compareAndSet(this, stateSnapshot, stateSnapshot | RESETTING)) { + return false; + } start = now; - owned = 0; + state = (stateSnapshot & ~(OWNED | RESETTING)) + 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,32 +506,30 @@ 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() has + // already leased and unlinked, or an offer whose transfer is in progress. Either way: + // unlink, never close: whoever holds the claim is responsible for the channel. + // Unlinking an already-unlinked node through the iterator is a harmless no-op. it.remove(); continue; } - boolean isIdleTimeoutExpired = isIdleTimeoutExpired(idleState, now); + boolean isIdleTimeoutExpired = isIdleTimeoutExpired(idleState.start(), 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(); - continue; + // Claim exactly the generation this verdict was computed from, in a single atomic step: + // the channel is never closed on a verdict that a lease + re-offer has invalidated, and a + // generation this cleaner did not evaluate is never even transiently owned (which would + // starve a concurrent poll() into dropping a live channel). The counter only ever + // advances, so a successful claim also proves the start read above belongs to it. + if (!idleState.tryTakeOwnership(stateSnapshot)) { + continue; // leased, tombstoned or re-offered meanwhile; that owner now handles it } LOGGER.debug("Closing Idle Channel {} isIdleTimeoutExpired={} isRemotelyClosed={} isTtlExpired={}", 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..5c14df91e 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/DefaultChannelPoolTest.java @@ -30,17 +30,21 @@ import java.time.Duration; import java.util.Collections; import java.util.Map; +import java.util.Queue; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -48,12 +52,17 @@ /** * White-box unit tests for {@link DefaultChannelPool} covering the bare-channel storage + reused - * {@code IdleState} attribute (plan 009) and the O(1) tombstone {@code removeAll} + tombstone-aware - * idle cleaner (plan 013). The cleaner is driven deterministically through a capturing {@link Timer}. + * {@code IdleState} attribute (plan 009), the O(1) tombstone {@code removeAll} + tombstone-aware idle + * cleaner (plan 013) and the generation-scoped cleaner claim (issue #2284). The cleaner is driven + * deterministically through a capturing {@link Timer}. */ public class DefaultChannelPoolTest { private static final Object KEY = "partition-key"; + private static final Object OTHER_KEY = "other-partition-key"; + // Deliberately not "pool-..." : the stall gate keys off this prefix, and Netty's timer worker is + // named by Executors.defaultThreadFactory() as "pool-N-thread-M". Keep the namespaces disjoint. + private static final String RACE_WORKER_PREFIX = "race-worker-"; private static DefaultChannelPool noReaperPool() { // No TTL, no idle timeout => no cleaner scheduled; removeAll is a no-op (unchanged behavior). @@ -245,6 +254,135 @@ public void channelReofferedAfterExpiryIsNotReaped() throws Exception { pool.destroy(); } + // ---- issue #2284: the cleaner must claim exactly the generation it evaluated ---- + + @Test + public void cleanerSparesChannelReofferedWhileItWasMidDecision() throws Exception { + // Issue #2284, end to end. The cleaner is parked inside its remote-close check, i.e. after it + // decided the channel is idle-expired and before it claims it; the channel is then leased and + // re-offered. A FIFO pool is what makes the lease target the stale channel: offer() is offerFirst, + // so the untouched, fresh channel necessarily sits ahead of the stale one in the cleaner's scan, + // and a FIFO lease takes from the back. + final long maxIdle = 1000; + CapturingTimer timer = new CapturingTimer(); + DefaultChannelPool pool = new DefaultChannelPool(Duration.ofMillis(maxIdle), Duration.ZERO, + PoolLeaseStrategy.FIFO, timer, Duration.ofMillis(1)); + + BlockingActiveChannel stale = new BlockingActiveChannel(); + assertTrue(pool.offer(stale, KEY)); + Thread.sleep(maxIdle + 100); // this checkout is now reapable + + Channel fresh = new EmbeddedChannel(); + assertTrue(pool.offer(fresh, KEY)); // deque is [fresh, stale], scanned in that order + + stale.armBlock(); + final AtomicReference cleanerFailure = new AtomicReference<>(); + Thread cleaner = new Thread(() -> { + try { + timer.fire(); + } catch (Throwable t) { + cleanerFailure.set(t); + } + }, "idle-cleaner"); + cleaner.start(); + + assertTrue(stale.awaitBlocked(30, TimeUnit.SECONDS), "cleaner must reach the blocked liveness check"); + assertSame(stale, pool.poll(KEY), "a FIFO lease takes the stale channel from the back"); + assertTrue(pool.offer(stale, KEY), "the re-offer stamps a fresh idle generation"); + stale.unblock(); + + cleaner.join(TimeUnit.SECONDS.toMillis(30)); + assertFalse(cleaner.isAlive(), "the cleaner pass must finish"); + assertNull(cleanerFailure.get(), () -> "cleaner threw: " + cleanerFailure.get()); + + assertTrue(stale.isActive(), "a channel re-offered mid-decision must not be closed on a stale verdict"); + assertTrue(fresh.isActive(), "the channel already scanned must not be touched"); + assertEquals(2, partitionSize(pool, KEY), "both entries must survive the tick"); + assertSame(fresh, pool.poll(KEY), "the untouched channel is still at the back for a FIFO lease"); + assertSame(stale, pool.poll(KEY), "the re-offered channel must still be leasable"); + assertNull(pool.poll(KEY)); + + pool.destroy(); + } + + @Test + public void duplicateOfferIsAcceptedAsANoOp() throws Exception { + CapturingTimer timer = new CapturingTimer(); + DefaultChannelPool pool = idlePool(timer, Duration.ofHours(1)); + Channel channel = new EmbeddedChannel(); + + assertTrue(pool.offer(channel, KEY)); + assertTrue(pool.offer(channel, OTHER_KEY), + "a duplicate offer must not be reported as rejected: the caller closes a rejected channel"); + + assertTrue(channel.isActive(), "a duplicate offer must not cost the caller a live keep-alive"); + assertEquals(1, partitionSize(pool, KEY), "the channel must stay pooled exactly once"); + assertFalse(hasPartition(pool, OTHER_KEY), "a rejected offer must not create an empty partition deque"); + assertSame(channel, pool.poll(KEY), "the channel must still be leasable from its original partition"); + assertNull(pool.poll(KEY)); + + pool.destroy(); + } + + @Test + public void leasedChannelCanBeOfferedToAnotherPartition() throws Exception { + // The generation transfer must not be so strict that it blocks a legitimate re-offer: a lease + // hands ownership to the caller, so the next offer may pool the channel under any key. + CapturingTimer timer = new CapturingTimer(); + DefaultChannelPool pool = idlePool(timer, Duration.ofHours(1)); + Channel channel = new EmbeddedChannel(); + + assertTrue(pool.offer(channel, KEY)); + assertSame(channel, pool.poll(KEY)); + assertTrue(pool.offer(channel, OTHER_KEY)); + + assertEquals(0, partitionSize(pool, KEY), "the channel must not be left in its old partition"); + assertEquals(1, partitionSize(pool, OTHER_KEY)); + assertSame(channel, pool.poll(OTHER_KEY)); + + pool.destroy(); + } + + @Test + public void duplicateOfferDoesNotRefreshTheIdleClock() throws Exception { + // A rejected generation transfer must leave the idle start alone. If it stamped a fresh one, an + // offer from a thread that does not own the channel would keep an expired connection alive. + final long maxIdle = 1000; + CapturingTimer timer = new CapturingTimer(); + DefaultChannelPool pool = idlePool(timer, Duration.ofMillis(maxIdle)); + Channel channel = new EmbeddedChannel(); + + assertTrue(pool.offer(channel, KEY)); + Thread.sleep(maxIdle + 100); + assertTrue(pool.offer(channel, KEY), "a duplicate offer is accepted as a no-op"); + + timer.fire(); + + assertFalse(channel.isActive(), "a duplicate offer must not reset the idle clock"); + assertEquals(0, partitionSize(pool, KEY)); + + pool.destroy(); + } + + @Test + public void generationsSharingATimestampAreStillDistinct() { + // unpreciseMillisTime() is millisecond-grained, so two checkouts can carry the same idle start. + // The generation counter, not the timestamp, is what identifies a checkout. + DefaultChannelPool.IdleState idleState = new DefaultChannelPool.IdleState(); + + assertTrue(idleState.reset(1000L), "a brand new state is owned by the thread making the first offer"); + long first = idleState.snapshot(); + assertFalse(DefaultChannelPool.IdleState.isOwned(first), "an offered channel is leasable"); + + assertTrue(idleState.takeOwnership(), "poll leases it"); + assertTrue(idleState.reset(1000L), "re-offered within the same millisecond"); + long second = idleState.snapshot(); + + assertEquals(1000L, idleState.start()); + assertFalse(DefaultChannelPool.IdleState.isOwned(second)); + assertNotEquals(first, second, "checkouts sharing a timestamp must still be distinct generations"); + } + // ---- reap pass unlinks many channels in a single tick (O(n) iterator-remove) ---- @Test @@ -275,6 +413,72 @@ public void cleanerReapsManyIdleExpiredChannelsInOneTick() throws Exception { pool.destroy(); } + // ---- flushPartitions claims before closing, like the cleaner ---- + + @Test + public void flushPartitionsClosesPooledChannels() { + DefaultChannelPool pool = noReaperPool(); + Channel flushed = new EmbeddedChannel(); + Channel other = new EmbeddedChannel(); + + assertTrue(pool.offer(flushed, KEY)); + assertTrue(pool.offer(other, OTHER_KEY)); + + pool.flushPartitions(KEY::equals); + + assertFalse(flushed.isActive(), "a pooled channel in a flushed partition must be closed"); + assertNull(pool.poll(KEY), "the flushed partition must be gone"); + assertTrue(other.isActive(), "a partition the predicate rejects must be untouched"); + assertSame(other, pool.poll(OTHER_KEY)); + + pool.destroy(); + } + + @Test + public void flushPartitionsSkipsAChannelClaimedBySomebodyElse() throws Exception { + // The state flush has to respect is in-deque AND owned. The motivating case is a concurrent + // poll(): removing the partition from the map does not stop a caller that already holds the deque + // reference, and flush's weakly consistent iterator can still see the node that poll unlinked, so + // closing unconditionally would kill a connection a request is already using. That interleaving + // cannot be forced deterministically, but a removeAll tombstone reaches the same state - claimed, + // still linked - and pins the rule: whoever holds the claim owns the close. + CapturingTimer timer = new CapturingTimer(); + DefaultChannelPool pool = ttlPool(timer); + Channel pooled = new EmbeddedChannel(); + Channel claimed = new EmbeddedChannel(); + + assertTrue(pool.offer(pooled, KEY)); + assertTrue(pool.offer(claimed, KEY)); + assertTrue(pool.removeAll(claimed), "tombstone leaves it claimed but still linked"); + assertEquals(2, partitionSize(pool, KEY)); + + pool.flushPartitions(KEY::equals); + + assertTrue(claimed.isActive(), "flush must not close a channel somebody else has claimed"); + assertFalse(pooled.isActive(), "the unclaimed pooled channel must still be closed"); + + pool.destroy(); + } + + @Test + public void flushedChannelIsNotReportedAsPooledByALaterOffer() throws Exception { + // Flush leaves the channels it closed owned, so the pool never holds one that is closed yet still + // reads as leasable. Were they left unowned, a later offer would take the accepted-no-op path and + // return true without pooling anything. + DefaultChannelPool pool = noReaperPool(); + Channel channel = new EmbeddedChannel(); + + assertTrue(pool.offer(channel, KEY)); + pool.flushPartitions(KEY::equals); + assertFalse(channel.isActive()); + + assertTrue(pool.offer(channel, KEY), "the closed channel is owned, so the transfer succeeds"); + assertEquals(1, partitionSize(pool, KEY), "a true return must mean the channel really was pooled"); + assertNull(pool.poll(KEY), "poll still refuses to lease a dead channel"); + + pool.destroy(); + } + @Test public void cleanerReapsExpiredButKeepsHealthyInSameTick() throws Exception { // A single reap pass must drop the expired channels AND keep the fresh ones leasable: the @@ -414,13 +618,29 @@ public void idleCountPerHostCountsOnlyLeasableChannels() { // ---- concurrency: no leaked tombstones, never leases a claimed channel ---- @Test + @org.junit.jupiter.api.Timeout(120) // io.netty.util.Timeout owns the simple name here 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 reaps concurrently with offer/poll/removeAll. Idle expiry is enabled + // on top of the TTL so the cleaner exercises its claim-and-close path under contention instead of + // only the tombstone-unlink branch. That is safe for the shared channels the workers lease: to + // close one the cleaner has to win the claim, and a poll that wins the claim locks the cleaner + // out for that whole checkout, so poll still never hands out a channel that is being closed. + // The fixed channel set does decay over the run: a channel that no poll reaches within maxIdle is + // closed, and re-offering it only puts a corpse back in the deque, so the population of leasable + // channels shrinks. That is why leaseCount is asserted - it stops this test from silently + // degenerating into "every poll returned null", which would make the assertions below vacuous. + // Sustained leasing pressure against a live cleaner is covered by + // concurrentPollAndOfferNeverStrandALiveChannel, which replaces the channels it drains. + final long maxIdle = 500; HashedWheelTimer timer = new HashedWheelTimer(10, TimeUnit.MILLISECONDS); - DefaultChannelPool pool = new DefaultChannelPool(Duration.ZERO, Duration.ofHours(1), + DefaultChannelPool pool = new DefaultChannelPool(Duration.ofMillis(maxIdle), Duration.ofHours(1), PoolLeaseStrategy.LIFO, timer, Duration.ofMillis(10)); + // Its own partition, never touched by the workers, so it ages out untouched: its close proves the + // cleaner really claimed and closed an idle-expired channel while the workers were hammering it. + EmbeddedChannel expiring = new EmbeddedChannel(); + assertTrue(pool.offer(expiring, OTHER_KEY)); + final int channelCount = 16; Channel[] channels = new Channel[channelCount]; for (int i = 0; i < channelCount; i++) { @@ -431,6 +651,7 @@ public void concurrentOfferPollRemoveAllIsConsistent() throws Exception { final AtomicBoolean stop = new AtomicBoolean(false); final AtomicReference failure = new AtomicReference<>(); final ConcurrentLinkedQueue leasedInactive = new ConcurrentLinkedQueue<>(); + final AtomicLong leaseCount = new AtomicLong(); final CountDownLatch done = new CountDownLatch(threads); for (int t = 0; t < threads; t++) { @@ -448,8 +669,11 @@ public void concurrentOfferPollRemoveAllIsConsistent() throws Exception { break; case 1: Channel leased = pool.poll(KEY); - if (leased != null && !leased.isActive()) { - leasedInactive.add(leased); // poll must never hand out a dead channel + if (leased != null) { + leaseCount.incrementAndGet(); + if (!leased.isActive()) { + leasedInactive.add(leased); // poll must never hand out a dead channel + } } break; default: @@ -466,26 +690,170 @@ public void concurrentOfferPollRemoveAllIsConsistent() throws Exception { worker.start(); } - Thread.sleep(1500); - stop.set(true); - assertTrue(done.await(10, TimeUnit.SECONDS), "workers must finish"); + try { + Thread.sleep(1500); + stop.set(true); + assertTrue(done.await(10, TimeUnit.SECONDS), "workers must finish"); - if (failure.get() != null) { - fail("worker threw: " + failure.get(), failure.get()); + if (failure.get() != null) { + fail("worker threw: " + failure.get(), failure.get()); + } + assertTrue(leasedInactive.isEmpty(), "poll must never lease an inactive channel"); + // A healthy run leases thousands; this floor is orders of magnitude below that, but high + // enough to fail if leasing collapses early instead of merely decaying. + assertTrue(leaseCount.get() >= channelCount * 10L, + "the run must have leased real channels, otherwise the assertions above are vacuous; was " + leaseCount); + assertTrue(expiring.closeFuture().await(30, TimeUnit.SECONDS), + "the cleaner must have claimed and closed the idle-expired channel"); + + // Drain leases, then let the cleaner run a couple of ticks and confirm no tombstone leak: + // every partition deque must collapse to at most the number of distinct channels. + while (pool.poll(KEY) != null) { + // drain + } + Thread.sleep(60); // a few cleaner ticks + int size = partitionSize(pool, KEY); + assertTrue(size <= channelCount, "tombstones must not accumulate unbounded, was " + size); + assertEquals(0, partitionSize(pool, OTHER_KEY), "the closed channel must be unlinked"); + } finally { + // Any assertion above throwing would otherwise leave the workers running and a 10ms-period + // timer reaping a live pool for the rest of the fork, spraying cleaner DEBUG lines into every + // test class that follows. + stop.set(true); + pool.destroy(); + timer.stop(); } - 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: - // every partition deque must collapse to at most the number of distinct channels. - while (pool.poll(KEY) != null) { - // drain - } - Thread.sleep(60); // a few cleaner ticks - int size = partitionSize(pool, KEY); - assertTrue(size <= channelCount, "tombstones must not accumulate unbounded, was " + size); + @Test + @org.junit.jupiter.api.Timeout(120) // io.netty.util.Timeout owns the simple name here + public void concurrentPollAndOfferNeverStrandALiveChannel() throws Exception { + // Issue #2284 race B, detected by accounting at quiescence rather than by pausing inside the + // window. If the cleaner ever owns a generation it never evaluated, a concurrent poll that has + // already unlinked the node fails its own claim and drops the channel, and the cleaner then + // releases the claim it should never have taken. The channel is left alive, unowned and in no + // partition at all: a permanent loss that survives to quiescence, where it is countable. + // + // Workers only poll-then-offer (never removeAll) and always complete the cycle before exiting, so + // the invariant is exact: at quiescence every channel ever created is either still pooled or was + // closed by the cleaner. Anything else was stranded. Workers mint a replacement whenever the pool + // runs dry, which keeps real leasing pressure on the cleaner for the whole run. + // YieldingChannel stalls only the cleaner thread inside its liveness check, widening the + // decide -> claim window; that is a test-only subclass, exactly like BlockingActiveChannel, and + // needs no hook in production code. + // + // This is a soak, not a deterministic test: its failure mode is a false negative, never a false + // positive. With the generation-scoped claim the count is 0 by construction, because + // tryTakeOwnership cannot own a generation it did not evaluate. Against the "hoist the snapshot, + // claim, compare, release" variant it fails in the large majority of runs (independently measured + // 9 of 12 and 7 of 11 on a loaded 2-core box); the fix itself passed every run measured, both + // plain and under the jacoco agent that ./mvnw verify attaches. Treat the ratio as a range, not a + // gate: it moves with the harness and the machine. + // + // The stall widens the cleaner's decide -> claim window, but it is NOT a floor. A shorter stall + // measured at least as well (100us and a bare yield both scored higher than 1ms), because the + // cleaner is single-threaded: every stall blocks the whole pass, so race opportunities are capped + // at wall-time / stall-duration. A re-offer landing in the same millisecond as the verdict leaves + // start unchanged, which makes that variant close the channel rather than strand it - invisible to + // this accounting - but that is a fraction of occurrences, not a gate on detection. + final int seedChannels = 64; + final int threads = 8; + final long soakMillis = 4000; + HashedWheelTimer timer = new HashedWheelTimer(1, TimeUnit.MILLISECONDS); + // 1ms idle timeout and a 1ms cleaner period: every pooled channel reads as expired on every tick, + // so the cleaner is permanently in its decide -> claim path, which is where the race lives. + DefaultChannelPool pool = new DefaultChannelPool(Duration.ofMillis(1), Duration.ofHours(1), + PoolLeaseStrategy.LIFO, timer, Duration.ofMillis(1)); + + final Queue created = new ConcurrentLinkedQueue<>(); + final AtomicBoolean stop = new AtomicBoolean(false); + final AtomicReference failure = new AtomicReference<>(); + final CountDownLatch done = new CountDownLatch(threads); + boolean timerStopped = false; + + // Nothing may throw between arming the stall and entering the try, or it would stay armed for the + // rest of the class. Only YieldingChannel reads it, and this is the only test that creates one. + YieldingChannel.stalling(true); + try { + for (int i = 0; i < seedChannels; i++) { + Channel seed = new YieldingChannel(); + created.add(seed); + assertTrue(pool.offer(seed, KEY)); + } - pool.destroy(); - timer.stop(); + for (int t = 0; t < threads; t++) { + Thread worker = new Thread(() -> { + try { + while (!stop.get()) { + Channel channel = pool.poll(KEY); + if (channel == null) { + channel = new YieldingChannel(); + created.add(channel); + } + // Always complete the cycle before re-checking stop: a channel left leased at + // quiescence would read as a loss and fail this test for the wrong reason. + pool.offer(channel, KEY); + } + } catch (Throwable th) { + failure.compareAndSet(null, th); + } finally { + done.countDown(); + } + }, RACE_WORKER_PREFIX + t); + worker.start(); + } + + Thread.sleep(soakMillis); + stop.set(true); + assertTrue(done.await(30, TimeUnit.SECONDS), "workers must finish"); + + if (failure.get() != null) { + fail("worker threw: " + failure.get(), failure.get()); + } + + // Freeze the cleaner before counting anything, otherwise a channel closed between the two + // counting loops below lands in neither bucket and reads as a loss. HashedWheelTimer.stop() + // interrupts and joins its worker, so once it returns no pass is in flight and the channel + // states are frozen; that join is the barrier, so no sleep is needed here. A pass caught + // mid-flight cannot reschedule itself, and the resulting IllegalStateException is caught and + // logged by Netty's HashedTimeout.expire(). + YieldingChannel.stalling(false); // the accounting below runs isActive() on this thread too + timer.stop(); + timerStopped = true; + + // Three buckets, so that the only thing left over is the race-B signature itself: a channel + // that is alive, owned by nobody, and in no partition. Counting a claimed-but-not-yet-closed + // channel as a loss would be a false positive - the cleaner closes inside a catch-all, so a + // claim whose close did not take effect leaves the channel alive and owned, which is not a + // strand. A strand cannot hide in that bucket: the variant this test exists to catch releases + // its claim, which is precisely what makes the channel unreachable. + long closedByCleaner = 0; + long claimedNotClosed = 0; + for (Channel channel : created) { + if (!channel.isActive()) { + closedByCleaner++; + } else if (((DefaultChannelPool.IdleState) idleState(channel)).isOwned()) { + claimedNotClosed++; + } + } + long stillPooled = 0; + while (pool.poll(KEY) != null) { + stillPooled++; + } + + long lost = created.size() - closedByCleaner - claimedNotClosed - stillPooled; + assertEquals(0, lost, "a live channel is in no partition and owned by nobody: the cleaner claimed " + + "a generation it never evaluated and a concurrent poll dropped the node (created=" + + created.size() + " closedByCleaner=" + closedByCleaner + " claimedNotClosed=" + + claimedNotClosed + " stillPooled=" + stillPooled + ")"); + } finally { + YieldingChannel.stalling(false); + stop.set(true); + pool.destroy(); + if (!timerStopped) { + timer.stop(); + } + } } // ---- helpers ---- @@ -511,15 +879,91 @@ protected SocketAddress remoteAddress0() { } @SuppressWarnings("unchecked") - private static int partitionSize(DefaultChannelPool pool, Object key) throws Exception { + private static ConcurrentHashMap> partitions(DefaultChannelPool pool) + throws Exception { Field partitionsField = DefaultChannelPool.class.getDeclaredField("partitions"); partitionsField.setAccessible(true); - ConcurrentHashMap> partitions = - (ConcurrentHashMap>) partitionsField.get(pool); - ConcurrentLinkedDeque partition = partitions.get(key); + return (ConcurrentHashMap>) partitionsField.get(pool); + } + + private static int partitionSize(DefaultChannelPool pool, Object key) throws Exception { + ConcurrentLinkedDeque partition = partitions(pool).get(key); return partition == null ? 0 : partition.size(); } + private static boolean hasPartition(DefaultChannelPool pool, Object key) throws Exception { + return partitions(pool).containsKey(key); + } + + /** + * An {@link EmbeddedChannel} whose liveness check occasionally stalls the idle cleaner, widening the + * window between its expiry decision and its claim. Only the cleaner is stalled: worker threads are + * recognised by their name, and the whole thing is disarmed once a run is over so the accounting pass + * on the test thread runs at full speed. + */ + private static final class YieldingChannel extends EmbeddedChannel { + + // Static, so the constructor's own isActive() call cannot read it before it is initialised - the + // trap BlockingActiveChannel has to null-guard around. + private static final AtomicBoolean STALLING = new AtomicBoolean(false); + + static void stalling(boolean enabled) { + STALLING.set(enabled); + } + + @Override + public boolean isActive() { + if (STALLING.get() && !Thread.currentThread().getName().startsWith(RACE_WORKER_PREFIX) + && ThreadLocalRandom.current().nextInt(8) == 0) { + try { + Thread.sleep(1); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + return super.isActive(); + } + } + + /** + * An {@link EmbeddedChannel} whose liveness check parks once, letting a test stop the idle cleaner + * exactly between its expiry decision and its claim. + */ + private static final class BlockingActiveChannel extends EmbeddedChannel { + + private final AtomicBoolean armed = new AtomicBoolean(false); + private final CountDownLatch blocked = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + @Override + public boolean isActive() { + // EmbeddedChannel's constructor registers the channel, which calls this override before this + // class's own final fields are assigned: the null check is load-bearing, not always-true. + // Without it that call NPEs inside a DefaultPromise listener, which swallows the failure. + if (armed != null && armed.compareAndSet(true, false)) { + blocked.countDown(); + try { + release.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + return super.isActive(); + } + + void armBlock() { + armed.set(true); + } + + boolean awaitBlocked(long timeout, TimeUnit unit) throws InterruptedException { + return blocked.await(timeout, unit); + } + + void unblock() { + release.countDown(); + } + } + /** * A {@link Timer} that captures the last-scheduled {@link TimerTask} (the pool's idle cleaner) so a * test can fire it synchronously instead of waiting on wall-clock time.