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 1/2] 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; + } +} From 1931113d3f5822f34407753dccbdbddb81f9823b Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Fri, 21 Aug 2026 10:32:33 +0200 Subject: [PATCH 2/2] Home an exchange's timeouts on its own loop Review feedback on the commit before this one, which claimed that without a channel any loop would do because what a wheel costs is a single thread and a rounded-up tick rather than the identity of the thread. That was wrong in a way the wheel had been hiding. The loop came from EventLoopGroup#next(), which is almost never the loop the channel ends up on: initAndRegister calls next() again, so for an N loop group the two agreed about one time in N. Every completion then cancelled an entry on a foreign loop, and until the original deadline that entry sat in a queue whose loop it would wake for a request that had long finished; the read timeout re-armed across loops for the same reason. Drawing from the chooser only to pick a timeout thread also advanced the counter that assigns channels to loops, so a fixed number of draws per request could settle registrations onto a subset of them. The loop is now only ever the channel's own. The pooled path has the channel in hand. The connect path arms on the timer, as it did before this branch, because the timeout has to bound address resolution and the connect itself; NettyConnectListener then moves it onto the loop once there is a channel, next to the attachChannel that publishes it on the future. That listener already runs on the channel's loop, so the move costs a same-thread schedule and no wakeup. The holder keeps the switch itself rather than making every caller consult the config, which also spares the listener a dependency on the request sender it does not otherwise need there. Arming has also left the TimeoutsHolder constructor. The task holds the holder and can run the moment it is armed, and an event loop does not round a short deadline up to the next tick, so the expiry could reach a holder whose fields were not yet frozen and a future that had not yet been handed it. On the pooled path it could also reach a future with no channel attached, abort with null, and leave the pooled socket open. The caller now publishes the holder and attaches the channel first and calls start() last. Two smaller races: arm records its handle after scheduling, so an exchange that finished in that window left an entry nobody would ever cancel, cancel() being one shot; arm now re-checks the flag afterwards. And cancelArmed did not catch what arm catches, so a late cancel on a closing client threw RejectedExecutionException out of ListenableFuture#cancel, which had never thrown before. The rest is what the review asked for and worth no argument: two typed handles instead of an Object and instanceof, so a scheduler changing its return type is a compile error rather than a cancellation that silently stops working; requestTimeoutArmed dropped for a null test on the task, which is the shape the code had before; the throws clause off run(Timeout), since the package private constructor makes the subclass it defended against impossible; the rationale in isUseEventLoopTimeouts() alone with the other three copies linking to it; and the new option in the timeouts group everywhere rather than between the two failedIpCooldown entries. Dropping that throws clause is the one thing here revapi objects to, and the only way to keep the dead catch out. It is scoped to the single method: the change is binary compatible, and the only source-level effect would be on code catching a narrower checked exception around a direct run(Timeout) call, which nothing outside the library does and no outside subclass can even reach. The tests asserted on substrings of thread names, which a pool name containing "timer" or a configured thread factory would have broken with no bug present, and only ever exercised the no-channel branch. They now hand the config their own Timer and EventLoopGroup and assert against those: the timer's own thread by identity, and for the event loop cases that the expiry arrived on the loop of the channel the handler was told about. A pooled exchange and a read timeout are covered as well. Claude Code on behalf of Pavel Ptashyts Co-Authored-By: Claude Opus 5 --- .../AsyncHttpClientConfig.java | 52 ++-- .../DefaultAsyncHttpClientConfig.java | 46 ++-- .../config/AsyncHttpClientConfigDefaults.java | 10 +- .../netty/channel/NettyConnectListener.java | 5 + .../netty/request/NettyRequestSender.java | 32 +-- .../netty/timeout/TimeoutTimerTask.java | 71 ++++-- .../netty/timeout/TimeoutsHolder.java | 101 +++++--- .../config/ahc-default.properties | 2 +- .../asynchttpclient/EventLoopTimeoutTest.java | 222 +++++++++++++----- pom.xml | 6 + 10 files changed, 365 insertions(+), 182 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index dedf8ae931..b29ce5306a 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java @@ -114,6 +114,34 @@ public interface AsyncHttpClientConfig { */ Duration getRequestTimeout(); + /** + * 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 loop is always the one that owns the exchange's channel. Until there is a channel -- while an address + * is being resolved and a connection made -- the timer carries the timeout, and the exchange moves it onto + * the loop once the connection succeeds. + *

+ * 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; + } + /** * Is HTTP redirect enabled * @@ -270,30 +298,6 @@ 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 8ddb0660d8..a1eed3cc97 100644 --- a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java @@ -63,7 +63,6 @@ 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; @@ -98,6 +97,7 @@ import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultStrict302Handling; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultTcpNoDelay; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultThreadPoolName; +import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultUseEventLoopTimeouts; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultUseInsecureTrustManager; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultUseLaxCookieEncoder; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultUseNativeTransport; @@ -139,7 +139,6 @@ 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; @@ -159,6 +158,7 @@ public class DefaultAsyncHttpClientConfig implements AsyncHttpClientConfig { private final Duration connectTimeout; private final Duration requestTimeout; private final Duration readTimeout; + private final boolean useEventLoopTimeouts; private final Duration shutdownQuietPeriod; private final Duration shutdownTimeout; @@ -245,7 +245,6 @@ private DefaultAsyncHttpClientConfig(// http int maxRequestRetry, LoadBalance loadBalance, boolean failedIpCooldownEnabled, - boolean useEventLoopTimeouts, Duration failedIpCooldownPeriod, boolean disableUrlEncodingForBoundRequests, boolean useLaxCookieEncoder, @@ -261,6 +260,7 @@ private DefaultAsyncHttpClientConfig(// http Duration connectTimeout, Duration requestTimeout, Duration readTimeout, + boolean useEventLoopTimeouts, Duration shutdownQuietPeriod, Duration shutdownTimeout, @@ -351,7 +351,6 @@ private DefaultAsyncHttpClientConfig(// http this.maxRequestRetry = maxRequestRetry; this.loadBalance = loadBalance; this.failedIpCooldownEnabled = failedIpCooldownEnabled; - this.useEventLoopTimeouts = useEventLoopTimeouts; this.failedIpCooldownPeriod = failedIpCooldownPeriod; this.disableUrlEncodingForBoundRequests = disableUrlEncodingForBoundRequests; this.useLaxCookieEncoder = useLaxCookieEncoder; @@ -371,6 +370,7 @@ private DefaultAsyncHttpClientConfig(// http this.connectTimeout = connectTimeout; this.requestTimeout = requestTimeout; this.readTimeout = readTimeout; + this.useEventLoopTimeouts = useEventLoopTimeouts; this.shutdownQuietPeriod = shutdownQuietPeriod; this.shutdownTimeout = shutdownTimeout; @@ -522,11 +522,6 @@ public boolean isFailedIpCooldownEnabled() { return failedIpCooldownEnabled; } - @Override - public boolean isUseEventLoopTimeouts() { - return useEventLoopTimeouts; - } - @Override public Duration getFailedIpCooldownPeriod() { return failedIpCooldownPeriod; @@ -594,6 +589,11 @@ public Duration getReadTimeout() { return readTimeout; } + @Override + public boolean isUseEventLoopTimeouts() { + return useEventLoopTimeouts; + } + @Override public Duration getShutdownQuietPeriod() { return shutdownQuietPeriod; @@ -946,7 +946,6 @@ 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(); @@ -968,6 +967,7 @@ public static class Builder { private Duration connectTimeout = defaultConnectTimeout(); private Duration requestTimeout = defaultRequestTimeout(); private Duration readTimeout = defaultReadTimeout(); + private boolean useEventLoopTimeouts = defaultUseEventLoopTimeouts(); private Duration shutdownQuietPeriod = defaultShutdownQuietPeriod(); private Duration shutdownTimeout = defaultShutdownTimeout(); @@ -1055,7 +1055,6 @@ 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(); @@ -1075,6 +1074,7 @@ public Builder(AsyncHttpClientConfig config) { connectTimeout = config.getConnectTimeout(); requestTimeout = config.getRequestTimeout(); readTimeout = config.getReadTimeout(); + useEventLoopTimeouts = config.isUseEventLoopTimeouts(); shutdownQuietPeriod = config.getShutdownQuietPeriod(); shutdownTimeout = config.getShutdownTimeout(); @@ -1255,17 +1255,6 @@ 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 @@ -1377,6 +1366,17 @@ public Builder setReadTimeout(Duration readTimeout) { 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; + } + public Builder setShutdownQuietPeriod(Duration shutdownQuietPeriod) { this.shutdownQuietPeriod = shutdownQuietPeriod; return this; @@ -1773,7 +1773,6 @@ public DefaultAsyncHttpClientConfig build() { maxRequestRetry, loadBalance, failedIpCooldownEnabled, - useEventLoopTimeouts, failedIpCooldownPeriod, disableUrlEncodingForBoundRequests, useLaxCookieEncoder, @@ -1787,6 +1786,7 @@ public DefaultAsyncHttpClientConfig build() { connectTimeout, requestTimeout, readTimeout, + useEventLoopTimeouts, shutdownQuietPeriod, shutdownTimeout, keepAlive, diff --git a/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java b/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java index 56c1ecef74..50fcd723aa 100644 --- a/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java +++ b/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java @@ -41,6 +41,7 @@ public final class AsyncHttpClientConfigDefaults { public static final String CONNECTION_POOL_CLEANER_PERIOD_CONFIG = "connectionPoolCleanerPeriod"; public static final String READ_TIMEOUT_CONFIG = "readTimeout"; public static final String REQUEST_TIMEOUT_CONFIG = "requestTimeout"; + public static final String USE_EVENT_LOOP_TIMEOUTS_CONFIG = "useEventLoopTimeouts"; public static final String CONNECTION_TTL_CONFIG = "connectionTtl"; public static final String FOLLOW_REDIRECT_CONFIG = "followRedirect"; public static final String MAX_REDIRECTS_CONFIG = "maxRedirects"; @@ -62,7 +63,6 @@ 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"; @@ -155,6 +155,10 @@ public static Duration defaultRequestTimeout() { return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getDuration(ASYNC_CLIENT_CONFIG_ROOT + REQUEST_TIMEOUT_CONFIG); } + public static boolean defaultUseEventLoopTimeouts() { + return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + USE_EVENT_LOOP_TIMEOUTS_CONFIG); + } + public static Duration defaultConnectionTtl() { return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getDuration(ASYNC_CLIENT_CONFIG_ROOT + CONNECTION_TTL_CONFIG); } @@ -184,10 +188,6 @@ 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/channel/NettyConnectListener.java b/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java index 049921c13f..cc03f3407c 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java @@ -124,6 +124,11 @@ public void onSuccess(Channel channel, InetSocketAddress remoteAddress) { // mid-handshake could not close the socket, stranding it until handshakeTimeout (issue #2189). future.attachChannel(channel, false); + // The timeouts were armed before there was a channel to arm them on; hand them the one the exchange + // ended up with. This listener runs on that channel's own loop, so the move needs no wakeup, and from + // here on an expiry runs on the thread that would have to close the socket. + timeoutsHolder.rehomeOn(channel.eventLoop()); + Request request = future.getTargetRequest(); Uri uri = request.getUri(); // don't set a null resolved address - if the remoteAddress is null we keep 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 a7ad7a88be..41dffa515f 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -402,15 +402,17 @@ private ListenableFuture sendRequestWithOpenChannel(NettyResponseFuture nettyResponseFuture, /** * @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. + * on the loop that owns 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, and + * {@code TimeoutsHolder#rehomeOn} moves it onto the loop once there is one. */ private void scheduleRequestTimeout(NettyResponseFuture nettyResponseFuture, InetSocketAddress originalRemoteAddress, @@ -1097,21 +1099,19 @@ private void scheduleRequestTimeout(NettyResponseFuture nettyResponseFuture, TimeoutsHolder timeoutsHolder = new TimeoutsHolder(nettyTimer, timeoutExecutor(channel), nettyResponseFuture, this, config, originalRemoteAddress); nettyResponseFuture.setTimeoutsHolder(timeoutsHolder); + // Only now that the future can be reached from the holder and the channel from the future, since either + // may be needed by an expiry that lands immediately; see TimeoutsHolder#start. + timeoutsHolder.start(); } /** - * 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. + * The loop to arm an exchange's timeouts on, or null to leave them on the client's timer. Only ever the + * exchange's own channel's loop: any other loop would be woken by an entry it has no interest in, and the + * group's chooser hands out channels from the same counter, so drawing from it here would shift which loops + * connections land on. */ private @Nullable EventExecutor timeoutExecutor(@Nullable Channel channel) { - if (!config.isUseEventLoopTimeouts()) { - return null; - } - if (channel != null) { - return channel.eventLoop(); - } - return channelManager.getEventLoopGroup().next(); + return config.isUseEventLoopTimeouts() && channel != null ? channel.eventLoop() : null; } private static void scheduleReadTimeout(NettyResponseFuture nettyResponseFuture) { 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 6bb1f05fb2..8f4cc95e17 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java @@ -17,6 +17,7 @@ import io.netty.util.Timeout; import io.netty.util.TimerTask; +import io.netty.util.concurrent.ScheduledFuture; import org.asynchttpclient.netty.NettyResponseFuture; import org.asynchttpclient.netty.request.NettyRequestSender; import org.jetbrains.annotations.Nullable; @@ -24,14 +25,15 @@ import org.slf4j.LoggerFactory; import java.net.InetSocketAddress; -import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; /** * 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. + * {@link TimerTask#run(Timeout)}, so the two entry points are interchangeable. Which one an exchange uses is + * {@link org.asynchttpclient.AsyncHttpClientConfig#isUseEventLoopTimeouts()}. */ public abstract class TimeoutTimerTask implements TimerTask, Runnable { @@ -41,12 +43,11 @@ public abstract class TimeoutTimerTask implements TimerTask, Runnable { 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; + // 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; TimeoutTimerTask(NettyResponseFuture nettyResponseFuture, NettyRequestSender requestSender, TimeoutsHolder timeoutsHolder) { this.nettyResponseFuture = nettyResponseFuture; @@ -54,34 +55,54 @@ public abstract class TimeoutTimerTask implements TimerTask, Runnable { this.timeoutsHolder = timeoutsHolder; } + /** + * Narrows {@link TimerTask#run(Timeout)} to not throw, so that {@link #run()} can call it with nothing to + * catch. Neither subclass throws, and no other can exist: the only constructor is package private. + */ + @Override + public abstract void run(Timeout timeout); + @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); - } + // The argument is the timer's handle on this task and nothing reads it, so an event loop, which + // schedules a Runnable and has no such handle, enters through the same body. + run(null); + } + + void armedOn(Timeout handle) { + timerHandle = handle; } - void armedOn(Object handle) { - armed = handle; + void armedOn(ScheduledFuture handle) { + loopHandle = 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. + * + * @return whether an entry was taken back out of its scheduler before it could run */ - void cancelArmed() { - Object handle = armed; - armed = null; - if (handle instanceof Timeout) { - ((Timeout) handle).cancel(); - } else if (handle instanceof Future) { - ((Future) handle).cancel(false); + boolean cancelArmed() { + Timeout timer = timerHandle; + if (timer != null) { + timerHandle = null; + return timer.cancel(); + } + ScheduledFuture scheduled = loopHandle; + if (scheduled != null) { + loopHandle = null; + try { + return scheduled.cancel(false); + } catch (RejectedExecutionException e) { + // Cancelling from off the loop enqueues the removal, which a loop that is already shutting down + // rejects. The entry dies with the loop either way, and this runs under + // ListenableFuture#cancel, which has never thrown for a client that is closing. + LOGGER.debug("Event loop rejected a timeout cancellation", e); + return false; + } } + return false; } /** 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 240dd2572a..10d4e96aa7 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -33,14 +33,9 @@ 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. + * The request and read timeouts of one exchange, armed either on the client's {@link Timer} or on the event + * loop of the channel the exchange runs on. What the two differ in, and why the choice is the caller's, is + * {@link AsyncHttpClientConfig#isUseEventLoopTimeouts()}. */ public class TimeoutsHolder { @@ -48,15 +43,12 @@ public class TimeoutsHolder { private final AtomicBoolean cancelled = new AtomicBoolean(); private final Timer nettyTimer; - private final @Nullable EventExecutor eventExecutor; + private volatile @Nullable EventExecutor eventExecutor; private final NettyRequestSender requestSender; private final long requestTimeoutMillisTime; private final long readTimeoutValue; + private final boolean useEventLoopTimeouts; 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; @@ -67,9 +59,11 @@ public TimeoutsHolder(Timer nettyTimer, NettyResponseFuture nettyResponseFutu } /** - * @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. + * @param eventExecutor the loop of the channel this exchange will run on, or {@code null} to arm the + * timeouts on {@code nettyTimer} instead. Only ever a channel's own loop, so that an + * expiry runs on the thread that would have to close the socket and cancelling one on + * completion touches no other loop's queue. Null until a channel exists; + * {@link #rehomeOn} moves the timeouts once one does. */ public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, NettyResponseFuture nettyResponseFuture, NettyRequestSender requestSender, AsyncHttpClientConfig config, InetSocketAddress originalRemoteAddress) { @@ -77,6 +71,7 @@ public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, N this.eventExecutor = eventExecutor; this.nettyResponseFuture = nettyResponseFuture; this.requestSender = requestSender; + useEventLoopTimeouts = config.isUseEventLoopTimeouts(); remoteAddress = originalRemoteAddress; final Request targetRequest = nettyResponseFuture.getTargetRequest(); @@ -92,14 +87,43 @@ public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, N if (requestTimeoutInMs > -1) { requestTimeoutMillisTime = unpreciseMillisTime() + requestTimeoutInMs; requestTimeoutTask = new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs); - requestTimeoutArmed = arm(requestTimeoutTask, requestTimeoutInMs); } else { requestTimeoutMillisTime = -1L; requestTimeoutTask = null; - requestTimeoutArmed = false; } } + /** + * Arms the request timeout, which the constructor deliberately leaves undone. The task holds this holder and + * can run the moment it is armed, and on an event loop nothing rounds a short deadline up to the next tick, + * so arming from the constructor let it run before its own fields were frozen, before the future had been + * handed the holder, and on the pooled path before the channel had been attached to the future -- an expiry + * that then had no channel to close. The caller does all three first and arms last. + */ + public void start() { + if (requestTimeoutTask != null) { + arm(requestTimeoutTask, remainingRequestTimeout()); + } + } + + /** + * 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 + * resolution and the connect as well -- so there the loop is only known once the connection succeeds. A + * no-op when the timeouts belong on the timer, or once the request timeout has fired or been cancelled. + */ + public void rehomeOn(EventExecutor executor) { + if (!useEventLoopTimeouts) { + return; + } + eventExecutor = executor; + RequestTimeoutTimerTask task = requestTimeoutTask; + if (task == null || cancelled.get() || task.isClaimed() || !task.cancelArmed()) { + return; + } + arm(task, remainingRequestTimeout()); + } + public void setResolvedRemoteAddress(InetSocketAddress address) { remoteAddress = address; } @@ -115,7 +139,7 @@ public void startReadTimeout() { } void startReadTimeout(@Nullable ReadTimeoutTimerTask task) { - if (!requestTimeoutArmed + if (requestTimeoutTask == null || !requestTimeoutTask.isClaimed() && readTimeoutValue < requestTimeoutMillisTime - unpreciseMillisTime()) { // only schedule a new readTimeout if the requestTimeout doesn't happen first if (task == null) { @@ -145,24 +169,30 @@ private static void release(@Nullable TimeoutTimerTask task) { } } + private long remainingRequestTimeout() { + // A deadline already behind us is armed at zero rather than negative, so the task still runs and still + // cancels its read-timeout sibling, which is bookkeeping only it does. + return Math.max(requestTimeoutMillisTime - unpreciseMillisTime(), 0L); + } + /** * 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 + * can cancel itself later. Leaves it unarmed when the client is shutting down, in which case there is no + * timeout to deliver anyway. */ - private boolean arm(TimeoutTimerTask task, long delay) { + private void 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 || requestSender.isClosed()) { - return false; + return; } - if (eventExecutor != null && !eventExecutor.isShuttingDown()) { + EventExecutor executor = eventExecutor; + if (executor != null && !executor.isShuttingDown()) { try { - task.armedOn(eventExecutor.schedule(task, delay, TimeUnit.MILLISECONDS)); - return true; + task.armedOn(executor.schedule(task, delay, TimeUnit.MILLISECONDS)); + cancelIfRaced(task); + return; } 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 @@ -171,9 +201,20 @@ private boolean arm(TimeoutTimerTask task, long delay) { } } if (nettyTimer == null) { - return false; + return; } task.armedOn(nettyTimer.newTimeout(task, delay, TimeUnit.MILLISECONDS)); - return true; + cancelIfRaced(task); + } + + /** + * Takes a just-armed entry back out of its scheduler when the exchange finished while it was being armed. + * {@link #cancel} is one shot, so a handle recorded after it ran is one nobody would ever cancel: the entry + * 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()) { + release(task); + } } } 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 f7ee6925b9..34fb663803 100644 --- a/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties +++ b/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties @@ -7,6 +7,7 @@ org.asynchttpclient.pooledConnectionIdleTimeout=PT1M org.asynchttpclient.connectionPoolCleanerPeriod=PT0.1S org.asynchttpclient.readTimeout=PT1M org.asynchttpclient.requestTimeout=PT1M +org.asynchttpclient.useEventLoopTimeouts=false org.asynchttpclient.connectionTtl=-PT0.001S org.asynchttpclient.followRedirect=false org.asynchttpclient.maxRedirects=5 @@ -26,7 +27,6 @@ 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 index 8ed3594d3c..52d3bc5a4d 100644 --- a/client/src/test/java/org/asynchttpclient/EventLoopTimeoutTest.java +++ b/client/src/test/java/org/asynchttpclient/EventLoopTimeoutTest.java @@ -15,14 +15,18 @@ */ package org.asynchttpclient; -import io.netty.handler.codec.http.DefaultHttpHeaders; -import io.netty.handler.codec.http.HttpHeaders; +import io.netty.channel.Channel; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.util.HashedWheelTimer; +import io.netty.util.concurrent.DefaultThreadFactory; 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.net.InetSocketAddress; import java.time.Duration; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -31,96 +35,198 @@ 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.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; 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. + * Which scheduler an exchange's timeouts are armed on, which is what + * {@link AsyncHttpClientConfig#isUseEventLoopTimeouts()} chooses. Off, every expiry in the client is delivered + * from the timer's one thread; on, it is delivered from the event loop of the channel the exchange runs on. + *

+ * The scheduler an expiry came from is observable through the thread {@link AsyncHandler#onThrowable} is called + * on, so these assert against the timer and the loops themselves rather than against thread names: a name is a + * property of whichever thread factory the config happens to carry, the loop that owns a channel is not. */ 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 static final Duration SHORT_TIMEOUT = Duration.ofMillis(200); private HttpServer server; + private EventLoopGroup eventLoopGroup; + private HashedWheelTimer timer; + private final AtomicReference timerThread = new AtomicReference<>(); + // Released before the server is closed, so a request left hanging on purpose never delays teardown. + private final CountDownLatch released = new CountDownLatch(1); @BeforeEach public void start() throws Throwable { server = new HttpServer(); server.start(); + eventLoopGroup = new NioEventLoopGroup(2, new DefaultThreadFactory("ahc-timeout-test")); + // The client's own wheel settings, so that the timer case is timed the way it would be in production. + timer = new HashedWheelTimer(runnable -> { + Thread thread = new Thread(runnable, "ahc-timeout-test-timer"); + thread.setDaemon(true); + timerThread.set(thread); + return thread; + }, 100, TimeUnit.MILLISECONDS, 512, false); } @AfterEach public void stop() throws Throwable { + released.countDown(); server.close(); + timer.stop(); + eventLoopGroup.shutdownGracefully(0, 0, TimeUnit.MILLISECONDS).await(10, TimeUnit.SECONDS); } @Test - public void byDefaultTheTimeoutIsDeliveredFromTheTimerThread() throws Throwable { - String thread = threadDeliveringRequestTimeout(false); + public void byDefaultAnExpiryIsDeliveredFromTheTimerThread() throws Throwable { + Recorder recorder = runAgainstAnUnansweringServer(baseConfig(false)); - assertTrue(thread.contains(TIMER_MARKER), - "expected the timer thread by default, got " + thread); + assertSame(timerThread.get(), recorder.deliveredOn.get(), + "expected the timer's own thread, got " + recorder.deliveredOn.get()); } @Test - public void withEventLoopTimeoutsTheTimeoutIsDeliveredFromAnEventLoop() throws Throwable { - String thread = threadDeliveringRequestTimeout(true); + public void anExchangeThatConnectedExpiresOnItsChannelsLoop() throws Throwable { + // The request timeout is armed before the connect, so on this path it starts on the timer and is moved + // to the loop once there is a channel. A deadline it could reach before connecting would be delivered + // from the timer quite correctly -- there was no channel to deliver it from -- and prove nothing, hence + // a budget the first connect of a JVM comfortably fits inside. + Recorder recorder = runAgainstAnUnansweringServer(baseConfig(true).setRequestTimeout(Duration.ofSeconds(1))); + + assertNull(recorder.pooledChannel.get(), "this request was meant to open its own connection"); + assertDeliveredOnTheLoopOf(recorder.connectedChannel.get(), recorder); + } + + @Test + public void anExchangeOnAPooledChannelExpiresOnThatChannelsLoop() throws Throwable { + Recorder first = new Recorder(); + Recorder second = new Recorder(); + + withClient(baseConfig(true)).run(client -> withServer(server).run(server -> { + server.enqueueOk(); + client.prepareGet(server.getHttpUrl() + "/foo/bar").execute(first); + first.awaitCompletion(); + // Waited for, not assumed: the connection is offered to the pool around the same time as the future + // completes, and a second request that overtook the offer would open its own connection and test + // the wrong branch. + assertTrue(first.offered.await(10, TimeUnit.SECONDS), "the first connection was never pooled"); + + server.enqueueResponse(response -> awaitRelease()); + client.prepareGet(server.getHttpUrl() + "/foo/bar").execute(second); + second.awaitTimeout(); + })); + + assertNotNull(second.pooledChannel.get(), "the second request did not reuse the pooled connection"); + assertDeliveredOnTheLoopOf(second.pooledChannel.get(), second); + } - 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); + @Test + public void aReadTimeoutIsDeliveredFromTheChannelsLoopAsWell() throws Throwable { + // A request timeout far enough out that the read timeout is the one that fires: the read timeout is + // armed after the request is written, by which point the exchange is already homed on its loop. + Recorder recorder = runAgainstAnUnansweringServer(baseConfig(true) + .setRequestTimeout(Duration.ofSeconds(10)) + .setReadTimeout(SHORT_TIMEOUT)); + + assertTrue(recorder.cause.get().getMessage().startsWith("Read timeout"), + "expected a read timeout, got " + recorder.cause.get().getMessage()); + assertDeliveredOnTheLoopOf(recorder.connectedChannel.get(), recorder); + } + + private DefaultAsyncHttpClientConfig.Builder baseConfig(boolean useEventLoopTimeouts) { + return config() + .setNettyTimer(timer) + .setEventLoopGroup(eventLoopGroup) + .setMaxRedirects(0) + .setRequestTimeout(SHORT_TIMEOUT) + .setUseEventLoopTimeouts(useEventLoopTimeouts); } /** - * 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. + * Runs one request against an endpoint that never answers, and returns what its handler saw. */ - 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); + private Recorder runAgainstAnUnansweringServer(DefaultAsyncHttpClientConfig.Builder builder) throws Throwable { + Recorder recorder = new Recorder(); 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"); + server.enqueueResponse(response -> awaitRelease()); + + client.prepareGet(server.getHttpUrl() + "/foo/bar").execute(recorder); + recorder.awaitTimeout(); })); - 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; + return recorder; + } + + private static void assertDeliveredOnTheLoopOf(Channel channel, Recorder recorder) { + assertNotNull(channel, "the exchange never reported a channel"); + assertTrue(channel.eventLoop().inEventLoop(recorder.deliveredOn.get()), + "expected the channel's own loop, got " + recorder.deliveredOn.get()); + } + + private void awaitRelease() { + try { + released.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + /** + * Records the channel the exchange ran on, and the thread its expiry was delivered from. + */ + private static final class Recorder extends AsyncCompletionHandler { + + private final CountDownLatch settled = new CountDownLatch(1); + private final CountDownLatch offered = new CountDownLatch(1); + private final AtomicReference connectedChannel = new AtomicReference<>(); + private final AtomicReference pooledChannel = new AtomicReference<>(); + private final AtomicReference deliveredOn = new AtomicReference<>(); + private final AtomicReference cause = new AtomicReference<>(); + + @Override + public void onTcpConnectSuccess(InetSocketAddress remoteAddress, Channel connection) { + connectedChannel.set(connection); + } + + @Override + public void onConnectionPooled(Channel connection) { + pooledChannel.set(connection); + } + + @Override + public void onConnectionOffer(Channel connection) { + offered.countDown(); + } + + @Override + public Void onCompleted(Response response) { + settled.countDown(); + return null; + } + + @Override + public void onThrowable(Throwable t) { + deliveredOn.set(Thread.currentThread()); + cause.set(t); + settled.countDown(); + } + + void awaitCompletion() throws InterruptedException { + assertTrue(settled.await(30, TimeUnit.SECONDS), "the request never settled"); + assertNull(cause.get(), "the request was meant to succeed, got " + cause.get()); + } + + void awaitTimeout() throws InterruptedException { + assertTrue(settled.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 timeout, got " + cause.get()); + assertNotNull(deliveredOn.get(), "onThrowable was not called"); + } } } diff --git a/pom.xml b/pom.xml index ed83e98417..a8a3872216 100644 --- a/pom.xml +++ b/pom.xml @@ -501,6 +501,12 @@ "new": "method java.lang.String org.asynchttpclient.util.AuthenticatorUtils::computeRspAuth(org.asynchttpclient.Realm, java.lang.String, java.lang.String, java.lang.String, java.lang.String, java.lang.String, java.lang.String, java.lang.String)", "justification": "The public computeRspAuth(Realm) was removed by the Digest mutual-authentication fix; revapi pairs it with the new private helper of the same name and reports the removal as a visibility reduction. It computed the expected rspauth from the Realm carried on the response future, whose uri is stale after a redirect and whose cnonce is regenerated on every build() - i.e. never the values actually sent on the wire - so it could only ever produce a value that disagrees with a conformant server. Verification now derives every parameter from the request's own Authorization header. The method had no caller inside the library; leaving it would invite it to be wired back in." }, + { + "code": "java.method.exception.checkedRemoved", + "old": "method void io.netty.util.TimerTask::run(io.netty.util.Timeout) throws java.lang.Exception @ org.asynchttpclient.netty.timeout.TimeoutTimerTask", + "new": "method void org.asynchttpclient.netty.timeout.TimeoutTimerTask::run(io.netty.util.Timeout)", + "justification": "TimeoutTimerTask now also implements Runnable, so a timeout can be armed on an event loop as well as on the client's Timer, and run() delegates to run(Timeout). Netty declares run(Timeout) throws Exception, which run() would have to catch and could only log - a handler for an exception neither subclass throws and no other subclass can throw, the only constructor being package private, so nothing outside this internal package can extend the class. Narrowing the declaration removes the dead handler. Binary compatible; the sole source effect would be on code catching a narrower checked exception around a direct run(Timeout) call, which has no caller outside the library. Scoped to this one method." + }, { "code": "java.annotation.removed", "old": "method void io.netty.channel.ChannelInboundHandlerAdapter::userEventTriggered(io.netty.channel.ChannelHandlerContext, java.lang.Object) throws java.lang.Exception @ org.asynchttpclient.netty.handler.Http2Handler",