Arm request timeouts on an event loop

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()}.
+ * <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 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.
+ * <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;
+ }
+
  /**
  * Is HTTP redirect enabled
  *

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,

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

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

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 <T> ListenableFuture<T> sendRequestWithOpenChannel(NettyResponseFuture<T
  return future;
  }
 
+ future.setChannelState(ChannelState.POOLED);
+ // Before the timeout is armed, not after: an expiry reaches the channel only through the future, and on
+ // an event loop a short enough deadline can be delivered before the next statement would have run.
+ future.attachChannel(channel, false);
+
  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);
- future.attachChannel(channel, false);
-
  if (LOGGER.isDebugEnabled()) {
  HttpRequest httpRequest = future.getNettyRequest().getHttpRequest();
  LOGGER.debug("Using open Channel {} for {} '{}'", channel, httpRequest.method(), httpRequest.uri());
@@ -1080,10 +1083,35 @@ 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. 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);
  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 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) {

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

@@ -15,31 +15,105 @@
  */
 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;
 
-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. Which one an exchange uses is
+ * {@link org.asynchttpclient.AsyncHttpClientConfig#isUseEventLoopTimeouts()}.
+ */
+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, 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;
  this.requestSender = requestSender;
  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() {
+ // 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(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
+ */
+ 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));