Skip to content

Claim exact idle generation before closing pooled channels #2298

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hyperxpro merged 1 commit into from
Aug 2, 2026
+623 −83
Merged

Claim exact idle generation before closing pooled channels #2298

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion client/src/main/java/org/asynchttpclient/channel/ChannelPool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand Down
204 changes: 149 additions & 55 deletions client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -52,10 +52,11 @@ public final class DefaultChannelPool implements ChannelPool {
private static final AttributeKey<IdleState> IDLE_STATE_ATTRIBUTE_KEY = AttributeKey.valueOf("channelIdleState");

// The partition deques hold the bare Channel; per-checkout idle state (start timestamp + the
// owned/tombstone CAS flag) lives on the channel's IDLE_STATE_ATTRIBUTE_KEY attribute, which is
// 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<Object, ConcurrentLinkedDeque<Channel>> 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;
Expand Down Expand Up @@ -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<Channel> partition = partitions.get(partitionKey);
if (partition == null) {
partition = partitions.computeIfAbsent(partitionKey, pk -> new ConcurrentLinkedDeque<>());
}
// Reuse the channel's IdleState instead of allocating a holder per offer; reset() stamps the
// idle start and clears the owned flag (must happen-before offerFirst publishes the channel,
// so any thread that observes it in the deque also observes owned == 0).
// 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<IdleState> 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<Channel> partition = partitions.get(partitionKey);
if (partition == null) {
partition = partitions.computeIfAbsent(partitionKey, pk -> new ConcurrentLinkedDeque<>());
}
idleState.reset(now);
return partition.offerFirst(channel);
}

Expand Down Expand Up @@ -221,7 +247,18 @@ private void flushPartition(Object partitionKey, ConcurrentLinkedDeque<Channel>
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);
}
}
}
}
Expand All @@ -241,8 +278,9 @@ public Map<String, Long> getIdleChannelCountPerHost() {
Map<String, Long> idleChannelsPerHost = new HashMap<>();
for (ConcurrentLinkedDeque<Channel> 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) {
Expand Down Expand Up @@ -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.
*
* <p>{@code owned} is a single CAS flag with two roles, both meaning "this idle entry is claimed,
* do not lease it": a successful {@code poll()} lease, or a {@code removeAll()} tombstone. The pool
* upholds the invariant that a channel sitting in a partition deque has {@code owned == 0} unless it
* was tombstoned, because {@link #reset(long)} clears the flag before {@code offerFirst} publishes
* the channel and {@code poll()} unlinks a channel from the deque before claiming it. {@code start}
* doubles as a generation token: it changes on every offer, letting the cleaner detect a channel
* that was leased and re-offered between its expiry check and its claim.
* <p>{@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}.
*
* <p>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<IdleState> 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<IdleState> 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.
*
* <p>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
Expand Down Expand Up @@ -410,32 +506,30 @@ private int reapPartition(ConcurrentLinkedDeque<Channel> partition, long now) {
continue;
}

if (idleState.isOwned()) {
// In-deque + owned ==> a removeAll(Channel) tombstone, or a node a concurrent poll()
// has already leased and unlinked. Either way: unlink, never close — the owner of the
// claim is responsible for closing it. Unlinking an already-unlinked node through the
// iterator is a harmless no-op.
long stateSnapshot = idleState.snapshot();
if (IdleState.isOwned(stateSnapshot)) {
// In-deque + owned ==> a removeAll(Channel) tombstone, a node a concurrent poll() 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={}",
Expand Down
Loading
Loading