Reimplement Netty provider timeouts, close AsyncHttpClient#457

Author: Stephane Landelle

providers/netty/src/main/java/org/asynchttpclient/providers/netty/channel/Channels.java

@@ -37,6 +37,9 @@
 import io.netty.handler.stream.ChunkedWriteHandler;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timeout;
+import io.netty.util.TimerTask;
 
 import java.io.IOException;
 import java.lang.reflect.Field;
@@ -46,6 +49,7 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
 
 import javax.net.ssl.SSLEngine;
 
@@ -104,6 +108,8 @@ public boolean remove(Object o) {
         }
     };
 
+    private final HashedWheelTimer hashedWheelTimer;
+
     public Channels(final AsyncHttpClientConfig config, NettyAsyncHttpProviderConfig asyncHttpProviderConfig) {
 
         this.config = config;
@@ -176,6 +182,9 @@ public Channels(final AsyncHttpClientConfig config, NettyAsyncHttpProviderConfig
         webSocketBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeOut);
         secureBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeOut);
         secureWebSocketBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeOut);
+
+        hashedWheelTimer = new HashedWheelTimer();
+        hashedWheelTimer.start();
     }
 
     private SSLEngine createSSLEngine() throws IOException, GeneralSecurityException {
@@ -271,13 +280,15 @@ public void close() {
             Object attribute = getDefaultAttribute(channel);
             if (attribute instanceof NettyResponseFuture<?>) {
                 NettyResponseFuture<?> future = (NettyResponseFuture<?>) attribute;
-                future.setReaperFuture(null);
+                future.cancelTimeouts();
             }
         }
         openChannels.close();
         if (allowReleaseEventLoopGroup) {
             eventLoopGroup.shutdownGracefully();
         }
+
+        hashedWheelTimer.stop();
     }
 
     // some servers can use the same port for HTTP and HTTPS
@@ -456,6 +467,10 @@ public void abort(NettyResponseFuture<?> future, Throwable t) {
         future.abort(t);
     }
 
+    public Timeout newTimeoutInMs(TimerTask task, long delayInMs) {
+        return hashedWheelTimer.newTimeout(task, delayInMs, TimeUnit.MILLISECONDS);
+    }
+
     public static SslHandler getSslHandler(Channel channel) {
         return (SslHandler) channel.pipeline().get(Channels.SSL_HANDLER);
     }

providers/netty/src/main/java/org/asynchttpclient/providers/netty/future/FutureReaper.java

@@ -41,6 +41,7 @@
 import org.asynchttpclient.providers.netty.DiscardEvent;
 import org.asynchttpclient.providers.netty.channel.Channels;
 import org.asynchttpclient.providers.netty.request.NettyRequest;
+import org.asynchttpclient.providers.netty.request.timeout.TimeoutsHolder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -59,7 +60,8 @@ public enum STATE {
     }
 
     private final int requestTimeoutInMs;
-    private final AsyncHttpClientConfig config;
+    private volatile boolean requestTimeoutReached;
+    private volatile boolean idleConnectionTimeoutReached;
     private final long start = millisTime();
     private final ConnectionPoolKeyStrategy connectionPoolKeyStrategy;
     private final ProxyServer proxyServer;
@@ -80,7 +82,7 @@ public enum STATE {
     private final AtomicBoolean throwableCalled = new AtomicBoolean(false);
     private final AtomicReference<V> content = new AtomicReference<V>();
     private final AtomicReference<ExecutionException> exEx = new AtomicReference<ExecutionException>();
-    private volatile FutureReaper reaperFuture;
+    private volatile TimeoutsHolder timeoutsHolder;
 
     // state mutated only inside the event loop
     private Channel channel;
@@ -111,7 +113,6 @@ public NettyResponseFuture(URI uri,//
         this.request = request;
         this.nettyRequest = nettyRequest;
         this.uri = uri;
-        this.config = config;
         this.connectionPoolKeyStrategy = connectionPoolKeyStrategy;
         this.proxyServer = proxyServer;
 
@@ -156,7 +157,7 @@ public void setAsyncHandler(AsyncHandler<V> asyncHandler) {
 
     @Override
     public boolean cancel(boolean force) {
-        cancelReaper();
+        cancelTimeouts();
 
         if (isCancelled.get())
             return false;
@@ -186,31 +187,39 @@ public boolean cancel(boolean force) {
      * @return <code>true</code> if response has expired and should be terminated.
      */
     public boolean hasExpired() {
-        long now = millisTime();
-        return hasConnectionIdleTimedOut(now) || hasRequestTimedOut(now);
+        return requestTimeoutReached || idleConnectionTimeoutReached;
     }
 
-    public boolean hasConnectionIdleTimedOut(long now) {
-        return config.getIdleConnectionTimeoutInMs() != -1 && (now - touch.get()) >= config.getIdleConnectionTimeoutInMs();
+    public void setRequestTimeoutReached() {
+        this.requestTimeoutReached = true;
     }
 
-    public boolean hasRequestTimedOut(long now) {
-        return requestTimeoutInMs != -1 && (now - start) >= requestTimeoutInMs;
+    public boolean isRequestTimeoutReached() {
+        return requestTimeoutReached;
+    }
+
+    public void setIdleConnectionTimeoutReached() {
+        this.idleConnectionTimeoutReached = true;
+    }
+
+    public boolean isIdleConnectionTimeoutReached() {
+        return idleConnectionTimeoutReached;
     }
 
     @Override
     public V get() throws InterruptedException, ExecutionException {
         try {
             return get(requestTimeoutInMs, TimeUnit.MILLISECONDS);
         } catch (TimeoutException e) {
-            cancelReaper();
+            cancelTimeouts();
             throw new ExecutionException(e);
         }
     }
 
-    public void cancelReaper() {
-        if (reaperFuture != null) {
-            reaperFuture.cancel(false);
+    public void cancelTimeouts() {
+        if (timeoutsHolder != null) {
+            timeoutsHolder.cancel();
+            timeoutsHolder = null;
         }
     }
 
@@ -242,7 +251,7 @@ public V get(long l, TimeUnit tu) throws InterruptedException, TimeoutException,
                         }
                         throw new ExecutionException(te);
                     } finally {
-                        cancelReaper();
+                        cancelTimeouts();
                     }
                 }
             }
@@ -278,7 +287,7 @@ private V getContent() throws ExecutionException {
                         }
                         throw new RuntimeException(ex);
                     } finally {
-                        cancelReaper();
+                        cancelTimeouts();
                     }
                 }
             }
@@ -290,7 +299,7 @@ private V getContent() throws ExecutionException {
     public final void done() {
 
         try {
-            cancelReaper();
+            cancelTimeouts();
 
             if (exEx.get() != null) {
                 return;
@@ -311,7 +320,7 @@ public final void done() {
     }
 
     public final void abort(final Throwable t) {
-        cancelReaper();
+        cancelTimeouts();
 
         if (isDone.get() || isCancelled.get())
             return;
@@ -366,9 +375,8 @@ public int incrementAndGetCurrentRedirectCount() {
         return redirectCount.incrementAndGet();
     }
 
-    public void setReaperFuture(FutureReaper reaperFuture) {
-        cancelReaper();
-        this.reaperFuture = reaperFuture;
+    public void setTimeoutsHolder(TimeoutsHolder timeoutsHolder) {
+        this.timeoutsHolder = timeoutsHolder;
     }
 
     public boolean isInAuth() {
@@ -411,6 +419,10 @@ public void setStreamWasAlreadyConsumed(boolean streamWasAlreadyConsumed) {
     public void touch() {
         touch.set(millisTime());
     }
+    
+    public long getLastTouch() {
+        return touch.get();
+    }
 
     @Override
     public boolean getAndSetWriteHeaders(boolean writeHeaders) {
@@ -499,7 +511,7 @@ public String toString() {
                 ",\n\thttpHeaders=" + httpHeaders + //
                 ",\n\texEx=" + exEx + //
                 ",\n\tredirectCount=" + redirectCount + //
-                ",\n\treaperFuture=" + reaperFuture + //
+                ",\n\timeoutsHolder=" + timeoutsHolder + //
                 ",\n\tinAuth=" + inAuth + //
                 ",\n\tstatusReceived=" + statusReceived + //
                 ",\n\ttouch=" + touch + //