From 59f814a8e6a8765255334787f6c3ada075176d8e Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 17 Aug 2026 20:45:41 +0200 Subject: [PATCH] Arm request timeouts on an event loop A hashed wheel fires on the first tick at or after a deadline, so a deadline near or below the tick duration is rounded up to it, and one timer thread carries every expiry for the whole client. Both hurt short deadlines: a tick is a large fraction of the budget, and a burst of expiries has no headroom to absorb. Measured over 2000 timeouts armed as one burst on Netty 4.2.16, a 20 ms deadline overshot by a mean of 2.7 ms and a p99 of 5 ms on a 5 ms wheel, 1.3/2 ms on a 1 ms wheel, and 0/0 ms scheduled on an event loop, which derives its select timeout from the nearest deadline and so rounds nothing. Add isUseEventLoopTimeouts(), off by default, which arms the request and read timeouts on an event loop instead. On the pooled path the channel is already in hand, so its own loop is used and the timeout expires on the thread that would have to close it. On the connect path there is no channel yet, deliberately, so that the timeout also bounds address resolution and the connect: any loop will do there, since what the wheel costs is a single thread and a rounded-up tick rather than the identity of the thread. Deliberately not a wheel per event loop, which is how the Aerospike client solves this. A wheel arms in O(1) against O(log n) for a deadline queue, but at a few thousand timeouts per loop that is a dozen comparisons, while the quantization it reintroduces costs milliseconds on a 20 ms budget; it also has to be ticked forever, waking every loop even with nothing armed. Aerospike wrote its own wheel because its EventLoop abstracts over NIO, Netty and direct NIO and needed one timer; AHC is Netty-only and gets a per-loop deadline queue for free. Arming allocates nothing beyond what the scheduler needs: the cancellation handle lives on the task, and the existing done flag stands in for the scheduler's already-expired flag, so no per-timeout wrapper is required. Left off by default because the expiry, and therefore whatever the caller chained onto the response future, then runs on an I/O thread. Blocking one stalls every connection it serves. Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5 --- .../AsyncHttpClientConfig.java | 24 ++++ .../DefaultAsyncHttpClientConfig.java | 23 ++++ .../config/AsyncHttpClientConfigDefaults.java | 5 + .../netty/request/NettyRequestSender.java | 34 ++++- .../netty/timeout/TimeoutTimerTask.java | 55 +++++++- .../netty/timeout/TimeoutsHolder.java | 100 +++++++++++--- .../config/ahc-default.properties | 1 + .../asynchttpclient/EventLoopTimeoutTest.java | 126 ++++++++++++++++++ 8 files changed, 343 insertions(+), 25 deletions(-) create mode 100644 client/src/test/java/org/asynchttpclient/EventLoopTimeoutTest.java diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index 7304626083..dedf8ae931 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java @@ -270,6 +270,30 @@ default Duration getFailedIpCooldownPeriod() { return Duration.ofSeconds(10); } + /** + * Whether request and read timeouts are armed on an event loop rather than on {@link #getNettyTimer()}. + *

+ * The timer is a hashed wheel: it fires on the first tick at or after the deadline, so a deadline near or + * below {@link #getHashedWheelTimerTickDuration()} is rounded up to it, and one thread carries every expiry + * for the whole client. An event loop instead schedules by deadline and derives its own select timeout from + * the nearest one, so nothing is rounded up, and the loops share the load rather than funnelling it through + * a single thread. Both effects matter most to short deadlines, where a tick is a large fraction of the + * budget and a burst of expiries has no headroom to absorb. + *

+ * The cost is where the expiry runs. On the timer it runs on the timer thread; on an event loop it runs on + * an I/O thread, and so does whatever the caller chained onto the response future, because that future is + * completed from there. Blocking an I/O thread stalls every connection it serves, so a caller enabling this + * should hand its own work off with {@code handleAsync} or an {@code AsyncHandler} that does the same. + * That is why this is opt-in rather than the default. + *

+ * The connection-pool cleaner stays on the timer either way. + * + * @return {@code true} to arm request and read timeouts on an event loop + */ + default boolean isUseEventLoopTimeouts() { + return false; + } + /** * @return the disableUrlEncodingForBoundRequests */ diff --git a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java index 75aa1bd16a..8ddb0660d8 100644 --- a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java @@ -63,6 +63,7 @@ import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultEnabledProtocols; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultExpiredCookieEvictionDelay; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultFailedIpCooldownEnabled; +import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultUseEventLoopTimeouts; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultFailedIpCooldownPeriod; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultFilterInsecureCipherSuites; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultFollowRedirect; @@ -138,6 +139,7 @@ public class DefaultAsyncHttpClientConfig implements AsyncHttpClientConfig { private final int maxRequestRetry; private final LoadBalance loadBalance; private final boolean failedIpCooldownEnabled; + private final boolean useEventLoopTimeouts; private final Duration failedIpCooldownPeriod; private final boolean disableUrlEncodingForBoundRequests; private final boolean useLaxCookieEncoder; @@ -243,6 +245,7 @@ private DefaultAsyncHttpClientConfig(// http int maxRequestRetry, LoadBalance loadBalance, boolean failedIpCooldownEnabled, + boolean useEventLoopTimeouts, Duration failedIpCooldownPeriod, boolean disableUrlEncodingForBoundRequests, boolean useLaxCookieEncoder, @@ -348,6 +351,7 @@ private DefaultAsyncHttpClientConfig(// http this.maxRequestRetry = maxRequestRetry; this.loadBalance = loadBalance; this.failedIpCooldownEnabled = failedIpCooldownEnabled; + this.useEventLoopTimeouts = useEventLoopTimeouts; this.failedIpCooldownPeriod = failedIpCooldownPeriod; this.disableUrlEncodingForBoundRequests = disableUrlEncodingForBoundRequests; this.useLaxCookieEncoder = useLaxCookieEncoder; @@ -518,6 +522,11 @@ public boolean isFailedIpCooldownEnabled() { return failedIpCooldownEnabled; } + @Override + public boolean isUseEventLoopTimeouts() { + return useEventLoopTimeouts; + } + @Override public Duration getFailedIpCooldownPeriod() { return failedIpCooldownPeriod; @@ -937,6 +946,7 @@ public static class Builder { private int maxRequestRetry = defaultMaxRequestRetry(); private LoadBalance loadBalance = defaultLoadBalance(); private boolean failedIpCooldownEnabled = defaultFailedIpCooldownEnabled(); + private boolean useEventLoopTimeouts = defaultUseEventLoopTimeouts(); private Duration failedIpCooldownPeriod = defaultFailedIpCooldownPeriod(); private boolean disableUrlEncodingForBoundRequests = defaultDisableUrlEncodingForBoundRequests(); private boolean useLaxCookieEncoder = defaultUseLaxCookieEncoder(); @@ -1045,6 +1055,7 @@ public Builder(AsyncHttpClientConfig config) { maxRequestRetry = config.getMaxRequestRetry(); loadBalance = config.getLoadBalance(); failedIpCooldownEnabled = config.isFailedIpCooldownEnabled(); + useEventLoopTimeouts = config.isUseEventLoopTimeouts(); failedIpCooldownPeriod = config.getFailedIpCooldownPeriod(); disableUrlEncodingForBoundRequests = config.isDisableUrlEncodingForBoundRequests(); useLaxCookieEncoder = config.isUseLaxCookieEncoder(); @@ -1244,6 +1255,17 @@ public Builder setFailedIpCooldownEnabled(boolean failedIpCooldownEnabled) { return this; } + /** + * @param useEventLoopTimeouts whether to arm request and read timeouts on an event loop instead of on + * the client's timer; see {@link AsyncHttpClientConfig#isUseEventLoopTimeouts()} + * for the trade-off this makes + * @return this + */ + public Builder setUseEventLoopTimeouts(boolean useEventLoopTimeouts) { + this.useEventLoopTimeouts = useEventLoopTimeouts; + return this; + } + /** * @param failedIpCooldownPeriod how long a failed IP is deprioritized before it is re-probed; * {@code null} resets to the default. Must not be negative; use @@ -1751,6 +1773,7 @@ public DefaultAsyncHttpClientConfig build() { maxRequestRetry, loadBalance, failedIpCooldownEnabled, + useEventLoopTimeouts, failedIpCooldownPeriod, disableUrlEncodingForBoundRequests, useLaxCookieEncoder, diff --git a/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java b/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java index a31fdf2855..56c1ecef74 100644 --- a/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java +++ b/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java @@ -62,6 +62,7 @@ public final class AsyncHttpClientConfigDefaults { public static final String MAX_REQUEST_RETRY_CONFIG = "maxRequestRetry"; public static final String LOAD_BALANCE_CONFIG = "loadBalance"; public static final String FAILED_IP_COOLDOWN_ENABLED_CONFIG = "failedIpCooldownEnabled"; + public static final String USE_EVENT_LOOP_TIMEOUTS_CONFIG = "useEventLoopTimeouts"; public static final String FAILED_IP_COOLDOWN_PERIOD_CONFIG = "failedIpCooldownPeriod"; public static final String DISABLE_URL_ENCODING_FOR_BOUND_REQUESTS_CONFIG = "disableUrlEncodingForBoundRequests"; public static final String USE_LAX_COOKIE_ENCODER_CONFIG = "useLaxCookieEncoder"; @@ -183,6 +184,10 @@ public static boolean defaultFailedIpCooldownEnabled() { return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + FAILED_IP_COOLDOWN_ENABLED_CONFIG); } + public static boolean defaultUseEventLoopTimeouts() { + return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + USE_EVENT_LOOP_TIMEOUTS_CONFIG); + } + public static Duration defaultFailedIpCooldownPeriod() { return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getDuration(ASYNC_CLIENT_CONFIG_ROOT + FAILED_IP_COOLDOWN_PERIOD_CONFIG); } diff --git a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java index af3610164d..a7ad7a88be 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -82,6 +82,7 @@ import org.asynchttpclient.resolver.RequestHostnameResolver; import org.asynchttpclient.uri.Uri; import org.asynchttpclient.ws.WebSocketUpgradeHandler; +import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -404,7 +405,7 @@ private ListenableFuture sendRequestWithOpenChannel(NettyResponseFuture handler, HttpReques private void scheduleRequestTimeout(NettyResponseFuture nettyResponseFuture, InetSocketAddress originalRemoteAddress) { + scheduleRequestTimeout(nettyResponseFuture, originalRemoteAddress, null); + } + + /** + * @param channel the channel the exchange will run on when it is already known, so the timeout can be armed + * on the loop that owns it and expire on the thread that would have to close it. Null on the + * connect path: the timeout is armed before the channel exists, deliberately, so that it also + * bounds address resolution and the connect itself. + */ + private void scheduleRequestTimeout(NettyResponseFuture nettyResponseFuture, + InetSocketAddress originalRemoteAddress, + @Nullable Channel channel) { nettyResponseFuture.touch(); - TimeoutsHolder timeoutsHolder = new TimeoutsHolder(nettyTimer, nettyResponseFuture, this, config, - originalRemoteAddress); + TimeoutsHolder timeoutsHolder = new TimeoutsHolder(nettyTimer, timeoutExecutor(channel), nettyResponseFuture, + this, config, originalRemoteAddress); nettyResponseFuture.setTimeoutsHolder(timeoutsHolder); } + /** + * The event loop to arm an exchange's timeouts on, or null to leave them on the client's timer. Prefers the + * channel's own loop; without a channel any loop will do, since what the wheel costs is a single thread for + * the whole client and a tick the deadline is rounded up to, not the identity of the thread. + */ + private @Nullable EventExecutor timeoutExecutor(@Nullable Channel channel) { + if (!config.isUseEventLoopTimeouts()) { + return null; + } + if (channel != null) { + return channel.eventLoop(); + } + return channelManager.getEventLoopGroup().next(); + } + private static void scheduleReadTimeout(NettyResponseFuture nettyResponseFuture) { TimeoutsHolder timeoutsHolder = nettyResponseFuture.getTimeoutsHolder(); if (timeoutsHolder != null) { 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 b7e678fa84..6bb1f05fb2 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java @@ -15,17 +15,25 @@ */ package org.asynchttpclient.netty.timeout; +import io.netty.util.Timeout; import io.netty.util.TimerTask; import org.asynchttpclient.netty.NettyResponseFuture; import org.asynchttpclient.netty.request.NettyRequestSender; +import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.net.InetSocketAddress; +import java.util.concurrent.Future; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; -public abstract class TimeoutTimerTask implements TimerTask { +/** + * Also a {@link Runnable} so the same task can be armed either on a {@link io.netty.util.Timer} or on an + * event loop, which schedules {@code Runnable}s. Neither subclass reads the {@link Timeout} handed to + * {@link TimerTask#run(Timeout)}, so the two entry points are interchangeable. + */ +public abstract class TimeoutTimerTask implements TimerTask, Runnable { private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutTimerTask.class); @@ -33,6 +41,12 @@ public abstract class TimeoutTimerTask implements TimerTask { protected final NettyRequestSender requestSender; final TimeoutsHolder timeoutsHolder; volatile NettyResponseFuture nettyResponseFuture; + /** + * The scheduled entry this task is armed on: an {@link Timeout} from a {@link io.netty.util.Timer}, or a + * {@link Future} from an event loop. Held here rather than in a wrapper so arming a timeout allocates + * nothing beyond what the scheduler itself needs. + */ + private volatile @Nullable Object armed; TimeoutTimerTask(NettyResponseFuture nettyResponseFuture, NettyRequestSender requestSender, TimeoutsHolder timeoutsHolder) { this.nettyResponseFuture = nettyResponseFuture; @@ -40,6 +54,45 @@ public abstract class TimeoutTimerTask implements TimerTask { this.timeoutsHolder = timeoutsHolder; } + @Override + public void run() { + try { + run(null); + } catch (Exception e) { + // TimerTask#run is declared to throw, and on this entry point the caller is an event loop, where an + // escaping exception would be swallowed into Netty's own handling. Neither task here throws, so this + // only matters for a subclass outside the library. + LOGGER.warn("Timeout task failed", e); + } + } + + void armedOn(Object handle) { + armed = handle; + } + + /** + * 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. + */ + void cancelArmed() { + Object handle = armed; + armed = null; + if (handle instanceof Timeout) { + ((Timeout) handle).cancel(); + } else if (handle instanceof Future) { + ((Future) handle).cancel(false); + } + } + + /** + * Whether this task has been claimed, either by firing or by {@link #clean()}. Stands in for the + * scheduler's own already-expired flag, which the two schedulers spell differently, and is if anything the + * more precise of the two: it flips when {@code run} is entered rather than when the entry is marked. + */ + boolean isClaimed() { + return done.get(); + } + void expire(String message, long time) { LOGGER.debug("{} for {} after {} ms", message, nettyResponseFuture, time); requestSender.abort(nettyResponseFuture.channel(), nettyResponseFuture, new TimeoutException(message)); 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 93f6b26a26..240dd2572a 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -15,35 +15,66 @@ */ package org.asynchttpclient.netty.timeout; -import io.netty.util.Timeout; import io.netty.util.Timer; -import io.netty.util.TimerTask; +import io.netty.util.concurrent.EventExecutor; import org.asynchttpclient.AsyncHttpClientConfig; import org.asynchttpclient.Request; import org.asynchttpclient.netty.NettyResponseFuture; import org.asynchttpclient.netty.request.NettyRequestSender; +import org.jetbrains.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.net.InetSocketAddress; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime; +/** + * The request and read timeouts of one exchange. + *

+ * Timeouts are armed either on the client's {@link Timer} or, when an {@link EventExecutor} is supplied, on + * that event loop. The two differ in more than which thread runs the task. A wheel fires on the first tick at + * or after the deadline, so a deadline near or below the tick duration is rounded up, and one thread carries + * every expiry for the whole client. An event loop schedules by deadline and derives its select timeout from + * the nearest one, so nothing is rounded, and the loops share the load. See + * {@link AsyncHttpClientConfig#isUseEventLoopTimeouts()} for what that costs. + */ public class TimeoutsHolder { - private final Timeout requestTimeout; + private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutsHolder.class); + private final AtomicBoolean cancelled = new AtomicBoolean(); private final Timer nettyTimer; + private final @Nullable EventExecutor eventExecutor; private final NettyRequestSender requestSender; private final long requestTimeoutMillisTime; private final long readTimeoutValue; - private volatile Timeout readTimeout; + private final @Nullable RequestTimeoutTimerTask requestTimeoutTask; + // Whether the request timeout was actually armed. Distinct from requestTimeoutTask being non-null: the + // task exists but is left unarmed when there is nothing to arm it on, and the read timeout is then free to + // run to its own deadline rather than assuming a request timeout will outrun it. + private final boolean requestTimeoutArmed; + private volatile @Nullable ReadTimeoutTimerTask readTimeoutTask; private final NettyResponseFuture nettyResponseFuture; private volatile InetSocketAddress remoteAddress; public TimeoutsHolder(Timer nettyTimer, NettyResponseFuture nettyResponseFuture, NettyRequestSender requestSender, AsyncHttpClientConfig config, InetSocketAddress originalRemoteAddress) { + this(nettyTimer, null, nettyResponseFuture, requestSender, config, originalRemoteAddress); + } + + /** + * @param eventExecutor the event loop to arm the timeouts on, or {@code null} to arm them on + * {@code nettyTimer}. Pass the loop that owns the exchange's channel when it is known, + * so the timeout fires on the thread that will have to close it. + */ + public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, NettyResponseFuture nettyResponseFuture, + NettyRequestSender requestSender, AsyncHttpClientConfig config, InetSocketAddress originalRemoteAddress) { this.nettyTimer = nettyTimer; + this.eventExecutor = eventExecutor; this.nettyResponseFuture = nettyResponseFuture; this.requestSender = requestSender; remoteAddress = originalRemoteAddress; @@ -60,10 +91,12 @@ public TimeoutsHolder(Timer nettyTimer, NettyResponseFuture nettyResponseFutu if (requestTimeoutInMs > -1) { requestTimeoutMillisTime = unpreciseMillisTime() + requestTimeoutInMs; - requestTimeout = newTimeout(new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs), requestTimeoutInMs); + requestTimeoutTask = new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs); + requestTimeoutArmed = arm(requestTimeoutTask, requestTimeoutInMs); } else { requestTimeoutMillisTime = -1L; - requestTimeout = null; + requestTimeoutTask = null; + requestTimeoutArmed = false; } } @@ -81,14 +114,16 @@ public void startReadTimeout() { } } - void startReadTimeout(ReadTimeoutTimerTask task) { - if (requestTimeout == null || !requestTimeout.isExpired() && readTimeoutValue < requestTimeoutMillisTime - unpreciseMillisTime()) { + void startReadTimeout(@Nullable ReadTimeoutTimerTask task) { + if (!requestTimeoutArmed + || !requestTimeoutTask.isClaimed() && readTimeoutValue < requestTimeoutMillisTime - unpreciseMillisTime()) { // only schedule a new readTimeout if the requestTimeout doesn't happen first if (task == null) { // first call triggered from outside (else is read timeout is re-scheduling itself) task = new ReadTimeoutTimerTask(nettyResponseFuture, requestSender, this, readTimeoutValue); } - readTimeout = newTimeout(task, readTimeoutValue); + readTimeoutTask = task; + arm(task, readTimeoutValue); } else if (task != null) { // read timeout couldn't re-scheduling itself, clean up @@ -98,24 +133,47 @@ void startReadTimeout(ReadTimeoutTimerTask task) { public void cancel() { if (cancelled.compareAndSet(false, true)) { - if (requestTimeout != null) { - requestTimeout.cancel(); - ((TimeoutTimerTask) requestTimeout.task()).clean(); - } - if (readTimeout != null) { - readTimeout.cancel(); - ((TimeoutTimerTask) readTimeout.task()).clean(); - } + release(requestTimeoutTask); + release(readTimeoutTask); } } - private Timeout newTimeout(TimerTask task, long delay) { + private static void release(@Nullable TimeoutTimerTask task) { + if (task != null) { + task.cancelArmed(); + task.clean(); + } + } + + /** + * Arms {@code task} to run after {@code delay} milliseconds, recording the scheduled entry on the task so it + * can cancel itself later. + * + * @return whether the task was armed. It is not when the client is shutting down, in which case there is no + * timeout to deliver anyway + */ + private boolean arm(TimeoutTimerTask task, long delay) { // requestSender or nettyTimer might be null in unit tests or in some edge // cases where a channel's remote address wasn't available. In such cases // avoid scheduling any timeouts rather than throwing a NPE. - if (requestSender == null || nettyTimer == null || requestSender.isClosed()) { - return null; + if (requestSender == null || requestSender.isClosed()) { + return false; + } + if (eventExecutor != null && !eventExecutor.isShuttingDown()) { + try { + task.armedOn(eventExecutor.schedule(task, delay, TimeUnit.MILLISECONDS)); + return true; + } catch (RejectedExecutionException e) { + // The loop began shutting down between the check above and here. Losing the timeout entirely + // would leave the exchange with nothing to end it, so fall through to the timer, which the + // client keeps running until it is itself closed. + LOGGER.debug("Event loop rejected a timeout, falling back to the timer", e); + } + } + if (nettyTimer == null) { + return false; } - return nettyTimer.newTimeout(task, delay, TimeUnit.MILLISECONDS); + task.armedOn(nettyTimer.newTimeout(task, delay, TimeUnit.MILLISECONDS)); + return true; } } diff --git a/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties b/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties index 6bf4e0f7b2..f7ee6925b9 100644 --- a/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties +++ b/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties @@ -26,6 +26,7 @@ org.asynchttpclient.keepAlive=true org.asynchttpclient.maxRequestRetry=5 org.asynchttpclient.loadBalance=DEFAULT org.asynchttpclient.failedIpCooldownEnabled=true +org.asynchttpclient.useEventLoopTimeouts=false org.asynchttpclient.failedIpCooldownPeriod=PT10S org.asynchttpclient.disableUrlEncodingForBoundRequests=false org.asynchttpclient.useLaxCookieEncoder=false diff --git a/client/src/test/java/org/asynchttpclient/EventLoopTimeoutTest.java b/client/src/test/java/org/asynchttpclient/EventLoopTimeoutTest.java new file mode 100644 index 0000000000..8ed3594d3c --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/EventLoopTimeoutTest.java @@ -0,0 +1,126 @@ +/* + * 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; + +import io.netty.handler.codec.http.DefaultHttpHeaders; +import io.netty.handler.codec.http.HttpHeaders; +import org.asynchttpclient.testserver.HttpServer; +import org.asynchttpclient.testserver.HttpTest; +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.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; + +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Where a request timeout is delivered from, which is what + * {@link AsyncHttpClientConfig#isUseEventLoopTimeouts()} changes. Off, every expiry in the client runs on the + * timer's single thread; on, it runs on an event loop. The thread a timeout is delivered on is observable + * through {@link AsyncHandler#onThrowable}, so these assert the switch rather than its side effects. + */ +public class EventLoopTimeoutTest extends HttpTest { + + private static final String IO_THREAD_POOL = "ahc-timeout-test"; + // Netty derives the timer's thread names from this, so a timer thread is the one carrying "timer". + private static final String TIMER_MARKER = "timer"; + + private HttpServer server; + + @BeforeEach + public void start() throws Throwable { + server = new HttpServer(); + server.start(); + } + + @AfterEach + public void stop() throws Throwable { + server.close(); + } + + @Test + public void byDefaultTheTimeoutIsDeliveredFromTheTimerThread() throws Throwable { + String thread = threadDeliveringRequestTimeout(false); + + assertTrue(thread.contains(TIMER_MARKER), + "expected the timer thread by default, got " + thread); + } + + @Test + public void withEventLoopTimeoutsTheTimeoutIsDeliveredFromAnEventLoop() throws Throwable { + String thread = threadDeliveringRequestTimeout(true); + + assertFalse(thread.contains(TIMER_MARKER), + "expected an event loop, not the timer thread, got " + thread); + assertTrue(thread.contains(IO_THREAD_POOL), + "expected one of the client's I/O threads, got " + thread); + } + + /** + * Runs one request against an endpoint that answers well after the request timeout, and returns the name of + * the thread {@code onThrowable} was called on. + */ + private String threadDeliveringRequestTimeout(boolean useEventLoopTimeouts) throws Throwable { + AtomicReference thread = new AtomicReference<>(); + AtomicReference cause = new AtomicReference<>(); + CountDownLatch aborted = new CountDownLatch(1); + + DefaultAsyncHttpClientConfig.Builder builder = config() + .setThreadPoolName(IO_THREAD_POOL) + .setRequestTimeout(Duration.ofMillis(200)) + .setUseEventLoopTimeouts(useEventLoopTimeouts); + + withClient(builder).run(client -> withServer(server).run(server -> { + HttpHeaders headers = new DefaultHttpHeaders(); + headers.add("X-Delay", 5_000); + server.enqueueEcho(); + + client.prepareGet(server.getHttpUrl() + "/foo/bar").setHeaders(headers) + .execute(new AsyncCompletionHandler() { + @Override + public Void onCompleted(Response response) { + aborted.countDown(); + return null; + } + + @Override + public void onThrowable(Throwable t) { + thread.set(Thread.currentThread().getName()); + cause.set(t); + aborted.countDown(); + } + }); + + assertTrue(aborted.await(30, TimeUnit.SECONDS), "the request neither completed nor timed out"); + })); + + assertNotNull(cause.get(), "expected the request to be aborted"); + assertEquals(TimeoutException.class, cause.get().getClass(), + "expected a request timeout, got " + cause.get()); + String name = thread.get(); + assertNotNull(name, "onThrowable was not called"); + return name; + } +}