diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index 7304626083..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 * diff --git a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java index 75aa1bd16a..a1eed3cc97 100644 --- a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java @@ -97,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; @@ -157,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; @@ -258,6 +260,7 @@ private DefaultAsyncHttpClientConfig(// http Duration connectTimeout, Duration requestTimeout, Duration readTimeout, + boolean useEventLoopTimeouts, Duration shutdownQuietPeriod, Duration shutdownTimeout, @@ -367,6 +370,7 @@ private DefaultAsyncHttpClientConfig(// http this.connectTimeout = connectTimeout; this.requestTimeout = requestTimeout; this.readTimeout = readTimeout; + this.useEventLoopTimeouts = useEventLoopTimeouts; this.shutdownQuietPeriod = shutdownQuietPeriod; this.shutdownTimeout = shutdownTimeout; @@ -585,6 +589,11 @@ public Duration getReadTimeout() { return readTimeout; } + @Override + public boolean isUseEventLoopTimeouts() { + return useEventLoopTimeouts; + } + @Override public Duration getShutdownQuietPeriod() { return shutdownQuietPeriod; @@ -958,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(); @@ -1064,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(); @@ -1355,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; @@ -1764,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 a31fdf2855..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"; @@ -154,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); } diff --git a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java b/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java index 86d312617e..bcd9032833 100755 --- a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java +++ b/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java @@ -466,6 +466,11 @@ public void setTimeoutsHolder(TimeoutsHolder timeoutsHolder) { if (ref != null) { ref.cancel(); } + if (timeoutsHolder != null) { + // Armed here rather than by the caller: a holder can run its timeout the moment it is armed, so it + // has to be reachable from this future first, and no caller can then install one that never arms. + timeoutsHolder.start(); + } } public boolean isInAuth() { 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 af3610164d..47775c4479 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; @@ -401,15 +402,17 @@ 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. 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, + @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); + // Arms the timeout as a part of installing the holder, which is why the pooled path attaches the + // channel first: an expiry that lands immediately reaches the channel only through the future. nettyResponseFuture.setTimeoutsHolder(timeoutsHolder); } + /** + * 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) { + return config.isUseEventLoopTimeouts() && channel != null ? channel.eventLoop() : null; + } + private static void scheduleReadTimeout(NettyResponseFuture nettyResponseFuture) { TimeoutsHolder timeoutsHolder = nettyResponseFuture.getTimeoutsHolder(); if (timeoutsHolder != null) { diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java index 8b0d4373a1..18d3078b66 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java @@ -22,7 +22,7 @@ import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime; -public class ReadTimeoutTimerTask extends TimeoutTimerTask { +public class ReadTimeoutTimerTask extends TimeoutTimerTask implements Runnable { private final long readTimeout; @@ -31,6 +31,15 @@ public class ReadTimeoutTimerTask extends TimeoutTimerTask { this.readTimeout = readTimeout; } + /** + * The event-loop entry point. Nothing below reads the {@link Timeout}, which is the timer's own handle on + * this task and something an event loop has no equivalent of, so both entry points share one body. + */ + @Override + public void run() { + run(null); + } + @Override public void run(Timeout timeout) { if (done.getAndSet(true) || requestSender.isClosed()) { diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java index 74c5d0197a..1b913ea7ad 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java @@ -22,7 +22,7 @@ import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime; -public class RequestTimeoutTimerTask extends TimeoutTimerTask { +public class RequestTimeoutTimerTask extends TimeoutTimerTask implements Runnable { private final long requestTimeout; @@ -34,6 +34,15 @@ public class RequestTimeoutTimerTask extends TimeoutTimerTask { this.requestTimeout = requestTimeout; } + /** + * The event-loop entry point. Nothing below reads the {@link Timeout}, which is the timer's own handle on + * this task and something an event loop has no equivalent of, so both entry points share one body. + */ + @Override + public void run() { + run(null); + } + @Override public void run(Timeout timeout) { if (done.getAndSet(true) || requestSender.isClosed()) { 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..6e81b7007d 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java @@ -15,16 +15,27 @@ */ package org.asynchttpclient.netty.timeout; +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; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.net.InetSocketAddress; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; +/** + * A timeout that can be armed either on a {@link io.netty.util.Timer} or on an event loop; which one an + * exchange uses is {@link org.asynchttpclient.AsyncHttpClientConfig#isUseEventLoopTimeouts()}. An event loop + * schedules {@link Runnable}s, so each subclass implements that as a second entry point into the same body. + * This class stays a {@link TimerTask} alone: {@code Runnable#run} declares no checked exception, so a + * {@code run()} here would have to catch what {@link TimerTask#run(Timeout)} declares and no subclass throws. + */ public abstract class TimeoutTimerTask implements TimerTask { private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutTimerTask.class); @@ -33,6 +44,11 @@ public abstract class TimeoutTimerTask implements TimerTask { protected final NettyRequestSender requestSender; final TimeoutsHolder timeoutsHolder; volatile NettyResponseFuture nettyResponseFuture; + // The scheduled entry this task is armed on, one field per scheduler so that a scheduler changing its + // return type is a compile error rather than a cancellation that silently stops working. At most one is + // ever set. Held here rather than in a wrapper so arming allocates nothing beyond what the scheduler needs. + private volatile @Nullable Timeout timerHandle; + private volatile @Nullable ScheduledFuture loopHandle; TimeoutTimerTask(NettyResponseFuture nettyResponseFuture, NettyRequestSender requestSender, TimeoutsHolder timeoutsHolder) { this.nettyResponseFuture = nettyResponseFuture; @@ -40,6 +56,55 @@ public abstract class TimeoutTimerTask implements TimerTask { this.timeoutsHolder = timeoutsHolder; } + void armedOn(Timeout handle) { + // Each clears the other, so a handle left over from a previous arming cannot mask the live one and + // leave its entry sitting in a scheduler, holding the future until a deadline nobody is waiting for. + loopHandle = null; + timerHandle = handle; + } + + void armedOn(ScheduledFuture handle) { + timerHandle = null; + 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 + */ + 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; + } + + /** + * 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..ba76939958 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -15,37 +15,65 @@ */ 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 io.netty.util.concurrent.ScheduledFuture; 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, 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 { - private final Timeout requestTimeout; + private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutsHolder.class); + private final AtomicBoolean cancelled = new AtomicBoolean(); private final Timer nettyTimer; + private volatile @Nullable EventExecutor eventExecutor; private final NettyRequestSender requestSender; private final long requestTimeoutMillisTime; + private final long requestTimeoutValue; private final long readTimeoutValue; - private volatile Timeout readTimeout; + private final boolean useEventLoopTimeouts; + private final @Nullable RequestTimeoutTimerTask requestTimeoutTask; + 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 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) { this.nettyTimer = nettyTimer; + this.eventExecutor = eventExecutor; this.nettyResponseFuture = nettyResponseFuture; this.requestSender = requestSender; + useEventLoopTimeouts = config.isUseEventLoopTimeouts(); remoteAddress = originalRemoteAddress; final Request targetRequest = nettyResponseFuture.getTargetRequest(); @@ -58,15 +86,52 @@ public TimeoutsHolder(Timer nettyTimer, NettyResponseFuture nettyResponseFutu requestTimeoutInMs = config.getRequestTimeout().toMillis(); } + requestTimeoutValue = requestTimeoutInMs; if (requestTimeoutInMs > -1) { requestTimeoutMillisTime = unpreciseMillisTime() + requestTimeoutInMs; - requestTimeout = newTimeout(new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs), requestTimeoutInMs); + requestTimeoutTask = new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs); } else { requestTimeoutMillisTime = -1L; - requestTimeout = null; + requestTimeoutTask = null; + } + } + + /** + * 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. + *

+ * Called by {@link org.asynchttpclient.netty.NettyResponseFuture#setTimeoutsHolder}, so that installing a + * holder is what arms it and neither can be done without the other. + */ + public void start() { + if (requestTimeoutTask != null) { + // The configured duration rather than the remaining time: this runs within microseconds of the + // constructor, and reading the clock again would only expose the deadline to a step between the two. + arm(requestTimeoutTask, requestTimeoutValue); } } + /** + * 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; } @@ -81,14 +146,16 @@ public void startReadTimeout() { } } - void startReadTimeout(ReadTimeoutTimerTask task) { - if (requestTimeout == null || !requestTimeout.isExpired() && readTimeoutValue < requestTimeoutMillisTime - unpreciseMillisTime()) { + void startReadTimeout(@Nullable ReadTimeoutTimerTask task) { + if (requestTimeoutTask == null + || !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 +165,73 @@ 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(); + } + } + + 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. Leaves it unarmed when the client is shutting down, in which case there is no + * timeout to deliver anyway. + * + * @param a task that both schedulers accept: the timer takes a {@link io.netty.util.TimerTask} and an + * event loop a {@link Runnable}, and only the concrete subclasses are both + */ + private void arm(T 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; + } + EventExecutor executor = eventExecutor; + if (executor != null && !executor.isShuttingDown()) { + ScheduledFuture handle = null; + try { + handle = executor.schedule(task, delay, TimeUnit.MILLISECONDS); + } 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); + } + // Outside the try: only the schedule above may fall back to the timer. Anything thrown while + // recording or unwinding the entry belongs to an exchange that is already armed. + if (handle != null) { + task.armedOn(handle); + cancelIfRaced(task); + return; + } + } + if (nettyTimer == null) { + return; + } + task.armedOn(nettyTimer.newTimeout(task, delay, TimeUnit.MILLISECONDS)); + 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); } - return nettyTimer.newTimeout(task, delay, TimeUnit.MILLISECONDS); } } 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..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 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..6b080fc077 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/EventLoopTimeoutTest.java @@ -0,0 +1,230 @@ +/* + * 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.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; +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.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; + +/** + * 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 Duration SHORT_TIMEOUT = Duration.ofMillis(200); + // For the cases whose request has to connect, or succeed, before the deadline can prove anything. + private static final Duration COLD_TIMEOUT = Duration.ofSeconds(1); + + 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(); + // Eight, not two: with two loops a timeout armed on the wrong one is on the right one half the time, + // and these assertions would pass about half the runs against the bug they exist to catch. + eventLoopGroup = new NioEventLoopGroup(8, 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 byDefaultAnExpiryIsDeliveredFromTheTimerThread() throws Throwable { + Recorder recorder = runAgainstAnUnansweringServer(baseConfig(false)); + + assertSame(timerThread.get(), recorder.deliveredOn.get(), + "expected the timer's own thread, got " + recorder.deliveredOn.get()); + } + + @Test + 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(COLD_TIMEOUT)); + + 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(); + + // The first request here is the cold one -- class loading, the connect, the server's own first + // response -- and it is meant to succeed, so it gets the same budget the connecting case needs. + withClient(baseConfig(true).setRequestTimeout(COLD_TIMEOUT)).run(client -> withServer(server).run(server -> { + server.enqueueOk(); + client.prepareGet(server.getHttpUrl() + "/foo/bar").execute(first); + first.awaitCompletion(); + + server.enqueueResponse(response -> awaitRelease()); + client.prepareGet(server.getHttpUrl() + "/foo/bar").execute(second); + second.awaitTimeout(); + })); + + // The pool offer happens before the future completes, so awaiting the first request above is enough to + // know the connection was there to be reused; this says the second one actually took it. + assertNotNull(second.pooledChannel.get(), "the second request did not reuse the pooled connection"); + assertDeliveredOnTheLoopOf(second.pooledChannel.get(), second); + } + + @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 never answers, and returns what its handler saw. + */ + private Recorder runAgainstAnUnansweringServer(DefaultAsyncHttpClientConfig.Builder builder) throws Throwable { + Recorder recorder = new Recorder(); + + withClient(builder).run(client -> withServer(server).run(server -> { + server.enqueueResponse(response -> awaitRelease()); + + client.prepareGet(server.getHttpUrl() + "/foo/bar").execute(recorder); + recorder.awaitTimeout(); + })); + + 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 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 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"); + } + } +}