Start request timeout before DNS and connect, close AsyncHttpClient#1060

Author: slandelle

client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java

@@ -325,6 +325,10 @@ public int incrementAndGetCurrentRedirectCount() {
  public void setTimeoutsHolder(TimeoutsHolder timeoutsHolder) {
  this.timeoutsHolder = timeoutsHolder;
  }
+ 
+ public TimeoutsHolder getTimeoutsHolder() {
+ return timeoutsHolder;
+ }
 
  public AtomicBoolean getInAuth() {
  return inAuth;

client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java

@@ -16,7 +16,6 @@
 import static org.asynchttpclient.util.Assertions.assertNotNull;
 import static org.asynchttpclient.util.AuthenticatorUtils.*;
 import static org.asynchttpclient.util.HttpConstants.Methods.*;
-import static org.asynchttpclient.util.HttpUtils.requestTimeout;
 import static org.asynchttpclient.util.MiscUtils.getCause;
 import static org.asynchttpclient.util.ProxyUtils.getProxyServer;
 import io.netty.bootstrap.Bootstrap;
@@ -27,14 +26,11 @@
 import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.codec.http.HttpRequest;
-import io.netty.util.Timeout;
 import io.netty.util.Timer;
-import io.netty.util.TimerTask;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.List;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.asynchttpclient.AsyncHandler;
@@ -56,8 +52,6 @@
 import org.asynchttpclient.netty.channel.ChannelState;
 import org.asynchttpclient.netty.channel.Channels;
 import org.asynchttpclient.netty.channel.NettyConnectListener;
-import org.asynchttpclient.netty.timeout.ReadTimeoutTimerTask;
-import org.asynchttpclient.netty.timeout.RequestTimeoutTimerTask;
 import org.asynchttpclient.netty.timeout.TimeoutsHolder;
 import org.asynchttpclient.proxy.ProxyServer;
 import org.asynchttpclient.resolver.RequestHostnameResolver;
@@ -218,6 +212,7 @@ private <T> ListenableFuture<T> sendRequestWithOpenChannel(Request request, Prox
  if (asyncHandler instanceof AsyncHandlerExtensions)
  AsyncHandlerExtensions.class.cast(asyncHandler).onConnectionPooled(channel);
 
+ scheduleRequestTimeout(future);
  future.setChannelState(ChannelState.POOLED);
  future.attachChannel(channel, false);
 
@@ -274,6 +269,8 @@ private <T> ListenableFuture<T> sendRequestWithNewChannel(//
  return future;
  }
 
+ scheduleRequestTimeout(future);
+
  RequestHostnameResolver.INSTANCE.resolve(request, proxy, asyncHandler)//
  .addListener(new SimpleFutureListener<List<InetSocketAddress>>() {
 
@@ -341,9 +338,9 @@ public <T> void writeRequest(NettyResponseFuture<T> future, Channel channel) {
  if (writeBody)
  nettyRequest.getBody().write(channel, future);
 
- // don't bother scheduling timeouts if channel became invalid
+ // don't bother scheduling read timeout if channel became invalid
  if (Channels.isChannelValid(channel))
- scheduleTimeouts(future);
+ scheduleReadTimeout(future);
 
  } catch (Exception e) {
  LOGGER.error("Can't write request", e);
@@ -356,27 +353,16 @@ private void configureTransferAdapter(AsyncHandler<?> handler, HttpRequest httpR
  TransferCompletionHandler.class.cast(handler).headers(h);
  }
 
- private void scheduleTimeouts(NettyResponseFuture<?> nettyResponseFuture) {
-
+ private void scheduleRequestTimeout(NettyResponseFuture<?> nettyResponseFuture) {
  nettyResponseFuture.touch();
- int requestTimeoutInMs = requestTimeout(config, nettyResponseFuture.getTargetRequest());
- TimeoutsHolder timeoutsHolder = new TimeoutsHolder();
- if (requestTimeoutInMs != -1) {
- Timeout requestTimeout = newTimeout(new RequestTimeoutTimerTask(nettyResponseFuture, this, timeoutsHolder, requestTimeoutInMs), requestTimeoutInMs);
- timeoutsHolder.requestTimeout = requestTimeout;
- }
-
- int readTimeoutValue = config.getReadTimeout();
- if (readTimeoutValue != -1 && readTimeoutValue < requestTimeoutInMs) {
- // no need to schedule a readTimeout if the requestTimeout happens first
- Timeout readTimeout = newTimeout(new ReadTimeoutTimerTask(nettyResponseFuture, this, timeoutsHolder, requestTimeoutInMs, readTimeoutValue), readTimeoutValue);
- timeoutsHolder.readTimeout = readTimeout;
- }
+ TimeoutsHolder timeoutsHolder = new TimeoutsHolder(nettyTimer, nettyResponseFuture, this, config);
  nettyResponseFuture.setTimeoutsHolder(timeoutsHolder);
  }
 
- public Timeout newTimeout(TimerTask task, long delay) {
- return nettyTimer.newTimeout(task, delay, TimeUnit.MILLISECONDS);
+ private void scheduleReadTimeout(NettyResponseFuture<?> nettyResponseFuture) {
+ nettyResponseFuture.touch();
+ TimeoutsHolder timeoutsHolder = nettyResponseFuture.getTimeoutsHolder();
+ timeoutsHolder.startReadTimeout();
  }
 
  public void abort(Channel channel, NettyResponseFuture<?> future, Throwable t) {

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

@@ -22,17 +22,14 @@
 public class ReadTimeoutTimerTask extends TimeoutTimerTask {
 
  private final long readTimeout;
- private final long requestTimeoutInstant;
 
  public ReadTimeoutTimerTask(//
  NettyResponseFuture<?> nettyResponseFuture,//
  NettyRequestSender requestSender,//
  TimeoutsHolder timeoutsHolder,//
- long requestTimeout,//
  long readTimeout) {
  super(nettyResponseFuture, requestSender, timeoutsHolder);
  this.readTimeout = readTimeout;
- requestTimeoutInstant = requestTimeout >= 0 ? nettyResponseFuture.getStart() + requestTimeout : Long.MAX_VALUE;
  }
 
  public void run(Timeout timeout) throws Exception {
@@ -52,20 +49,15 @@ public void run(Timeout timeout) throws Exception {
 
  if (durationBeforeCurrentReadTimeout <= 0L) {
  // idleConnectTimeout reached
- String message = "Read timeout to " + remoteAddress + " after " + readTimeout + " ms";
+ String message = "Read timeout to " + timeoutsHolder.remoteAddress() + " after " + readTimeout + " ms";
  long durationSinceLastTouch = now - nettyResponseFuture.getLastTouch();
  expire(message, durationSinceLastTouch);
  // cancel request timeout sibling
  timeoutsHolder.cancel();
 
- } else if (currentReadTimeoutInstant < requestTimeoutInstant) {
- // reschedule
- done.set(false);
- timeoutsHolder.readTimeout = requestSender.newTimeout(this, durationBeforeCurrentReadTimeout);
-
  } else {
- // otherwise, no need to reschedule: requestTimeout will happen sooner
- timeoutsHolder.readTimeout = null;
+ done.set(false);
+ timeoutsHolder.startReadTimeout(this);
  }
  }
 }

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

@@ -43,7 +43,7 @@ public void run(Timeout timeout) throws Exception {
  if (nettyResponseFuture.isDone())
  return;
 
- String message = "Request timeout to " + remoteAddress + " after " + requestTimeout + "ms";
+ String message = "Request timeout to " + timeoutsHolder.remoteAddress() + " after " + requestTimeout + "ms";
  long age = millisTime() - nettyResponseFuture.getStart();
  expire(message, age);
  }

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

@@ -13,10 +13,8 @@
  */
 package org.asynchttpclient.netty.timeout;
 
-import io.netty.channel.Channel;
 import io.netty.util.TimerTask;
 
-import java.net.SocketAddress;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -33,16 +31,11 @@ public abstract class TimeoutTimerTask implements TimerTask {
  protected volatile NettyResponseFuture<?> nettyResponseFuture;
  protected final NettyRequestSender requestSender;
  protected final TimeoutsHolder timeoutsHolder;
- protected final String remoteAddress;
 
  public TimeoutTimerTask(NettyResponseFuture<?> nettyResponseFuture, NettyRequestSender requestSender, TimeoutsHolder timeoutsHolder) {
  this.nettyResponseFuture = nettyResponseFuture;
  this.requestSender = requestSender;
  this.timeoutsHolder = timeoutsHolder;
- // saving remote address as the channel might be removed from the future when an exception occurs
- Channel channel = nettyResponseFuture.channel();
- SocketAddress sa = channel == null ? null : channel.remoteAddress();
- remoteAddress = sa == null ? "not-connected" : sa.toString();
  }
 
  protected void expire(String message, long time) {
@@ -55,7 +48,8 @@ protected void expire(String message, long time) {
  * Holding a reference to the future might mean holding a reference to the channel, and heavy objects such as SslEngines
  */
  public void clean() {
- if (done.compareAndSet(false, true))
+ if (done.compareAndSet(false, true)) {
  nettyResponseFuture = null;
+ }
  }
 }

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

@@ -13,28 +13,106 @@
  */
 package org.asynchttpclient.netty.timeout;
 
+import static org.asynchttpclient.util.DateUtils.millisTime;
+import io.netty.channel.Channel;
 import io.netty.util.Timeout;
+import io.netty.util.Timer;
+import io.netty.util.TimerTask;
 
+import java.net.SocketAddress;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.asynchttpclient.AsyncHttpClientConfig;
+import org.asynchttpclient.netty.NettyResponseFuture;
+import org.asynchttpclient.netty.request.NettyRequestSender;
+
 public class TimeoutsHolder {
 
  private final AtomicBoolean cancelled = new AtomicBoolean();
- public volatile Timeout requestTimeout;
+
+ private final Timer nettyTimer;
+ private final NettyRequestSender requestSender;
+ private final long requestTimeoutMillisTime;
+ private final int readTimeoutValue;
+
+ private volatile NettyResponseFuture<?> nettyResponseFuture;
+ public final Timeout requestTimeout;
  public volatile Timeout readTimeout;
+ private volatile String remoteAddress = "not-connected";
+
+ public TimeoutsHolder(Timer nettyTimer, NettyResponseFuture<?> nettyResponseFuture, NettyRequestSender requestSender, AsyncHttpClientConfig config) {
+ this.nettyTimer = nettyTimer;
+ this.nettyResponseFuture = nettyResponseFuture;
+ this.requestSender = requestSender;
+ this.readTimeoutValue = config.getReadTimeout();
+
+ int requestTimeoutInMs = nettyResponseFuture.getTargetRequest().getRequestTimeout();
+ if (requestTimeoutInMs == 0) {
+ requestTimeoutInMs = config.getRequestTimeout();
+ }
+
+ if (requestTimeoutInMs != -1) {
+ requestTimeoutMillisTime = millisTime() + requestTimeoutInMs;
+ requestTimeout = newTimeout(new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs), requestTimeoutInMs);
+ } else {
+ requestTimeoutMillisTime = -1L;
+ requestTimeout = null;
+ }
+ }
+
+ private void initRemoteAddress() {
+ Channel channel = nettyResponseFuture.channel();
+ if (channel != null) {
+ SocketAddress sa = channel.remoteAddress();
+ if (sa != null) {
+ remoteAddress = sa.toString();
+ }
+ }
+ }
+
+ public void startReadTimeout() {
+ // we should be connected now
+ initRemoteAddress();
+ if (readTimeoutValue != -1) {
+ startReadTimeout(null);
+ }
+ }
+
+ void startReadTimeout(ReadTimeoutTimerTask task) {
+ if (requestTimeout == null || (!requestTimeout.isExpired() && readTimeoutValue > (requestTimeoutMillisTime - millisTime()))) {
+ // 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);
+ }
+ Timeout readTimeout = newTimeout(task, readTimeoutValue);
+ this.readTimeout = readTimeout;
+
+ } else if (task != null) {
+ // read timeout couldn't re-scheduling itself, clean up
+ task.clean();
+ }
+ }
 
  public void cancel() {
  if (cancelled.compareAndSet(false, true)) {
  if (requestTimeout != null) {
  requestTimeout.cancel();
  RequestTimeoutTimerTask.class.cast(requestTimeout.task()).clean();
- requestTimeout = null;
  }
  if (readTimeout != null) {
  readTimeout.cancel();
  ReadTimeoutTimerTask.class.cast(readTimeout.task()).clean();
- readTimeout = null;
  }
  }
  }
+
+ private Timeout newTimeout(TimerTask task, long delay) {
+ return nettyTimer.newTimeout(task, delay, TimeUnit.MILLISECONDS);
+ }
+ 
+ String remoteAddress() {
+ return remoteAddress;
+ }
 }

client/src/main/java/org/asynchttpclient/util/HttpUtils.java

@@ -78,10 +78,6 @@ public static Charset parseCharset(String contentType) {
  return null;
  }
 
- public static int requestTimeout(AsyncHttpClientConfig config, Request request) {
- return request.getRequestTimeout() != 0 ? request.getRequestTimeout() : config.getRequestTimeout();
- }
-
  public static boolean followRedirect(AsyncHttpClientConfig config, Request request) {
  return request.getFollowRedirect() != null ? request.getFollowRedirect().booleanValue() : config.isFollowRedirect();
  }