Arm request timeouts on an event loop

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()}.
+     * <p>
+     * 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.
+     * <p>
+     * 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.
+     * <p>
+     * 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
      */

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,

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);
     }

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 <T> ListenableFuture<T> sendRequestWithOpenChannel(NettyResponseFuture<T
         SocketAddress channelRemoteAddress = channel.remoteAddress();
         if (channelRemoteAddress != null) {
             // otherwise, bad luck, the channel was closed, see bellow
-            scheduleRequestTimeout(future, (InetSocketAddress) channelRemoteAddress);
+            scheduleRequestTimeout(future, (InetSocketAddress) channelRemoteAddress, channel);
         }
 
         future.setChannelState(ChannelState.POOLED);
@@ -1080,12 +1081,39 @@ private static void configureTransferAdapter(AsyncHandler<?> 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) {

client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java

@@ -15,31 +15,84 @@
  */
 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);
 
     protected final AtomicBoolean done = new AtomicBoolean();
     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;
         this.requestSender = requestSender;
         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));