From 57fc570d77623f44d10ef02ca4dbaf4e1a96aa23 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Fri, 6 Nov 2026 08:19:56 +0100 Subject: [PATCH 01/11] Add a benchmark for arming request timeouts Every attempt builds a TimeoutsHolder, arms its request timeout on the client's HashedWheelTimer, starts its read timeout and cancels both when the exchange ends. The timer's pending counter and its two MPSC queues are shared by every caller, so this cost grows with the number of threads sending requests. The benchmark measures that lifecycle with the read timeout shorter and not shorter than the request timeout, next to one bare timer entry as a floor, to judge changes to how the two timeouts are armed. Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5.5 --- .../bench/RequestTimeoutsArmBenchmark.java | 133 ++++++++++++++++++ 1 file changed, 133 insertions(+) create mode 100644 client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsArmBenchmark.java diff --git a/client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsArmBenchmark.java b/client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsArmBenchmark.java new file mode 100644 index 000000000..5c2948252 --- /dev/null +++ b/client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsArmBenchmark.java @@ -0,0 +1,133 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.bench; + +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timeout; +import io.netty.util.TimerTask; +import org.asynchttpclient.AsyncCompletionHandler; +import org.asynchttpclient.AsyncHttpClientConfig; +import org.asynchttpclient.DefaultAsyncHttpClient; +import org.asynchttpclient.DefaultAsyncHttpClientConfig; +import org.asynchttpclient.Request; +import org.asynchttpclient.RequestBuilder; +import org.asynchttpclient.Response; +import org.asynchttpclient.channel.ChannelPoolPartitioning; +import org.asynchttpclient.netty.NettyResponseFuture; +import org.asynchttpclient.netty.request.NettyRequestSender; +import org.asynchttpclient.netty.timeout.TimeoutsHolder; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; + +import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.concurrent.TimeUnit; + +/** + * What arming and cancelling an exchange's timeouts costs on the client's {@link HashedWheelTimer}: one + * {@link TimeoutsHolder} built, started, given its read timeout and cancelled, as every attempt does. + *

+ * {@code readTimeoutMs} picks which timeout comes first: at 60000 the read timeout is not shorter than what is + * left of the 60 s request timeout, so only the request timeout needs arming; at 10000 the read timeout is + * due first and has to be served as well. {@code rawTimerEntry} is one bare {@code newTimeout} plus + * {@code cancel}, the floor for a single entry. + *

+ * The timer ticks every millisecond rather than the default 100: its worker moves at most 100000 new entries + * into the wheel per tick, so at the rates measured here a 100 ms tick leaves a backlog that grows for the + * whole run and turns the result into a measure of the collector. + *

+ * Run with several thread counts, since the timer's queues and pending counter are shared by every caller: + * {@code -t 1}, {@code -t 8}, {@code -t 32}, with {@code -prof gc}. + */ +@State(Scope.Benchmark) +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.MICROSECONDS) +public class RequestTimeoutsArmBenchmark { + + private static final TimerTask NOOP = timeout -> { + }; + + @Param({"60000", "10000"}) + public int readTimeoutMs; + + private HashedWheelTimer timer; + private DefaultAsyncHttpClient client; + private AsyncHttpClientConfig config; + private NettyRequestSender requestSender; + private InetSocketAddress remoteAddress; + + @State(Scope.Thread) + public static class Exchange { + NettyResponseFuture future; + + @Setup + public void setup() { + Request request = new RequestBuilder().setUrl("http://localhost:8080/").build(); + future = new NettyResponseFuture<>(request, new AsyncCompletionHandler() { + @Override + public Object onCompleted(Response response) { + return null; + } + }, null, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null); + } + } + + @Setup + public void setup() throws ReflectiveOperationException { + timer = new HashedWheelTimer(1, TimeUnit.MILLISECONDS, 512); + timer.start(); + config = new DefaultAsyncHttpClientConfig.Builder() + .setNettyTimer(timer) + .setRequestTimeout(Duration.ofMillis(60000)) + .setReadTimeout(Duration.ofMillis(readTimeoutMs)) + .build(); + client = new DefaultAsyncHttpClient(config); + Field field = DefaultAsyncHttpClient.class.getDeclaredField("requestSender"); + field.setAccessible(true); + requestSender = (NettyRequestSender) field.get(client); + remoteAddress = InetSocketAddress.createUnresolved("localhost", 8080); + } + + @TearDown + public void tearDown() throws Exception { + client.close(); + timer.stop(); + } + + @Benchmark + public TimeoutsHolder armAndCancel(Exchange exchange) { + TimeoutsHolder holder = new TimeoutsHolder(timer, null, exchange.future, requestSender, config, remoteAddress); + holder.start(); + holder.startReadTimeout(); + holder.cancel(); + return holder; + } + + @Benchmark + public Timeout rawTimerEntry() { + Timeout timeout = timer.newTimeout(NOOP, 60000, TimeUnit.MILLISECONDS); + timeout.cancel(); + return timeout; + } +} From d9363186dc52651f6a529dc6a39fc5afb021bcd8 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Fri, 6 Nov 2026 08:20:18 +0100 Subject: [PATCH 02/11] Keep private timeout guards in field updaters Every attempt allocates a TimeoutsHolder with two AtomicBoolean guards, and every read timeout task a third for its suspension warning. Each is a 16-byte object reached through a pointer, allocated and dropped per request. Volatile ints behind static AtomicIntegerFieldUpdaters keep the same compare-and-set semantics without those objects: RequestTimeoutsArmBenchmark allocates 216 instead of 248 bytes per exchange when only the request timeout is armed, and 344 instead of 392 when the read timeout is armed as well. TimeoutTimerTask.done stays an AtomicBoolean: it is a protected field of a public class, and replacing it is the API break that closed #2263. Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5.5 --- .../netty/timeout/ReadTimeoutTimerTask.java | 9 ++++++--- .../netty/timeout/TimeoutsHolder.java | 19 ++++++++++++------- 2 files changed, 18 insertions(+), 10 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java index 1ee82da2a..01aedabee 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java @@ -23,7 +23,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime; @@ -31,8 +31,11 @@ public class ReadTimeoutTimerTask extends TimeoutTimerTask implements Runnable { private static final Logger LOGGER = LoggerFactory.getLogger(ReadTimeoutTimerTask.class); + private static final AtomicIntegerFieldUpdater INDEFINITE_SUSPENSION_WARNING_LOGGED = + AtomicIntegerFieldUpdater.newUpdater(ReadTimeoutTimerTask.class, "indefiniteSuspensionWarningLogged"); + private final long readTimeout; - private final AtomicBoolean indefiniteSuspensionWarningLogged = new AtomicBoolean(); + private volatile int indefiniteSuspensionWarningLogged; ReadTimeoutTimerTask(NettyResponseFuture nettyResponseFuture, NettyRequestSender requestSender, TimeoutsHolder timeoutsHolder, long readTimeout) { super(nettyResponseFuture, requestSender, timeoutsHolder); @@ -89,7 +92,7 @@ public void run(Timeout timeout) { void warnIfIndefinitelySuspended() { if (timeoutsHolder.isRequestTimeoutDisabled() - && indefiniteSuspensionWarningLogged.compareAndSet(false, true)) { + && INDEFINITE_SUSPENSION_WARNING_LOGGED.compareAndSet(this, 0, 1)) { LOGGER.warn("Response body reads for {} remain suspended while the request timeout is disabled; " + "the exchange retains its transport resources until it is resumed or canceled", nettyResponseFuture.getUri().toUrlWithoutUserInfo()); diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java index 8c5c269d6..55161ec3a 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -30,7 +30,7 @@ import java.net.InetSocketAddress; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.function.LongSupplier; /** @@ -42,8 +42,13 @@ public class TimeoutsHolder { private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutsHolder.class); - private final AtomicBoolean cancelled = new AtomicBoolean(); - private final AtomicBoolean started = new AtomicBoolean(); + private static final AtomicIntegerFieldUpdater CANCELLED = + AtomicIntegerFieldUpdater.newUpdater(TimeoutsHolder.class, "cancelled"); + private static final AtomicIntegerFieldUpdater STARTED = + AtomicIntegerFieldUpdater.newUpdater(TimeoutsHolder.class, "started"); + + private volatile int cancelled; + private volatile int started; private final Timer nettyTimer; private volatile @Nullable EventExecutor eventExecutor; private final NettyRequestSender requestSender; @@ -129,7 +134,7 @@ public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, N * cancel to sit in its scheduler until the full deadline, pinning the task, the future and the channel. */ public void start() { - if (!started.compareAndSet(false, true)) { + if (!STARTED.compareAndSet(this, 0, 1)) { return; } if (requestTimeoutTask != null) { @@ -202,7 +207,7 @@ public void rehomeOn(EventExecutor executor) { } eventExecutor = executor; RequestTimeoutTimerTask task = requestTimeoutTask; - if (task == null || cancelled.get() || task.isClaimed() || !task.cancelArmed()) { + if (task == null || cancelled != 0 || task.isClaimed() || !task.cancelArmed()) { return; } arm(task, remainingRequestTimeout()); @@ -240,7 +245,7 @@ void startReadTimeout(@Nullable ReadTimeoutTimerTask task) { } public void cancel() { - if (cancelled.compareAndSet(false, true)) { + if (CANCELLED.compareAndSet(this, 0, 1)) { release(requestTimeoutTask); release(readTimeoutTask); } @@ -306,7 +311,7 @@ private void arm(T task, long delay) { * would sit in the scheduler until the full deadline, waking a loop for a request that is long done. */ private void cancelIfRaced(TimeoutTimerTask task) { - if (cancelled.get()) { + if (cancelled != 0) { release(task); } } From b3afd1da94d3320d3fa8fddf6703b05f588b0256 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Fri, 6 Nov 2026 08:40:23 +0100 Subject: [PATCH 03/11] Serve the read timeout from the request entry When the read timeout is shorter than the request timeout, an attempt armed two timer entries: the request timeout when it started and the read timeout once the request was written. Almost every exchange finishes before either fires, so both were inserted and cancelled for nothing. On the shared HashedWheelTimer each entry is an increment of a client-wide counter and an offer to a client-wide queue, and each cancel an offer to a second one, all drained by its single thread; on an event loop each is a scheduled task to add and remove. Arm the request timeout's entry for when the read timeout would be due instead. Once the read timeout starts it rides on that entry, which on firing fails the exchange with a read timeout when one is due and otherwise re-arms for whichever deadline comes next: what is left of the read window, capped by the request deadline. A read timeout that has not started by the time the entry fires goes back to an entry of its own, so a slow connect or upload does not wake the scheduler every window. Without start(), a holder still arms the read timeout on its own entry as before. RequestTimeoutsArmBenchmark, read timeout 10 s and request timeout 60 s, JDK 17, three forks: threads before (ops/us) after (ops/us) bytes/op 1 1.61 +- 0.60 2.25 +- 0.86 391 -> 216 8 3.59 +- 0.93 6.89 +- 2.51 394 -> 217 32 1.07 +- 0.65 3.31 +- 0.77 395 -> 218 The host was shared with other builds, hence the error bars. The "after" column includes the previous commit's field updaters. Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5.5 --- .../timeout/RequestTimeoutTimerTask.java | 44 +++ .../netty/timeout/TimeoutsHolder.java | 102 ++++++- .../timeout/RequestTimeoutTimerTaskTest.java | 277 ++++++++++++++++++ 3 files changed, 413 insertions(+), 10 deletions(-) create mode 100644 client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java index 1b913ea7a..680e39709 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java @@ -16,6 +16,7 @@ package org.asynchttpclient.netty.timeout; import io.netty.util.Timeout; +import org.asynchttpclient.netty.NettyResponseBodyControl; import org.asynchttpclient.netty.NettyResponseFuture; import org.asynchttpclient.netty.request.NettyRequestSender; import org.asynchttpclient.util.StringBuilderPool; @@ -49,6 +50,10 @@ public void run(Timeout timeout) { return; } + if (timeoutsHolder.isRequestTimeoutArmedEarly() && checkReadTimeout()) { + return; + } + // in any case, cancel possible readTimeout sibling timeoutsHolder.cancel(); @@ -62,4 +67,43 @@ public void run(Timeout timeout) { long age = unpreciseMillisTime() - nettyResponseFuture.getStart(); expire(message, age); } + + /** + * Runs when this entry was armed ahead of the request deadline, for when the read timeout could be due, so + * that one entry serves both. Fails the exchange with a read timeout when it is due, and otherwise arms + * the entry again for whichever of the two comes next. + * + * @return {@code false} when the request deadline has passed as well, which leaves the request timeout to + * be delivered as though the entry had been armed for it + */ + private boolean checkReadTimeout() { + if (nettyResponseFuture.isDone()) { + timeoutsHolder.cancel(); + return true; + } + + long now = timeoutsHolder.millisTime(); + if (now >= timeoutsHolder.requestTimeoutMillisTime()) { + return false; + } + boolean suspended = NettyResponseBodyControl.isSuspended(nettyResponseFuture); + if (timeoutsHolder.isReadTimeoutCovered() && !suspended) { + long readTimeout = timeoutsHolder.readTimeout(); + long sinceLastTouch = now - nettyResponseFuture.getLastTouch(); + if (sinceLastTouch >= readTimeout) { + StringBuilder sb = StringBuilderPool.DEFAULT.stringBuilder().append("Read timeout to "); + appendRemoteAddress(sb); + String message = sb.append(" after ").append(readTimeout).append(" ms").toString(); + expire(message, sinceLastTouch); + timeoutsHolder.cancel(); + return true; + } + } + + // Released before re-arming: the holder may hand the read timeout an entry of its own, which is only + // armed while this task is unclaimed. + done.set(false); + timeoutsHolder.rearmRequestTimeout(this, now, suspended); + return true; + } } diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java index 55161ec3a..214a4f6b5 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -46,9 +46,21 @@ public class TimeoutsHolder { AtomicIntegerFieldUpdater.newUpdater(TimeoutsHolder.class, "cancelled"); private static final AtomicIntegerFieldUpdater STARTED = AtomicIntegerFieldUpdater.newUpdater(TimeoutsHolder.class, "started"); + private static final AtomicIntegerFieldUpdater READ_STATE = + AtomicIntegerFieldUpdater.newUpdater(TimeoutsHolder.class, "readState"); + + // How the read timeout is armed. Separate: on an entry of its own. Pending: start armed the request + // timeout's entry for when a read timeout would be due, so that once the read timeout starts that one + // entry serves both. Covered: the read timeout has started and rides on the request timeout's entry. + private static final int READ_SEPARATE = 0; + private static final int READ_PENDING = 1; + private static final int READ_COVERED = 2; private volatile int cancelled; private volatile int started; + private volatile int readState; + // Whether the request timeout's entry is armed ahead of the request deadline, to check the read timeout. + private volatile boolean requestTimeoutArmedEarly; private final Timer nettyTimer; private volatile @Nullable EventExecutor eventExecutor; private final NettyRequestSender requestSender; @@ -142,8 +154,18 @@ public void start() { // reading the clock again would only expose the deadline to a step between the two reads. An // absolute deadline was anchored before this holder existed, so there the remainder is the budget, // floored at zero: a task armed at zero still runs, and running is how the exchange gets failed. - arm(requestTimeoutTask, absoluteDeadline - ? Math.max(remainingBudget(requestTimeoutValue, nettyResponseFuture, nanoClock), 0L) : requestTimeoutValue); + long untilDeadline = absoluteDeadline + ? Math.max(remainingBudget(requestTimeoutValue, nettyResponseFuture, nanoClock), 0L) : requestTimeoutValue; + if (readTimeoutValue != -1 && readTimeoutValue < untilDeadline) { + // The read timeout starts once the request is written, and would then take an entry of its own + // that the exchange almost always cancels unfired. Arming this one for when a read timeout + // would be due instead lets it serve both: every entry costs a shared timer two queue + // operations and a slot on its single thread, and an event loop a wakeup. + readState = READ_PENDING; + armRequestTimeout(requestTimeoutTask, readTimeoutValue, true); + } else { + armRequestTimeout(requestTimeoutTask, untilDeadline, false); + } } } @@ -191,10 +213,27 @@ long requestTimeoutMillisTime() { return requestTimeoutMillisTime; } + // The clock lastTouch is read against, which tests replace. + long millisTime() { + return millisClock.getAsLong(); + } + boolean isRequestTimeoutDisabled() { return requestTimeoutTask == null; } + long readTimeout() { + return readTimeoutValue; + } + + boolean isRequestTimeoutArmedEarly() { + return requestTimeoutArmedEarly; + } + + boolean isReadTimeoutCovered() { + return readState == READ_COVERED; + } + /** * Moves this exchange's timeouts onto {@code executor}, the loop of the channel it turned out to run on. The * connect path arms the request timeout before there is a channel -- deliberately, since it bounds address @@ -210,7 +249,7 @@ public void rehomeOn(EventExecutor executor) { if (task == null || cancelled != 0 || task.isClaimed() || !task.cancelArmed()) { return; } - arm(task, remainingRequestTimeout()); + armRequestTimeoutFrom(task, millisClock.getAsLong(), false); } public void setResolvedRemoteAddress(InetSocketAddress address) { @@ -222,11 +261,60 @@ InetSocketAddress remoteAddress() { } public void startReadTimeout() { - if (readTimeoutValue != -1) { + if (readTimeoutValue != -1 && !coverReadTimeout()) { startReadTimeout(null); } } + /** + * Hands the read timeout to the request timeout's entry when {@link #start} armed that one early for it. + * Loses to an entry that already fired without finding a read timeout to check, which hands the read + * timeout back to an entry of its own. + */ + private boolean coverReadTimeout() { + int state = readState; + return state == READ_COVERED || state == READ_PENDING && READ_STATE.compareAndSet(this, READ_PENDING, READ_COVERED); + } + + /** + * Re-arms the request timeout after its entry fired ahead of the request deadline and found no timeout due. + * + * @param suspended whether response body reads are suspended, which holds off the read timeout for a whole + * window however long ago the channel was last read from + */ + void rearmRequestTimeout(RequestTimeoutTimerTask task, long now, boolean suspended) { + // A read timeout that has not started by now may not start for a long time -- a slow connect or upload + // -- and checking for it every window would wake the scheduler for nothing, so it gets its own entry + // when it does start. + READ_STATE.compareAndSet(this, READ_PENDING, READ_SEPARATE); + armRequestTimeoutFrom(task, now, suspended); + } + + private void armRequestTimeoutFrom(RequestTimeoutTimerTask task, long now, boolean suspended) { + // Floored at zero rather than passed on negative: a scheduler has no use for a negative delay, and the + // task has to run either way, since running is what fails the exchange. + long untilDeadline = Math.max(requestTimeoutMillisTime - now, 0L); + long untilRead = Long.MAX_VALUE; + int state = readState; + if (state == READ_PENDING) { + untilRead = readTimeoutValue; + } else if (state == READ_COVERED) { + untilRead = suspended ? readTimeoutValue + : Math.max(nettyResponseFuture.getLastTouch() + readTimeoutValue - now, 0L); + } + if (untilRead < untilDeadline) { + armRequestTimeout(task, untilRead, true); + } else { + armRequestTimeout(task, untilDeadline, false); + } + } + + private void armRequestTimeout(RequestTimeoutTimerTask task, long delay, boolean early) { + // Written before arming: on an event loop the entry can run before arm returns. + requestTimeoutArmedEarly = early; + arm(task, delay); + } + void startReadTimeout(@Nullable ReadTimeoutTimerTask task) { if (requestTimeoutTask == null || !requestTimeoutTask.isClaimed() && readTimeoutValue < requestTimeoutMillisTime - millisClock.getAsLong()) { @@ -258,12 +346,6 @@ private static void release(@Nullable TimeoutTimerTask task) { } } - private long remainingRequestTimeout() { - // Floored at zero rather than passed on negative: a scheduler has no use for a negative delay, and the - // task has to run either way, since running is what fails the exchange. - return Math.max(requestTimeoutMillisTime - millisClock.getAsLong(), 0L); - } - /** * Arms {@code task} to run after {@code delay} milliseconds, recording the scheduled entry on the task so it * can cancel itself later. Leaves it unarmed when the client is shutting down, in which case there is no diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java new file mode 100644 index 000000000..73f80f3f6 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java @@ -0,0 +1,277 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.netty.timeout; + +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timeout; +import io.netty.util.Timer; +import io.netty.util.TimerTask; +import org.asynchttpclient.AsyncCompletionHandler; +import org.asynchttpclient.AsyncHttpClientConfig; +import org.asynchttpclient.AsyncHttpClientState; +import org.asynchttpclient.DefaultAsyncHttpClientConfig; +import org.asynchttpclient.Request; +import org.asynchttpclient.RequestBuilder; +import org.asynchttpclient.Response; +import org.asynchttpclient.channel.ChannelPoolPartitioning; +import org.asynchttpclient.netty.NettyResponseBodyControl; +import org.asynchttpclient.netty.NettyResponseFuture; +import org.asynchttpclient.netty.channel.ChannelManager; +import org.asynchttpclient.netty.request.NettyRequestSender; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicLong; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * One timer entry serving both the request and the read timeout of an exchange. When the read timeout is the + * shorter of the two, {@link TimeoutsHolder#start} arms the request timeout's entry for when a read timeout + * would be due, and the read timeout then rides on it instead of taking an entry of its own. + *

+ * The holder's clock and the timer are both fakes: the timer records what is armed and runs nothing, so each + * test fires an entry by hand at the instant it sets on the clock. + */ +public class RequestTimeoutTimerTaskTest { + + private static final long READ_TIMEOUT_MS = 1_000; + private static final long REQUEST_TIMEOUT_MS = 5_000; + + private final AtomicLong elapsedMillis = new AtomicLong(); + private final List armed = new ArrayList<>(); + private long origin; + + private HashedWheelTimer senderTimer; + private ChannelManager channelManager; + private Timer timer; + private NettyResponseFuture future; + + @BeforeEach + void setUp() { + timer = mock(Timer.class); + when(timer.newTimeout(any(), anyLong(), any())).thenAnswer(invocation -> { + Timeout timeout = mock(Timeout.class); + armed.add(new Armed(invocation.getArgument(0), invocation.getArgument(1), timeout)); + return timeout; + }); + } + + @AfterEach + void tearDown() { + if (channelManager != null) { + channelManager.close(); + } + if (senderTimer != null) { + senderTimer.stop(); + } + } + + @Test + public void oneEntryServesBothTimeouts() { + TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + holder.startReadTimeout(); + + assertEquals(1, armed.size(), "the read timeout should not have taken an entry of its own"); + assertInstanceOf(RequestTimeoutTimerTask.class, armed.get(0).task); + assertEquals(READ_TIMEOUT_MS, armed.get(0).delay); + } + + @Test + public void anEarlyEntryWaitsOnlyForWhatIsLeftOfTheReadWindow() { + TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + holder.startReadTimeout(); + + fire(0, 600); + + assertFalse(future.isDone()); + assertEquals(2, armed.size()); + assertSame(armed.get(0).task, armed.get(1).task); + assertEquals(400, armed.get(1).delay, "the entry should be due when the read window runs out"); + } + + @Test + public void aReadTimeoutIsDeliveredFromTheSharedEntry() { + TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + holder.startReadTimeout(); + + fire(0, READ_TIMEOUT_MS); + + assertTimedOut("Read timeout"); + assertEquals(1, armed.size(), "an expired exchange should arm nothing more"); + } + + @Test + public void aReadTimeoutThatHasNotStartedGetsItsOwnEntry() { + TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + + // Still connecting or writing when the entry fires: nothing to check yet, and checking again every + // window until the read timeout starts would wake the timer for nothing. + fire(0, READ_TIMEOUT_MS); + assertEquals(2, armed.size()); + assertEquals(REQUEST_TIMEOUT_MS - READ_TIMEOUT_MS, armed.get(1).delay, + "the entry should go back to the request deadline"); + + holder.startReadTimeout(); + assertEquals(3, armed.size()); + assertInstanceOf(ReadTimeoutTimerTask.class, armed.get(2).task); + assertEquals(READ_TIMEOUT_MS, armed.get(2).delay); + } + + @Test + public void aRequestDeadlineAheadOfTheReadTimeoutTakesTheEntryAlone() { + TimeoutsHolder holder = startedHolder(READ_TIMEOUT_MS / 2); + holder.startReadTimeout(); + + assertEquals(1, armed.size()); + assertEquals(READ_TIMEOUT_MS / 2, armed.get(0).delay); + + fire(0, READ_TIMEOUT_MS / 2); + assertTimedOut("Request timeout"); + } + + @Test + public void anEarlyEntryThatFiresPastTheRequestDeadlineDeliversTheRequestTimeout() { + TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + holder.startReadTimeout(); + + // A timer runs an entry on the first tick at or after it is due, which can be past a deadline that + // fell between the two. + fire(0, REQUEST_TIMEOUT_MS); + + assertTimedOut("Request timeout"); + } + + @Test + public void aSuspendedBodyHoldsOffTheReadTimeoutForAWholeWindow() { + TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + holder.startReadTimeout(); + EmbeddedChannel channel = new EmbeddedChannel(); + try { + NettyResponseBodyControl.create(future, channel, () -> { }, ignored -> { }).suspend(); + + fire(0, 3 * READ_TIMEOUT_MS); + + assertFalse(future.isDone(), "a suspended body is not a read timeout"); + assertEquals(2, armed.size()); + assertEquals(READ_TIMEOUT_MS, armed.get(1).delay); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + public void cancellingAfterAnEarlyFireTakesTheNewEntryBackOut() { + TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + holder.startReadTimeout(); + fire(0, READ_TIMEOUT_MS / 2); + + holder.cancel(); + + verify(armed.get(1).timeout).cancel(); + RequestTimeoutTimerTask task = (RequestTimeoutTimerTask) armed.get(1).task; + assertTrue(task.isClaimed()); + assertNull(task.nettyResponseFuture, "a cancelled entry should not pin the exchange"); + } + + @Test + public void aCompletedExchangeArmsNothingMore() { + TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + holder.startReadTimeout(); + future.done(); + + fire(0, READ_TIMEOUT_MS); + + assertEquals(1, armed.size(), "a finished exchange should arm nothing more"); + assertTrue(((RequestTimeoutTimerTask) armed.get(0).task).isClaimed()); + } + + private TimeoutsHolder startedHolder(long requestTimeoutMs) { + AsyncHttpClientConfig config = new DefaultAsyncHttpClientConfig.Builder() + .setReadTimeout(Duration.ofMillis(READ_TIMEOUT_MS)) + .setRequestTimeout(Duration.ofMillis(requestTimeoutMs)) + .build(); + senderTimer = new HashedWheelTimer(); + channelManager = new ChannelManager(config, senderTimer); + NettyRequestSender requestSender = new NettyRequestSender(config, channelManager, senderTimer, + mock(AsyncHttpClientState.class)); + + Request request = new RequestBuilder().setUrl("http://example.com:12345").build(); + future = new NettyResponseFuture<>(request, new AsyncCompletionHandler() { + @Override + public Object onCompleted(Response response) { + return null; + } + }, null, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null); + origin = future.getLastTouch(); + + TimeoutsHolder holder = new TimeoutsHolder(timer, null, future, requestSender, config, null, + this::millis, System::nanoTime); + holder.start(); + return holder; + } + + private void fire(int entry, long atElapsedMillis) { + elapsedMillis.set(atElapsedMillis); + Armed armedEntry = armed.get(entry); + try { + armedEntry.task.run(armedEntry.timeout); + } catch (Exception e) { + throw new AssertionError(e); + } + } + + private void assertTimedOut(String prefix) { + ExecutionException failure = assertThrows(ExecutionException.class, () -> future.get(0, TimeUnit.MILLISECONDS)); + assertInstanceOf(TimeoutException.class, failure.getCause()); + assertTrue(failure.getCause().getMessage().startsWith(prefix), failure.getCause().getMessage()); + } + + private long millis() { + return origin + elapsedMillis.get(); + } + + private static final class Armed { + final TimerTask task; + final long delay; + final Timeout timeout; + + Armed(TimerTask task, long delay, Timeout timeout) { + this.task = task; + this.delay = delay; + this.timeout = timeout; + } + } +} From f2cddcc6f2fa0de777135053191630234549c749 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Fri, 6 Nov 2026 21:18:29 +0100 Subject: [PATCH 04/11] Fix clock and handle races in timeout re-arms Review of the shared request/read entry found two defects in it. The request deadline was re-derived from System.currentTimeMillis() every time the shared entry fired early. A timer entry is timed on a monotonic clock, so before that change a step of the wall clock never moved a request timeout; afterwards a step forward failed a request early with a request timeout and a step back pushed its deadline out. The holder now keeps the deadline on its nano clock as well and re-arms and checks against that. rehomeOn re-arms from it too. The wall clock is still what lastTouch, and so the read timeout, is measured on. A task that re-arms itself can record its new entry from the thread it fires on before the thread that armed it has recorded the old one. The late write then replaced the live entry with the spent one, so cancel() could no longer take the live entry out, and it held the future until the request deadline. Clearing the handle in two steps in cancelArmed had the same effect against a concurrent re-arm. The handle is now one field updated atomically: an entry that has already run never replaces one that can still run, and cancelArmed takes the handle in one step. Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5.5 --- .../timeout/RequestTimeoutTimerTask.java | 4 +- .../netty/timeout/TimeoutTimerTask.java | 72 +++++-- .../netty/timeout/TimeoutsHolder.java | 33 ++- .../timeout/RequestTimeoutTimerTaskTest.java | 190 +++++++++++++++--- 4 files changed, 249 insertions(+), 50 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java index 680e39709..b88dbfb00 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java @@ -82,10 +82,10 @@ private boolean checkReadTimeout() { return true; } - long now = timeoutsHolder.millisTime(); - if (now >= timeoutsHolder.requestTimeoutMillisTime()) { + if (timeoutsHolder.isRequestDeadlinePassed()) { return false; } + long now = timeoutsHolder.millisTime(); boolean suspended = NettyResponseBodyControl.isSuspended(nettyResponseFuture); if (timeoutsHolder.isReadTimeoutCovered() && !suspended) { long readTimeout = timeoutsHolder.readTimeout(); diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java index 6e81b7007..31d481700 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java @@ -26,8 +26,10 @@ import java.net.InetSocketAddress; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; /** * A timeout that can be armed either on a {@link io.netty.util.Timer} or on an event loop; which one an @@ -39,16 +41,19 @@ public abstract class TimeoutTimerTask implements TimerTask { private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutTimerTask.class); + private static final AtomicReferenceFieldUpdater HANDLE = + AtomicReferenceFieldUpdater.newUpdater(TimeoutTimerTask.class, Object.class, "handle"); protected final AtomicBoolean done = new AtomicBoolean(); protected final NettyRequestSender requestSender; final TimeoutsHolder timeoutsHolder; volatile NettyResponseFuture nettyResponseFuture; - // The scheduled entry this task is armed on, one field per scheduler so that a scheduler changing its - // return type is a compile error rather than a cancellation that silently stops working. At most one is - // ever set. Held here rather than in a wrapper so arming allocates nothing beyond what the scheduler needs. - private volatile @Nullable Timeout timerHandle; - private volatile @Nullable ScheduledFuture loopHandle; + // The scheduled entry this task is armed on: a timer's Timeout or an event loop's ScheduledFuture, which + // only the typed armedOn overloads below can store. One field, so that recording an entry and taking it + // back are each a single atomic step: a task that re-arms itself can record an entry from the thread it + // fires on while the thread that armed it is still recording the previous one. Held here rather than in a + // wrapper so arming allocates nothing beyond what the scheduler needs. + private volatile @Nullable Object handle; TimeoutTimerTask(NettyResponseFuture nettyResponseFuture, NettyRequestSender requestSender, TimeoutsHolder timeoutsHolder) { this.nettyResponseFuture = nettyResponseFuture; @@ -57,15 +62,46 @@ public abstract class TimeoutTimerTask implements TimerTask { } void armedOn(Timeout handle) { - // Each clears the other, so a handle left over from a previous arming cannot mask the live one and - // leave its entry sitting in a scheduler, holding the future until a deadline nobody is waiting for. - loopHandle = null; - timerHandle = handle; + record(handle); } void armedOn(ScheduledFuture handle) { - timerHandle = null; - loopHandle = handle; + record(handle); + } + + /** + * Records {@code entry} in place of whatever entry was recorded before, unless that one can still run and + * {@code entry} already has. That is the entry of an arming that lost the race against its own firing: + * its run re-armed the task and recorded the newer entry first, and overwriting that one would leave it in + * its scheduler with nobody to cancel it, holding the future until a deadline nobody is waiting for. + */ + private void record(Object entry) { + for (;;) { + Object current = handle; + if (current != null && canStillRun(current) && hasRun(entry)) { + return; + } + if (HANDLE.compareAndSet(this, current, entry)) { + return; + } + } + } + + private static boolean canStillRun(Object entry) { + if (entry instanceof Timeout) { + Timeout timeout = (Timeout) entry; + return !timeout.isExpired() && !timeout.isCancelled(); + } + return !((ScheduledFuture) entry).isDone(); + } + + // An event loop marks an entry done only once it has run, so one that is running counts by being due. + private static boolean hasRun(Object entry) { + if (entry instanceof Timeout) { + return ((Timeout) entry).isExpired(); + } + ScheduledFuture scheduled = (ScheduledFuture) entry; + return scheduled.isDone() || scheduled.getDelay(TimeUnit.NANOSECONDS) <= 0; } /** @@ -75,14 +111,14 @@ void armedOn(ScheduledFuture handle) { * @return whether an entry was taken back out of its scheduler before it could run */ boolean cancelArmed() { - Timeout timer = timerHandle; - if (timer != null) { - timerHandle = null; - return timer.cancel(); + // Taken in one step: a read followed by a separate clear could clear an entry the task recorded in + // between from its own firing, and that one would then never be cancelled. + Object entry = HANDLE.getAndSet(this, null); + if (entry instanceof Timeout) { + return ((Timeout) entry).cancel(); } - ScheduledFuture scheduled = loopHandle; - if (scheduled != null) { - loopHandle = null; + if (entry != null) { + ScheduledFuture scheduled = (ScheduledFuture) entry; try { return scheduled.cancel(false); } catch (RejectedExecutionException e) { diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java index 214a4f6b5..cb330be98 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -65,6 +65,10 @@ public class TimeoutsHolder { private volatile @Nullable EventExecutor eventExecutor; private final NettyRequestSender requestSender; private final long requestTimeoutMillisTime; + // The same deadline on nanoClock, for everything that re-arms or checks it after start: a timer entry is + // already timed on a monotonic clock, and a deadline re-derived from the wall clock would move with every + // step of it. + private final long requestDeadlineNanos; private final long requestTimeoutValue; private final boolean absoluteDeadline; private final long readTimeoutValue; @@ -123,11 +127,14 @@ public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, N // exchange has already spent bounds it as a whole instead. Which one applies is the caller's // choice, per request or per client. Left negative when the deadline is already behind us, which is // what stops startReadTimeout arming a sibling for an exchange that is over. - requestTimeoutMillisTime = millisClock.getAsLong() - + (absoluteDeadline ? remainingBudget(requestTimeoutInMs, nettyResponseFuture, nanoClock) : requestTimeoutInMs); + long untilDeadline = absoluteDeadline + ? remainingBudget(requestTimeoutInMs, nettyResponseFuture, nanoClock) : requestTimeoutInMs; + requestTimeoutMillisTime = millisClock.getAsLong() + untilDeadline; + requestDeadlineNanos = nanoClock.getAsLong() + TimeUnit.MILLISECONDS.toNanos(untilDeadline); requestTimeoutTask = new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs); } else { requestTimeoutMillisTime = -1L; + requestDeadlineNanos = 0L; requestTimeoutTask = null; } } @@ -234,6 +241,24 @@ boolean isReadTimeoutCovered() { return readState == READ_COVERED; } + boolean isRequestDeadlinePassed() { + return requestDeadlineNanos - nanoClock.getAsLong() <= 0; + } + + /** + * What is left until the request deadline in whole milliseconds, rounded up so an entry never runs ahead of + * it, and floored at zero: a scheduler has no use for a negative delay, and the task has to run either way, + * since running is what fails the exchange. + */ + private long untilRequestDeadline() { + long left = requestDeadlineNanos - nanoClock.getAsLong(); + if (left <= 0) { + return 0L; + } + long millis = TimeUnit.NANOSECONDS.toMillis(left); + return TimeUnit.MILLISECONDS.toNanos(millis) == left ? millis : millis + 1; + } + /** * Moves this exchange's timeouts onto {@code executor}, the loop of the channel it turned out to run on. The * connect path arms the request timeout before there is a channel -- deliberately, since it bounds address @@ -291,9 +316,7 @@ void rearmRequestTimeout(RequestTimeoutTimerTask task, long now, boolean suspend } private void armRequestTimeoutFrom(RequestTimeoutTimerTask task, long now, boolean suspended) { - // Floored at zero rather than passed on negative: a scheduler has no use for a negative delay, and the - // task has to run either way, since running is what fails the exchange. - long untilDeadline = Math.max(requestTimeoutMillisTime - now, 0L); + long untilDeadline = untilRequestDeadline(); long untilRead = Long.MAX_VALUE; int state = readState; if (state == READ_PENDING) { diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java index 73f80f3f6..cc2c5b7f6 100644 --- a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java @@ -19,7 +19,8 @@ import io.netty.util.HashedWheelTimer; import io.netty.util.Timeout; import io.netty.util.Timer; -import io.netty.util.TimerTask; +import io.netty.util.concurrent.EventExecutor; +import io.netty.util.concurrent.ScheduledFuture; import org.asynchttpclient.AsyncCompletionHandler; import org.asynchttpclient.AsyncHttpClientConfig; import org.asynchttpclient.AsyncHttpClientState; @@ -32,6 +33,7 @@ import org.asynchttpclient.netty.NettyResponseFuture; import org.asynchttpclient.netty.channel.ChannelManager; import org.asynchttpclient.netty.request.NettyRequestSender; +import org.jetbrains.annotations.Nullable; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -42,7 +44,6 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicLong; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -52,6 +53,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; @@ -62,21 +64,29 @@ * shorter of the two, {@link TimeoutsHolder#start} arms the request timeout's entry for when a read timeout * would be due, and the read timeout then rides on it instead of taking an entry of its own. *

- * The holder's clock and the timer are both fakes: the timer records what is armed and runs nothing, so each - * test fires an entry by hand at the instant it sets on the clock. + * The holder's clocks, the timer and the event loop are all fakes: the schedulers record what is armed and run + * nothing, so each test fires an entry by hand at the instant it sets on the clocks. The wall clock can be + * stepped apart from the monotonic one, as a clock correction does. */ public class RequestTimeoutTimerTaskTest { private static final long READ_TIMEOUT_MS = 1_000; private static final long REQUEST_TIMEOUT_MS = 5_000; + private static final long ONE_HOUR_MS = TimeUnit.HOURS.toMillis(1); - private final AtomicLong elapsedMillis = new AtomicLong(); private final List armed = new ArrayList<>(); - private long origin; + private long elapsedMillis; + private long wallClockStepMillis; + private long wallOrigin; + private long nanoOrigin; + // When set, the first entry armed runs before the scheduler hands its handle back, as it can when the + // arming thread is descheduled for longer than the entry's delay. + private boolean firstEntryRunsBeforeItIsRecorded; private HashedWheelTimer senderTimer; private ChannelManager channelManager; private Timer timer; + private EventExecutor loop; private NettyResponseFuture future; @BeforeEach @@ -84,8 +94,21 @@ void setUp() { timer = mock(Timer.class); when(timer.newTimeout(any(), anyLong(), any())).thenAnswer(invocation -> { Timeout timeout = mock(Timeout.class); - armed.add(new Armed(invocation.getArgument(0), invocation.getArgument(1), timeout)); - return timeout; + Armed entry = arm(invocation.getArgument(0), invocation.getArgument(1), timeout); + when(timeout.isExpired()).thenAnswer(ignored -> entry.ran); + when(timeout.cancel()).thenAnswer(ignored -> !entry.ran); + return handBack(entry, timeout); + }); + loop = mock(EventExecutor.class); + when(loop.schedule(any(Runnable.class), anyLong(), any())).thenAnswer(invocation -> { + ScheduledFuture scheduled = mock(ScheduledFuture.class); + long delay = invocation.getArgument(1); + Armed entry = arm(invocation.getArgument(0), delay, scheduled); + when(scheduled.isDone()).thenAnswer(ignored -> entry.done); + when(scheduled.getDelay(any())).thenAnswer(ignored -> + entry.ran ? 0L : TimeUnit.MILLISECONDS.toNanos(Math.max(delay, 1L))); + when(scheduled.cancel(anyBoolean())).thenAnswer(ignored -> !entry.ran); + return handBack(entry, scheduled); }); } @@ -162,6 +185,20 @@ public void aRequestDeadlineAheadOfTheReadTimeoutTakesTheEntryAlone() { assertTimedOut("Request timeout"); } + @Test + public void anAbsoluteDeadlineWithLessLeftThanTheReadTimeoutTakesTheEntryAlone() { + // An earlier hop already spent most of the exchange's budget. + TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, true, false, null, REQUEST_TIMEOUT_MS - READ_TIMEOUT_MS / 2); + holder.start(); + holder.startReadTimeout(); + + assertEquals(1, armed.size(), "the read timeout cannot be due before what is left of the request timeout"); + assertEquals(READ_TIMEOUT_MS / 2, armed.get(0).delay); + + fire(0, REQUEST_TIMEOUT_MS); + assertTimedOut("Request timeout"); + } + @Test public void anEarlyEntryThatFiresPastTheRequestDeadlineDeliversTheRequestTimeout() { TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); @@ -174,6 +211,30 @@ public void anEarlyEntryThatFiresPastTheRequestDeadlineDeliversTheRequestTimeout assertTimedOut("Request timeout"); } + @Test + public void aWallClockStepForwardDoesNotBringTheRequestDeadlineForward() { + startedHolder(REQUEST_TIMEOUT_MS); + + wallClockStepMillis = ONE_HOUR_MS; + fire(0, READ_TIMEOUT_MS); + + assertFalse(future.isDone(), "a step of the wall clock is not time spent on the request"); + assertEquals(2, armed.size()); + assertEquals(REQUEST_TIMEOUT_MS - READ_TIMEOUT_MS, armed.get(1).delay); + } + + @Test + public void aWallClockStepBackDoesNotPushTheRequestDeadlineBack() { + startedHolder(REQUEST_TIMEOUT_MS); + + wallClockStepMillis = -ONE_HOUR_MS; + fire(0, READ_TIMEOUT_MS); + + assertEquals(2, armed.size()); + assertEquals(REQUEST_TIMEOUT_MS - READ_TIMEOUT_MS, armed.get(1).delay, + "the request deadline should stay where the monotonic clock put it"); + } + @Test public void aSuspendedBodyHoldsOffTheReadTimeoutForAWholeWindow() { TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); @@ -200,12 +261,52 @@ public void cancellingAfterAnEarlyFireTakesTheNewEntryBackOut() { holder.cancel(); - verify(armed.get(1).timeout).cancel(); + verify((Timeout) armed.get(1).timeout).cancel(); RequestTimeoutTimerTask task = (RequestTimeoutTimerTask) armed.get(1).task; assertTrue(task.isClaimed()); assertNull(task.nettyResponseFuture, "a cancelled entry should not pin the exchange"); } + @Test + public void anEntryThatFiresBeforeItIsRecordedDoesNotHideItsSuccessorOnTheTimer() { + firstEntryRunsBeforeItIsRecorded = true; + TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + assertEquals(2, armed.size(), "the first entry should have re-armed while it was still being recorded"); + + holder.cancel(); + + verify((Timeout) armed.get(1).timeout).cancel(); + } + + @Test + public void anEntryThatFiresBeforeItIsRecordedDoesNotHideItsSuccessorOnTheLoop() { + firstEntryRunsBeforeItIsRecorded = true; + TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, true, loop, 0); + holder.start(); + assertEquals(2, armed.size(), "the first entry should have re-armed while it was still being recorded"); + + holder.cancel(); + + verify((ScheduledFuture) armed.get(1).timeout).cancel(false); + } + + @Test + public void rehomingBeforeTheReadTimeoutStartsKeepsTheEntryEarly() { + // The connect path arms on the timer, before there is a channel whose loop could carry the entry. + TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, true, null, 0); + holder.start(); + assertEquals(1, armed.size()); + + elapsedMillis = 300; + holder.rehomeOn(loop); + + verify((Timeout) armed.get(0).timeout).cancel(); + assertEquals(2, armed.size()); + assertInstanceOf(ScheduledFuture.class, armed.get(1).timeout, "the entry should have moved to the loop"); + assertEquals(READ_TIMEOUT_MS, armed.get(1).delay, "the entry should still be armed to check a read timeout"); + assertTrue(holder.isRequestTimeoutArmedEarly()); + } + @Test public void aCompletedExchangeArmsNothingMore() { TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); @@ -219,10 +320,17 @@ public void aCompletedExchangeArmsNothingMore() { } private TimeoutsHolder startedHolder(long requestTimeoutMs) { - AsyncHttpClientConfig config = new DefaultAsyncHttpClientConfig.Builder() - .setReadTimeout(Duration.ofMillis(READ_TIMEOUT_MS)) - .setRequestTimeout(Duration.ofMillis(requestTimeoutMs)) - .build(); + TimeoutsHolder holder = holder(requestTimeoutMs, false, false, null, 0); + holder.start(); + return holder; + } + + /** + * @param spentMillis how long the exchange ran before this attempt, which only an absolute deadline counts + */ + private TimeoutsHolder holder(long requestTimeoutMs, boolean absoluteDeadline, boolean useEventLoopTimeouts, + @Nullable EventExecutor executor, long spentMillis) { + AsyncHttpClientConfig config = config(requestTimeoutMs, useEventLoopTimeouts); senderTimer = new HashedWheelTimer(); channelManager = new ChannelManager(config, senderTimer); NettyRequestSender requestSender = new NettyRequestSender(config, channelManager, senderTimer, @@ -235,22 +343,46 @@ public Object onCompleted(Response response) { return null; } }, null, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null); - origin = future.getLastTouch(); + future.setUseAbsoluteRequestDeadline(absoluteDeadline); + wallOrigin = future.getLastTouch(); + nanoOrigin = future.getStartNanos() + TimeUnit.MILLISECONDS.toNanos(spentMillis); + return new TimeoutsHolder(timer, executor, future, requestSender, config, null, this::millis, this::nanos); + } - TimeoutsHolder holder = new TimeoutsHolder(timer, null, future, requestSender, config, null, - this::millis, System::nanoTime); - holder.start(); - return holder; + private static AsyncHttpClientConfig config(long requestTimeoutMs, boolean useEventLoopTimeouts) { + return new DefaultAsyncHttpClientConfig.Builder() + .setReadTimeout(Duration.ofMillis(READ_TIMEOUT_MS)) + .setRequestTimeout(Duration.ofMillis(requestTimeoutMs)) + .setUseEventLoopTimeouts(useEventLoopTimeouts) + .build(); + } + + private Armed arm(Object task, long delay, Object handle) { + Armed entry = new Armed((TimeoutTimerTask) task, delay, handle); + armed.add(entry); + return entry; + } + + private H handBack(Armed entry, H handle) { + if (firstEntryRunsBeforeItIsRecorded && armed.size() == 1) { + run(entry, READ_TIMEOUT_MS); + } + return handle; } private void fire(int entry, long atElapsedMillis) { - elapsedMillis.set(atElapsedMillis); - Armed armedEntry = armed.get(entry); + run(armed.get(entry), atElapsedMillis); + } + + private void run(Armed entry, long atElapsedMillis) { + elapsedMillis = atElapsedMillis; + entry.ran = true; try { - armedEntry.task.run(armedEntry.timeout); + entry.task.run(entry.timeout instanceof Timeout ? (Timeout) entry.timeout : null); } catch (Exception e) { throw new AssertionError(e); } + entry.done = true; } private void assertTimedOut(String prefix) { @@ -260,15 +392,23 @@ private void assertTimedOut(String prefix) { } private long millis() { - return origin + elapsedMillis.get(); + return wallOrigin + elapsedMillis + wallClockStepMillis; + } + + private long nanos() { + return nanoOrigin + TimeUnit.MILLISECONDS.toNanos(elapsedMillis); } private static final class Armed { - final TimerTask task; + final TimeoutTimerTask task; final long delay; - final Timeout timeout; + final Object timeout; + // Set when the entry starts running, as a timer marks it expired; done when it has finished, as an + // event loop marks it. + volatile boolean ran; + volatile boolean done; - Armed(TimerTask task, long delay, Timeout timeout) { + Armed(TimeoutTimerTask task, long delay, Object timeout) { this.task = task; this.delay = delay; this.timeout = timeout; From 3daa3b7d2125061fcfb89f22749bc8eb1e0ac69e Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Fri, 6 Nov 2026 21:31:13 +0100 Subject: [PATCH 05/11] Order timeout re-arms explicitly and cap the window Second review round on the shared request/read entry. Telling a stale handle from a live one by asking the scheduler whether an entry had run was a guess on an event loop: a loop marks an entry done only after it returns, so the code read "due" as "has run", and a successor that was already due when it was recorded was dropped as stale. cancel() then could not take it out. Each arming now takes a number from a per-task counter before it schedules, and an entry is recorded only if no later arming has begun, which says exactly what the guess approximated. The shared entry re-armed for what was left of the read window with no upper bound. A wall clock stepping back made that remainder longer than the request deadline, so the entry was armed for the deadline and the read timeout was not checked again before it. The remainder is now capped at a whole window, as the separate read timeout task does since #2377. Also covers the read timeout starting while the entry re-arms. Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5.5 --- .../netty/timeout/TimeoutTimerTask.java | 51 ++++++++-------- .../netty/timeout/TimeoutsHolder.java | 8 ++- .../timeout/RequestTimeoutTimerTaskTest.java | 59 ++++++++++++++++++- 3 files changed, 86 insertions(+), 32 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java index 31d481700..d02b8e10d 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java @@ -26,9 +26,9 @@ import java.net.InetSocketAddress; import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; /** @@ -43,6 +43,8 @@ public abstract class TimeoutTimerTask implements TimerTask { private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutTimerTask.class); private static final AtomicReferenceFieldUpdater HANDLE = AtomicReferenceFieldUpdater.newUpdater(TimeoutTimerTask.class, Object.class, "handle"); + private static final AtomicIntegerFieldUpdater ARMINGS = + AtomicIntegerFieldUpdater.newUpdater(TimeoutTimerTask.class, "armings"); protected final AtomicBoolean done = new AtomicBoolean(); protected final NettyRequestSender requestSender; @@ -54,6 +56,8 @@ public abstract class TimeoutTimerTask implements TimerTask { // fires on while the thread that armed it is still recording the previous one. Held here rather than in a // wrapper so arming allocates nothing beyond what the scheduler needs. private volatile @Nullable Object handle; + // How many times this task has been armed, so that an entry recorded late can tell it has been superseded. + private volatile int armings; TimeoutTimerTask(NettyResponseFuture nettyResponseFuture, NettyRequestSender requestSender, TimeoutsHolder timeoutsHolder) { this.nettyResponseFuture = nettyResponseFuture; @@ -61,49 +65,40 @@ public abstract class TimeoutTimerTask implements TimerTask { this.timeoutsHolder = timeoutsHolder; } - void armedOn(Timeout handle) { - record(handle); + /** + * Starts arming this task. The value returned identifies the arming to {@code armedOn}. + */ + int beginArming() { + return ARMINGS.incrementAndGet(this); + } + + void armedOn(Timeout handle, int arming) { + record(handle, arming); } - void armedOn(ScheduledFuture handle) { - record(handle); + void armedOn(ScheduledFuture handle, int arming) { + record(handle, arming); } /** - * Records {@code entry} in place of whatever entry was recorded before, unless that one can still run and - * {@code entry} already has. That is the entry of an arming that lost the race against its own firing: - * its run re-armed the task and recorded the newer entry first, and overwriting that one would leave it in - * its scheduler with nobody to cancel it, holding the future until a deadline nobody is waiting for. + * Records {@code entry}, unless the task has been armed again since {@code arming} began. That is an arming + * that lost the race against its own firing: its run re-armed the task, and overwriting the newer entry + * would leave it in its scheduler with nobody to cancel it, holding the future until a deadline nobody is + * waiting for. The entry passed over has already run, so there is nothing left to cancel on it. */ - private void record(Object entry) { + private void record(Object entry, int arming) { for (;;) { Object current = handle; - if (current != null && canStillRun(current) && hasRun(entry)) { + if (armings != arming) { return; } + // Fails when a newer arming recorded its entry after the check above, which the next pass then sees. if (HANDLE.compareAndSet(this, current, entry)) { return; } } } - private static boolean canStillRun(Object entry) { - if (entry instanceof Timeout) { - Timeout timeout = (Timeout) entry; - return !timeout.isExpired() && !timeout.isCancelled(); - } - return !((ScheduledFuture) entry).isDone(); - } - - // An event loop marks an entry done only once it has run, so one that is running counts by being due. - private static boolean hasRun(Object entry) { - if (entry instanceof Timeout) { - return ((Timeout) entry).isExpired(); - } - ScheduledFuture scheduled = (ScheduledFuture) entry; - return scheduled.isDone() || scheduled.getDelay(TimeUnit.NANOSECONDS) <= 0; - } - /** * Cancels the scheduled entry this task was armed on, if any. Never interrupts: on the event-loop path the * task may be running on the very thread this is called from, and nothing in it answers interruption. diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java index cb330be98..c72325054 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -322,8 +322,9 @@ private void armRequestTimeoutFrom(RequestTimeoutTimerTask task, long now, boole if (state == READ_PENDING) { untilRead = readTimeoutValue; } else if (state == READ_COVERED) { + // Capped at a whole window, so a wall clock stepping back cannot put the next check out of reach. untilRead = suspended ? readTimeoutValue - : Math.max(nettyResponseFuture.getLastTouch() + readTimeoutValue - now, 0L); + : Math.min(Math.max(nettyResponseFuture.getLastTouch() + readTimeoutValue - now, 0L), readTimeoutValue); } if (untilRead < untilDeadline) { armRequestTimeout(task, untilRead, true); @@ -384,6 +385,7 @@ private void arm(T task, long delay) { if (requestSender == null || requestSender.isClosed()) { return; } + int arming = task.beginArming(); EventExecutor executor = eventExecutor; if (executor != null && !executor.isShuttingDown()) { ScheduledFuture handle = null; @@ -398,7 +400,7 @@ private void arm(T task, long delay) { // Outside the try: only the schedule above may fall back to the timer. Anything thrown while // recording or unwinding the entry belongs to an exchange that is already armed. if (handle != null) { - task.armedOn(handle); + task.armedOn(handle, arming); cancelIfRaced(task); return; } @@ -406,7 +408,7 @@ private void arm(T task, long delay) { if (nettyTimer == null) { return; } - task.armedOn(nettyTimer.newTimeout(task, delay, TimeUnit.MILLISECONDS)); + task.armedOn(nettyTimer.newTimeout(task, delay, TimeUnit.MILLISECONDS), arming); cancelIfRaced(task); } diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java index cc2c5b7f6..f5e0c4c42 100644 --- a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java @@ -44,6 +44,7 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.function.Consumer; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -82,6 +83,11 @@ public class RequestTimeoutTimerTaskTest { // When set, the first entry armed runs before the scheduler hands its handle back, as it can when the // arming thread is descheduled for longer than the entry's delay. private boolean firstEntryRunsBeforeItIsRecorded; + // When set, an event loop reports every entry as due as soon as it is scheduled, as it does once the + // arming thread has been descheduled for longer than the delay. + private boolean entriesAreDueAtOnce; + // Runs as each entry is armed, before the scheduler hands its handle back. + private Consumer onArm = entry -> { }; private HashedWheelTimer senderTimer; private ChannelManager channelManager; @@ -106,7 +112,7 @@ void setUp() { Armed entry = arm(invocation.getArgument(0), delay, scheduled); when(scheduled.isDone()).thenAnswer(ignored -> entry.done); when(scheduled.getDelay(any())).thenAnswer(ignored -> - entry.ran ? 0L : TimeUnit.MILLISECONDS.toNanos(Math.max(delay, 1L))); + entry.ran || entriesAreDueAtOnce ? 0L : TimeUnit.MILLISECONDS.toNanos(delay)); when(scheduled.cancel(anyBoolean())).thenAnswer(ignored -> !entry.ran); return handBack(entry, scheduled); }); @@ -235,6 +241,41 @@ public void aWallClockStepBackDoesNotPushTheRequestDeadlineBack() { "the request deadline should stay where the monotonic clock put it"); } + @Test + public void aWallClockStepBackDoesNotPutTheNextReadCheckOutOfReach() { + TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + holder.startReadTimeout(); + + wallClockStepMillis = -ONE_HOUR_MS; + fire(0, READ_TIMEOUT_MS / 2); + + assertEquals(2, armed.size()); + assertEquals(READ_TIMEOUT_MS, armed.get(1).delay, "the read timeout should be checked again within a window"); + assertTrue(holder.isRequestTimeoutArmedEarly()); + } + + @Test + public void aReadTimeoutStartingWhileTheEntryReArmsGetsItsOwnEntry() { + TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + // The request is written while the entry, finding no read timeout yet, is re-arming for the deadline. + onArm = entry -> { + if (armed.size() == 2) { + holder.startReadTimeout(); + } + }; + + fire(0, READ_TIMEOUT_MS); + + assertEquals(3, armed.size()); + assertEquals(REQUEST_TIMEOUT_MS - READ_TIMEOUT_MS, armed.get(1).delay); + assertInstanceOf(ReadTimeoutTimerTask.class, armed.get(2).task, "the read timeout should not be lost"); + assertEquals(READ_TIMEOUT_MS, armed.get(2).delay); + + holder.cancel(); + verify((Timeout) armed.get(1).timeout).cancel(); + verify((Timeout) armed.get(2).timeout).cancel(); + } + @Test public void aSuspendedBodyHoldsOffTheReadTimeoutForAWholeWindow() { TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); @@ -290,6 +331,21 @@ public void anEntryThatFiresBeforeItIsRecordedDoesNotHideItsSuccessorOnTheLoop() verify((ScheduledFuture) armed.get(1).timeout).cancel(false); } + @Test + public void aSuccessorAlreadyDueOnTheLoopIsStillRecorded() { + entriesAreDueAtOnce = true; + TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, true, loop, 0); + holder.start(); + holder.startReadTimeout(); + + fire(0, READ_TIMEOUT_MS / 2); + assertEquals(2, armed.size()); + + holder.cancel(); + + verify((ScheduledFuture) armed.get(1).timeout).cancel(false); + } + @Test public void rehomingBeforeTheReadTimeoutStartsKeepsTheEntryEarly() { // The connect path arms on the timer, before there is a channel whose loop could carry the entry. @@ -364,6 +420,7 @@ private Armed arm(Object task, long delay, Object handle) { } private H handBack(Armed entry, H handle) { + onArm.accept(entry); if (firstEntryRunsBeforeItIsRecorded && armed.size() == 1) { run(entry, READ_TIMEOUT_MS); } From cbb6a78c1140ff1ccadd36d219cf158502b234a1 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Fri, 6 Nov 2026 21:42:27 +0100 Subject: [PATCH 06/11] Anchor absolute deadlines and fail on lost re-arms Third review round on the shared request/read entry. The monotonic request deadline took the absolute budget from one clock reading and added it to another, so time passing between the two in the constructor extended an absolute deadline by that much. It is now derived from the exchange's start directly. Re-arming the shared entry runs on the timer thread. A timer that rejects the new entry, a user timer that is full or stopped, used to throw there, where HashedWheelTimer only logs it, and the exchange was left with neither a read nor a request timeout. Before the shared entry the request timeout was never re-armed, so it stayed as a backstop. The exchange is now failed with the rejection instead of left to hang. The test fakes no longer stub scheduler state the code stopped reading. Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5.5 --- .../timeout/RequestTimeoutTimerTask.java | 11 +++- .../netty/timeout/TimeoutsHolder.java | 5 +- .../timeout/RequestTimeoutTimerTaskTest.java | 56 +++++++++++++------ 3 files changed, 54 insertions(+), 18 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java index b88dbfb00..853fbf1d5 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java @@ -100,10 +100,19 @@ private boolean checkReadTimeout() { } } + // Read before release: a concurrent cancel cleans the field once the task is unclaimed. + NettyResponseFuture future = nettyResponseFuture; // Released before re-arming: the holder may hand the read timeout an entry of its own, which is only // armed while this task is unclaimed. done.set(false); - timeoutsHolder.rearmRequestTimeout(this, now, suspended); + try { + timeoutsHolder.rearmRequestTimeout(this, now, suspended); + } catch (RuntimeException e) { + // A timer that rejects the entry, being full or stopped, would leave the exchange with no deadline + // at all, so it is failed instead of left to hang. + timeoutsHolder.cancel(); + requestSender.abort(future.channel(), future, e); + } return true; } } diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java index c72325054..fc774889b 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -130,7 +130,10 @@ public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, N long untilDeadline = absoluteDeadline ? remainingBudget(requestTimeoutInMs, nettyResponseFuture, nanoClock) : requestTimeoutInMs; requestTimeoutMillisTime = millisClock.getAsLong() + untilDeadline; - requestDeadlineNanos = nanoClock.getAsLong() + TimeUnit.MILLISECONDS.toNanos(untilDeadline); + // An absolute deadline straight off the exchange's start rather than off another clock reading, so + // that time passing between the two cannot extend it. + requestDeadlineNanos = (absoluteDeadline ? nettyResponseFuture.getStartNanos() : nanoClock.getAsLong()) + + TimeUnit.MILLISECONDS.toNanos(requestTimeoutInMs); requestTimeoutTask = new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs); } else { requestTimeoutMillisTime = -1L; diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java index f5e0c4c42..d867371d5 100644 --- a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java @@ -42,6 +42,7 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.ExecutionException; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.function.Consumer; @@ -83,9 +84,8 @@ public class RequestTimeoutTimerTaskTest { // When set, the first entry armed runs before the scheduler hands its handle back, as it can when the // arming thread is descheduled for longer than the entry's delay. private boolean firstEntryRunsBeforeItIsRecorded; - // When set, an event loop reports every entry as due as soon as it is scheduled, as it does once the - // arming thread has been descheduled for longer than the delay. - private boolean entriesAreDueAtOnce; + // When set, the monotonic clock jumps ahead by this much right after it is next read. + private long pauseAfterNextNanoReadMillis; // Runs as each entry is armed, before the scheduler hands its handle back. private Consumer onArm = entry -> { }; @@ -101,18 +101,13 @@ void setUp() { when(timer.newTimeout(any(), anyLong(), any())).thenAnswer(invocation -> { Timeout timeout = mock(Timeout.class); Armed entry = arm(invocation.getArgument(0), invocation.getArgument(1), timeout); - when(timeout.isExpired()).thenAnswer(ignored -> entry.ran); when(timeout.cancel()).thenAnswer(ignored -> !entry.ran); return handBack(entry, timeout); }); loop = mock(EventExecutor.class); when(loop.schedule(any(Runnable.class), anyLong(), any())).thenAnswer(invocation -> { ScheduledFuture scheduled = mock(ScheduledFuture.class); - long delay = invocation.getArgument(1); - Armed entry = arm(invocation.getArgument(0), delay, scheduled); - when(scheduled.isDone()).thenAnswer(ignored -> entry.done); - when(scheduled.getDelay(any())).thenAnswer(ignored -> - entry.ran || entriesAreDueAtOnce ? 0L : TimeUnit.MILLISECONDS.toNanos(delay)); + Armed entry = arm(invocation.getArgument(0), invocation.getArgument(1), scheduled); when(scheduled.cancel(anyBoolean())).thenAnswer(ignored -> !entry.ran); return handBack(entry, scheduled); }); @@ -205,6 +200,37 @@ public void anAbsoluteDeadlineWithLessLeftThanTheReadTimeoutTakesTheEntryAlone() assertTimedOut("Request timeout"); } + @Test + public void anAbsoluteDeadlineIsNotExtendedByTimePassingWhileTheHolderIsBuilt() { + pauseAfterNextNanoReadMillis = 3_000; + TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, true, false, null, 0); + holder.start(); + assertEquals(1, armed.size()); + + // The read timeout has not started, so the entry goes back to the request deadline. + fire(0, 4_000); + + assertEquals(2, armed.size()); + assertEquals(REQUEST_TIMEOUT_MS - 4_000, armed.get(1).delay, + "the exchange's deadline should stay where its start and budget put it"); + } + + @Test + public void anEntryTheTimerRejectsFailsTheExchangeRatherThanLeaveItWithoutADeadline() { + TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + holder.startReadTimeout(); + onArm = entry -> { + if (armed.size() == 2) { + throw new RejectedExecutionException("timer full"); + } + }; + + fire(0, READ_TIMEOUT_MS / 2); + + ExecutionException failure = assertThrows(ExecutionException.class, () -> future.get(0, TimeUnit.MILLISECONDS)); + assertInstanceOf(RejectedExecutionException.class, failure.getCause()); + } + @Test public void anEarlyEntryThatFiresPastTheRequestDeadlineDeliversTheRequestTimeout() { TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); @@ -332,8 +358,7 @@ public void anEntryThatFiresBeforeItIsRecordedDoesNotHideItsSuccessorOnTheLoop() } @Test - public void aSuccessorAlreadyDueOnTheLoopIsStillRecorded() { - entriesAreDueAtOnce = true; + public void cancellingAfterAnEarlyFireOnTheLoopTakesTheNewEntryBackOut() { TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, true, loop, 0); holder.start(); holder.startReadTimeout(); @@ -439,7 +464,6 @@ private void run(Armed entry, long atElapsedMillis) { } catch (Exception e) { throw new AssertionError(e); } - entry.done = true; } private void assertTimedOut(String prefix) { @@ -453,17 +477,17 @@ private long millis() { } private long nanos() { - return nanoOrigin + TimeUnit.MILLISECONDS.toNanos(elapsedMillis); + long nanos = nanoOrigin + TimeUnit.MILLISECONDS.toNanos(elapsedMillis); + elapsedMillis += pauseAfterNextNanoReadMillis; + pauseAfterNextNanoReadMillis = 0; + return nanos; } private static final class Armed { final TimeoutTimerTask task; final long delay; final Object timeout; - // Set when the entry starts running, as a timer marks it expired; done when it has finished, as an - // event loop marks it. volatile boolean ran; - volatile boolean done; Armed(TimeoutTimerTask task, long delay, Object timeout) { this.task = task; From fe9fc58dc27566fa1dc1b75eeb10257efefc1240 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Fri, 6 Nov 2026 21:52:21 +0100 Subject: [PATCH 07/11] Abort on a rejected re-arm only for a live holder Fourth review round on the shared request/read entry. The previous commit fails the exchange when the timer rejects the re-armed entry. The task releases itself before re-arming, so by the time the rejection lands the holder can have been retired: by completion, after which the channel may already serve another exchange from the pool, or by a redirect, retry or auth replay that installed the next attempt's holder. Aborting then failed an exchange this holder no longer speaks for and closed its channel. The abort now happens only when this call is the one that cancelled the holder, the holder is still the future's, and the future is not done. TimeoutsHolder.retire() is cancel() reporting whether it did the cancelling; cancel() keeps its public signature. Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5.5 --- .../timeout/RequestTimeoutTimerTask.java | 9 ++-- .../netty/timeout/TimeoutsHolder.java | 11 +++++ .../timeout/RequestTimeoutTimerTaskTest.java | 47 +++++++++++++++++-- 3 files changed, 60 insertions(+), 7 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java index 853fbf1d5..acdd06ea6 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java @@ -109,9 +109,12 @@ private boolean checkReadTimeout() { timeoutsHolder.rearmRequestTimeout(this, now, suspended); } catch (RuntimeException e) { // A timer that rejects the entry, being full or stopped, would leave the exchange with no deadline - // at all, so it is failed instead of left to hang. - timeoutsHolder.cancel(); - requestSender.abort(future.channel(), future, e); + // at all, so it is failed instead of left to hang. Only while this holder is still the exchange's: + // one that was retired meanwhile no longer speaks for it, and the channel may already be serving + // the next attempt or, back in the pool, another exchange. + if (timeoutsHolder.retire() && future.getTimeoutsHolder() == timeoutsHolder && !future.isDone()) { + requestSender.abort(future.channel(), future, e); + } } return true; } diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java index fc774889b..d00bd7acc 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -360,10 +360,21 @@ void startReadTimeout(@Nullable ReadTimeoutTimerTask task) { } public void cancel() { + retire(); + } + + /** + * {@link #cancel}, telling whether this call is the one that cancelled, so that a caller acting on the + * exchange's behalf can tell a holder still in charge of it from one a completion, a redirect, a retry or + * an auth replay has already retired. + */ + boolean retire() { if (CANCELLED.compareAndSet(this, 0, 1)) { release(requestTimeoutTask); release(readTimeoutTask); + return true; } + return false; } private static void release(@Nullable TimeoutTimerTask task) { diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java index d867371d5..f9212c7cf 100644 --- a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java @@ -94,6 +94,8 @@ public class RequestTimeoutTimerTaskTest { private Timer timer; private EventExecutor loop; private NettyResponseFuture future; + private NettyRequestSender requestSender; + private AsyncHttpClientConfig config; @BeforeEach void setUp() { @@ -217,7 +219,8 @@ public void anAbsoluteDeadlineIsNotExtendedByTimePassingWhileTheHolderIsBuilt() @Test public void anEntryTheTimerRejectsFailsTheExchangeRatherThanLeaveItWithoutADeadline() { - TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, false, null, 0); + future.setTimeoutsHolder(holder); holder.startReadTimeout(); onArm = entry -> { if (armed.size() == 2) { @@ -231,6 +234,43 @@ public void anEntryTheTimerRejectsFailsTheExchangeRatherThanLeaveItWithoutADeadl assertInstanceOf(RejectedExecutionException.class, failure.getCause()); } + @Test + public void aRejectedReArmOfARetiredHolderLeavesTheNextAttemptAlone() { + TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, false, null, 0); + future.setTimeoutsHolder(holder); + holder.startReadTimeout(); + // A redirect installs the next attempt's holder while the old entry is re-arming, and then the timer + // rejects the old entry. + onArm = entry -> { + if (armed.size() == 2) { + future.setTimeoutsHolder(new TimeoutsHolder(timer, null, future, requestSender, config, null, + this::millis, this::nanos)); + throw new RejectedExecutionException("timer full"); + } + }; + + fire(0, READ_TIMEOUT_MS / 2); + + assertFalse(future.isDone(), "the next attempt should not be failed by the holder it replaced"); + } + + @Test + public void aRejectedReArmAfterTheExchangeCompletedAbortsNothing() { + TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, false, null, 0); + future.setTimeoutsHolder(holder); + holder.startReadTimeout(); + onArm = entry -> { + if (armed.size() == 2) { + future.cancelTimeouts(); + throw new RejectedExecutionException("timer full"); + } + }; + + fire(0, READ_TIMEOUT_MS / 2); + + assertFalse(future.isDone(), "a holder retired by completion should leave the exchange alone"); + } + @Test public void anEarlyEntryThatFiresPastTheRequestDeadlineDeliversTheRequestTimeout() { TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); @@ -411,11 +451,10 @@ private TimeoutsHolder startedHolder(long requestTimeoutMs) { */ private TimeoutsHolder holder(long requestTimeoutMs, boolean absoluteDeadline, boolean useEventLoopTimeouts, @Nullable EventExecutor executor, long spentMillis) { - AsyncHttpClientConfig config = config(requestTimeoutMs, useEventLoopTimeouts); + config = config(requestTimeoutMs, useEventLoopTimeouts); senderTimer = new HashedWheelTimer(); channelManager = new ChannelManager(config, senderTimer); - NettyRequestSender requestSender = new NettyRequestSender(config, channelManager, senderTimer, - mock(AsyncHttpClientState.class)); + requestSender = new NettyRequestSender(config, channelManager, senderTimer, mock(AsyncHttpClientState.class)); Request request = new RequestBuilder().setUrl("http://example.com:12345").build(); future = new NettyResponseFuture<>(request, new AsyncCompletionHandler() { From e678f09c1a55bcdec1feed78d26b81bcfdaf74cd Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Fri, 6 Nov 2026 22:02:15 +0100 Subject: [PATCH 08/11] Test a rejected re-arm of a detached holder Fifth review round: the check that a holder is still the future's before a rejected re-arm aborts the exchange had no test of its own. Completion and the next attempt both take the holder off the future before they cancel it, and a rejection landing in between would fail an exchange that is completing. The new test detaches the holder at that point and fails without the check. Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5.5 --- .../timeout/RequestTimeoutTimerTaskTest.java | 31 +++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java index f9212c7cf..52c788be2 100644 --- a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java @@ -38,6 +38,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.lang.reflect.Field; import java.time.Duration; import java.util.ArrayList; import java.util.List; @@ -254,6 +255,25 @@ public void aRejectedReArmOfARetiredHolderLeavesTheNextAttemptAlone() { assertFalse(future.isDone(), "the next attempt should not be failed by the holder it replaced"); } + @Test + public void aRejectedReArmOfAHolderTheFutureNoLongerHoldsAbortsNothing() { + TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, false, null, 0); + future.setTimeoutsHolder(holder); + holder.startReadTimeout(); + // Completion and the next attempt both take the holder off the future before they cancel it; the + // rejection lands in between. + onArm = entry -> { + if (armed.size() == 2) { + detachTimeoutsHolder(); + throw new RejectedExecutionException("timer full"); + } + }; + + fire(0, READ_TIMEOUT_MS / 2); + + assertFalse(future.isDone(), "a holder the future no longer holds should leave the exchange alone"); + } + @Test public void aRejectedReArmAfterTheExchangeCompletedAbortsNothing() { TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, false, null, 0); @@ -469,6 +489,17 @@ public Object onCompleted(Response response) { return new TimeoutsHolder(timer, executor, future, requestSender, config, null, this::millis, this::nanos); } + // Takes the holder off the future without cancelling it, as NettyResponseFuture does before it cancels. + private void detachTimeoutsHolder() { + try { + Field field = NettyResponseFuture.class.getDeclaredField("timeoutsHolder"); + field.setAccessible(true); + field.set(future, null); + } catch (ReflectiveOperationException e) { + throw new AssertionError(e); + } + } + private static AsyncHttpClientConfig config(long requestTimeoutMs, boolean useEventLoopTimeouts) { return new DefaultAsyncHttpClientConfig.Builder() .setReadTimeout(Duration.ofMillis(READ_TIMEOUT_MS)) From 12291dff6e898810d5ff8a8d6bb8a38b54dc7bde Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Sat, 7 Nov 2026 10:11:39 +0100 Subject: [PATCH 09/11] Read the nano clock only for entries that move Keeping the request deadline on the nano clock added a System.nanoTime read to every holder. In the default configuration, a read timeout no shorter than the request timeout on the client's timer, nothing ever reads that deadline: the entry is armed once for it and cancelled. RequestTimeoutsArmBenchmark measured that attempt about 8% slower than main on one thread (3.9 -> 3.6 ops/us). The deadline is now set only where something reads it: in start() when the entry is armed early or rehomeOn may move it, anchored where the entry is armed, and from the exchange's start for an absolute deadline, which takes no clock reading. The default attempt is back at parity with main (3.83-3.94 against 3.87-3.96 ops/us over four runs of four forks), and the read-shorter-than-request case keeps its gain (2.53 -> 3.27 ops/us, 391 -> 223 bytes/op). Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5.5 --- .../netty/timeout/TimeoutsHolder.java | 22 ++++++++++++------- .../timeout/RequestTimeoutTimerTaskTest.java | 12 ++++++++++ 2 files changed, 26 insertions(+), 8 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java index d00bd7acc..094fa76d1 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -67,8 +67,9 @@ public class TimeoutsHolder { private final long requestTimeoutMillisTime; // The same deadline on nanoClock, for everything that re-arms or checks it after start: a timer entry is // already timed on a monotonic clock, and a deadline re-derived from the wall clock would move with every - // step of it. - private final long requestDeadlineNanos; + // step of it. Set only where something will read it -- an entry armed early, or one rehomeOn may move -- + // since reading the clock is a measurable part of what an attempt that needs neither costs. + private volatile long requestDeadlineNanos; private final long requestTimeoutValue; private final boolean absoluteDeadline; private final long readTimeoutValue; @@ -130,14 +131,14 @@ public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, N long untilDeadline = absoluteDeadline ? remainingBudget(requestTimeoutInMs, nettyResponseFuture, nanoClock) : requestTimeoutInMs; requestTimeoutMillisTime = millisClock.getAsLong() + untilDeadline; - // An absolute deadline straight off the exchange's start rather than off another clock reading, so - // that time passing between the two cannot extend it. - requestDeadlineNanos = (absoluteDeadline ? nettyResponseFuture.getStartNanos() : nanoClock.getAsLong()) - + TimeUnit.MILLISECONDS.toNanos(requestTimeoutInMs); + if (absoluteDeadline) { + // Straight off the exchange's start rather than off another clock reading, so that time passing + // between the two cannot extend it. + requestDeadlineNanos = nettyResponseFuture.getStartNanos() + TimeUnit.MILLISECONDS.toNanos(requestTimeoutInMs); + } requestTimeoutTask = new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs); } else { requestTimeoutMillisTime = -1L; - requestDeadlineNanos = 0L; requestTimeoutTask = null; } } @@ -166,7 +167,12 @@ public void start() { // floored at zero: a task armed at zero still runs, and running is how the exchange gets failed. long untilDeadline = absoluteDeadline ? Math.max(remainingBudget(requestTimeoutValue, nettyResponseFuture, nanoClock), 0L) : requestTimeoutValue; - if (readTimeoutValue != -1 && readTimeoutValue < untilDeadline) { + boolean early = readTimeoutValue != -1 && readTimeoutValue < untilDeadline; + if (!absoluteDeadline && (early || useEventLoopTimeouts)) { + // Anchored where the entry is armed, from the same duration. + requestDeadlineNanos = nanoClock.getAsLong() + TimeUnit.MILLISECONDS.toNanos(requestTimeoutValue); + } + if (early) { // The read timeout starts once the request is written, and would then take an entry of its own // that the exchange almost always cancels unfired. Arming this one for when a read timeout // would be due instead lets it serve both: every entry costs a shared timer two queue diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java index 52c788be2..30dcda12f 100644 --- a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java @@ -87,6 +87,7 @@ public class RequestTimeoutTimerTaskTest { private boolean firstEntryRunsBeforeItIsRecorded; // When set, the monotonic clock jumps ahead by this much right after it is next read. private long pauseAfterNextNanoReadMillis; + private int nanoReads; // Runs as each entry is armed, before the scheduler hands its handle back. private Consumer onArm = entry -> { }; @@ -177,6 +178,16 @@ public void aReadTimeoutThatHasNotStartedGetsItsOwnEntry() { assertEquals(READ_TIMEOUT_MS, armed.get(2).delay); } + @Test + public void anAttemptWithNothingToReArmDoesNotReadTheMonotonicClock() { + // The default: a read timeout no shorter than the request timeout, on the timer. + TimeoutsHolder holder = startedHolder(READ_TIMEOUT_MS); + holder.startReadTimeout(); + holder.cancel(); + + assertEquals(0, nanoReads, "only an entry that may be re-armed needs the request deadline on the nano clock"); + } + @Test public void aRequestDeadlineAheadOfTheReadTimeoutTakesTheEntryAlone() { TimeoutsHolder holder = startedHolder(READ_TIMEOUT_MS / 2); @@ -547,6 +558,7 @@ private long millis() { } private long nanos() { + nanoReads++; long nanos = nanoOrigin + TimeUnit.MILLISECONDS.toNanos(elapsedMillis); elapsedMillis += pauseAfterNextNanoReadMillis; pauseAfterNextNanoReadMillis = 0; From 0da14a665393fad1b6556da8df1dfdb0e4ac809b Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Sat, 7 Nov 2026 20:00:34 +0100 Subject: [PATCH 10/11] Cover review cases and bench a firing entry Review of #2382 asked for proof of three behaviours and for a benchmark of an exchange that outlives its read window. The completed-exchange test fired at the very end of the read window, where the expiry branch arms nothing either, so it passed without the isDone() check; it now fires inside the window. A new test runs a request timeout of Long.MAX_VALUE ms with a positive nano clock, where the deadline wraps and only a comparison by subtraction still finds it ahead. RequestTimeoutsFireBenchmark fires the entry that checks the read timeout one or five times per exchange, each finding a recent read and arming again, so the cost of re-arming shows next to the cost of arming and cancelling. Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5.5 --- .../bench/RequestTimeoutsFireBenchmark.java | 175 ++++++++++++++++++ .../timeout/RequestTimeoutTimerTaskTest.java | 19 +- 2 files changed, 193 insertions(+), 1 deletion(-) create mode 100644 client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsFireBenchmark.java diff --git a/client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsFireBenchmark.java b/client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsFireBenchmark.java new file mode 100644 index 000000000..1bfaff41f --- /dev/null +++ b/client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsFireBenchmark.java @@ -0,0 +1,175 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.bench; + +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timeout; +import io.netty.util.Timer; +import io.netty.util.TimerTask; +import org.asynchttpclient.AsyncCompletionHandler; +import org.asynchttpclient.AsyncHttpClientConfig; +import org.asynchttpclient.DefaultAsyncHttpClient; +import org.asynchttpclient.DefaultAsyncHttpClientConfig; +import org.asynchttpclient.Request; +import org.asynchttpclient.RequestBuilder; +import org.asynchttpclient.Response; +import org.asynchttpclient.channel.ChannelPoolPartitioning; +import org.asynchttpclient.netty.NettyResponseFuture; +import org.asynchttpclient.netty.request.NettyRequestSender; +import org.asynchttpclient.netty.timeout.TimeoutsHolder; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; + +import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +/** + * An exchange that outlives its read window: the {@link RequestTimeoutsArmBenchmark#armAndCancel} lifecycle + * with the entry that checks the read timeout firing {@code fires} times in between, each time finding the + * channel read from within the window and arming again. Read timeout 10 s, request timeout 60 s, on the + * client's {@link HashedWheelTimer}. + *

+ * The benchmark fires the entry itself, on its own thread, rather than waiting out the window: it takes the + * shortest entry the attempt has pending out of the wheel and runs its task. Taking it out is a cancel, which + * a real expiry does not cost the caller; both builds pay it alike, once per fire. + *

+ * Run with {@code -t 1}, {@code -t 8}, {@code -t 32}, with {@code -prof gc}. + */ +@State(Scope.Benchmark) +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.MICROSECONDS) +public class RequestTimeoutsFireBenchmark { + + @Param({"1", "5"}) + public int fires; + + private HashedWheelTimer timer; + private DefaultAsyncHttpClient client; + private AsyncHttpClientConfig config; + private NettyRequestSender requestSender; + private InetSocketAddress remoteAddress; + + @State(Scope.Thread) + public static class Exchange { + NettyResponseFuture future; + ShortestEntryTimer timer; + + @Setup + public void setup(RequestTimeoutsFireBenchmark benchmark) { + Request request = new RequestBuilder().setUrl("http://localhost:8080/").build(); + future = new NettyResponseFuture<>(request, new AsyncCompletionHandler() { + @Override + public Object onCompleted(Response response) { + return null; + } + }, null, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null); + timer = new ShortestEntryTimer(benchmark.timer); + } + } + + /** + * Arms on the shared timer and remembers this thread's shortest pending entry, which is the one that checks + * the read timeout. + */ + static final class ShortestEntryTimer implements Timer { + private final Timer delegate; + private TimerTask task; + private Timeout timeout; + private long delay = Long.MAX_VALUE; + + ShortestEntryTimer(Timer delegate) { + this.delegate = delegate; + } + + @Override + public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) { + Timeout timeout = delegate.newTimeout(task, delay, unit); + if (delay < this.delay) { + this.task = task; + this.timeout = timeout; + this.delay = delay; + } + return timeout; + } + + void fireShortest() throws Exception { + TimerTask task = this.task; + Timeout timeout = this.timeout; + this.task = null; + this.timeout = null; + delay = Long.MAX_VALUE; + timeout.cancel(); + task.run(timeout); + } + + void reset() { + task = null; + timeout = null; + delay = Long.MAX_VALUE; + } + + @Override + public Set stop() { + throw new UnsupportedOperationException(); + } + } + + @Setup + public void setup() throws ReflectiveOperationException { + timer = new HashedWheelTimer(1, TimeUnit.MILLISECONDS, 512); + timer.start(); + config = new DefaultAsyncHttpClientConfig.Builder() + .setNettyTimer(timer) + .setRequestTimeout(Duration.ofMillis(60000)) + .setReadTimeout(Duration.ofMillis(10000)) + .build(); + client = new DefaultAsyncHttpClient(config); + Field field = DefaultAsyncHttpClient.class.getDeclaredField("requestSender"); + field.setAccessible(true); + requestSender = (NettyRequestSender) field.get(client); + remoteAddress = InetSocketAddress.createUnresolved("localhost", 8080); + } + + @TearDown + public void tearDown() throws Exception { + client.close(); + timer.stop(); + } + + @Benchmark + public TimeoutsHolder armFireAndCancel(Exchange exchange) throws Exception { + TimeoutsHolder holder = new TimeoutsHolder(exchange.timer, null, exchange.future, requestSender, config, remoteAddress); + holder.start(); + holder.startReadTimeout(); + for (int i = 0; i < fires; i++) { + exchange.future.touch(); + exchange.timer.fireShortest(); + } + holder.cancel(); + exchange.timer.reset(); + return holder; + } +} diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java index 30dcda12f..f7bd58fdc 100644 --- a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java @@ -338,6 +338,22 @@ public void aWallClockStepBackDoesNotPushTheRequestDeadlineBack() { "the request deadline should stay where the monotonic clock put it"); } + @Test + public void aRequestTimeoutPastWhatTheMonotonicClockCanHoldIsNotTakenForPassed() { + TimeoutsHolder holder = holder(Long.MAX_VALUE, false, false, null, 0); + // Any positive reading plus the longest timeout overflows, which only a comparison by subtraction + // survives. + nanoOrigin = ONE_HOUR_MS; + holder.start(); + holder.startReadTimeout(); + + fire(0, READ_TIMEOUT_MS / 2); + + assertFalse(future.isDone(), "the request deadline is nowhere near"); + assertEquals(2, armed.size()); + assertTrue(holder.isRequestTimeoutArmedEarly()); + } + @Test public void aWallClockStepBackDoesNotPutTheNextReadCheckOutOfReach() { TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); @@ -465,7 +481,8 @@ public void aCompletedExchangeArmsNothingMore() { holder.startReadTimeout(); future.done(); - fire(0, READ_TIMEOUT_MS); + // Inside the read window, where an exchange still running would be re-armed. + fire(0, READ_TIMEOUT_MS / 2); assertEquals(1, armed.size(), "a finished exchange should arm nothing more"); assertTrue(((RequestTimeoutTimerTask) armed.get(0).task).isClaimed()); From 44b7bc3ce4593a99522122e115627919928ffd62 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Sat, 7 Nov 2026 20:35:01 +0100 Subject: [PATCH 11/11] Trim what an early fire costs The firing benchmark showed an exchange that outlives its read window paying more per fire than main's separate read entry: two monotonic clock readings, a CAS on the read state that fails once the read timeout is covered, and a volatile write of a flag that rarely changes. One reading now answers both whether the request deadline has passed and how long to re-arm for, the CAS runs only while the read timeout is still pending, and the flag is written only when it changes. A test pins the single clock reading. Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5.5 --- .../timeout/RequestTimeoutTimerTask.java | 5 ++- .../netty/timeout/TimeoutsHolder.java | 37 ++++++++++--------- .../timeout/RequestTimeoutTimerTaskTest.java | 12 ++++++ 3 files changed, 35 insertions(+), 19 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java index acdd06ea6..80aaba7a0 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java @@ -82,7 +82,8 @@ private boolean checkReadTimeout() { return true; } - if (timeoutsHolder.isRequestDeadlinePassed()) { + long untilDeadline = timeoutsHolder.untilRequestDeadline(); + if (untilDeadline == 0) { return false; } long now = timeoutsHolder.millisTime(); @@ -106,7 +107,7 @@ private boolean checkReadTimeout() { // armed while this task is unclaimed. done.set(false); try { - timeoutsHolder.rearmRequestTimeout(this, now, suspended); + timeoutsHolder.rearmRequestTimeout(this, now, untilDeadline, suspended); } catch (RuntimeException e) { // A timer that rejects the entry, being full or stopped, would leave the exchange with no deadline // at all, so it is failed instead of left to hang. Only while this holder is still the exchange's: diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java index 094fa76d1..8b613ab00 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -250,16 +250,13 @@ boolean isReadTimeoutCovered() { return readState == READ_COVERED; } - boolean isRequestDeadlinePassed() { - return requestDeadlineNanos - nanoClock.getAsLong() <= 0; - } - /** * What is left until the request deadline in whole milliseconds, rounded up so an entry never runs ahead of * it, and floored at zero: a scheduler has no use for a negative delay, and the task has to run either way, - * since running is what fails the exchange. + * since running is what fails the exchange. Zero exactly when the deadline has passed. One clock reading + * answers both whether it has and how long to re-arm for, so an entry that fires early reads it once. */ - private long untilRequestDeadline() { + long untilRequestDeadline() { long left = requestDeadlineNanos - nanoClock.getAsLong(); if (left <= 0) { return 0L; @@ -283,7 +280,7 @@ public void rehomeOn(EventExecutor executor) { if (task == null || cancelled != 0 || task.isClaimed() || !task.cancelArmed()) { return; } - armRequestTimeoutFrom(task, millisClock.getAsLong(), false); + armRequestTimeoutFrom(task, millisClock.getAsLong(), untilRequestDeadline(), false); } public void setResolvedRemoteAddress(InetSocketAddress address) { @@ -313,19 +310,22 @@ private boolean coverReadTimeout() { /** * Re-arms the request timeout after its entry fired ahead of the request deadline and found no timeout due. * - * @param suspended whether response body reads are suspended, which holds off the read timeout for a whole - * window however long ago the channel was last read from + * @param untilDeadline what {@link #untilRequestDeadline} returned when the entry fired + * @param suspended whether response body reads are suspended, which holds off the read timeout for a + * whole window however long ago the channel was last read from */ - void rearmRequestTimeout(RequestTimeoutTimerTask task, long now, boolean suspended) { + void rearmRequestTimeout(RequestTimeoutTimerTask task, long now, long untilDeadline, boolean suspended) { // A read timeout that has not started by now may not start for a long time -- a slow connect or upload // -- and checking for it every window would wake the scheduler for nothing, so it gets its own entry - // when it does start. - READ_STATE.compareAndSet(this, READ_PENDING, READ_SEPARATE); - armRequestTimeoutFrom(task, now, suspended); + // when it does start. Read first: on most fires the read timeout is long covered, and a failed CAS + // costs as much as one that succeeds. + if (readState == READ_PENDING) { + READ_STATE.compareAndSet(this, READ_PENDING, READ_SEPARATE); + } + armRequestTimeoutFrom(task, now, untilDeadline, suspended); } - private void armRequestTimeoutFrom(RequestTimeoutTimerTask task, long now, boolean suspended) { - long untilDeadline = untilRequestDeadline(); + private void armRequestTimeoutFrom(RequestTimeoutTimerTask task, long now, long untilDeadline, boolean suspended) { long untilRead = Long.MAX_VALUE; int state = readState; if (state == READ_PENDING) { @@ -343,8 +343,11 @@ private void armRequestTimeoutFrom(RequestTimeoutTimerTask task, long now, boole } private void armRequestTimeout(RequestTimeoutTimerTask task, long delay, boolean early) { - // Written before arming: on an event loop the entry can run before arm returns. - requestTimeoutArmedEarly = early; + // Written before arming: on an event loop the entry can run before arm returns. Only when it changes, + // since an entry that keeps re-arming early would otherwise pay a volatile write on every fire. + if (requestTimeoutArmedEarly != early) { + requestTimeoutArmedEarly = early; + } arm(task, delay); } diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java index f7bd58fdc..17ad5e359 100644 --- a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java @@ -188,6 +188,18 @@ public void anAttemptWithNothingToReArmDoesNotReadTheMonotonicClock() { assertEquals(0, nanoReads, "only an entry that may be re-armed needs the request deadline on the nano clock"); } + @Test + public void anEarlyFireReadsTheMonotonicClockOnce() { + TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS); + holder.startReadTimeout(); + nanoReads = 0; + + fire(0, READ_TIMEOUT_MS / 2); + + assertEquals(2, armed.size()); + assertEquals(1, nanoReads, "one reading should both check the request deadline and time the next entry"); + } + @Test public void aRequestDeadlineAheadOfTheReadTimeoutTakesTheEntryAlone() { TimeoutsHolder holder = startedHolder(READ_TIMEOUT_MS / 2);