Reimplement Netty provider timeouts, backport 4d02959

Author: Stephane Landelle

src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java

@@ -41,6 +41,7 @@
 import com.ning.http.client.ProxyServer;
 import com.ning.http.client.Request;
 import com.ning.http.client.listenable.AbstractListenableFuture;
+import com.ning.http.client.providers.netty.timeout.TimeoutsHolder;
 
 /**
  * A {@link Future} that can be used to track when an asynchronous HTTP request has been fully processed.
@@ -70,7 +71,9 @@ enum STATE {
     private HttpResponse httpResponse;
     private final AtomicReference<ExecutionException> exEx = new AtomicReference<ExecutionException>();
     private final AtomicInteger redirectCount = new AtomicInteger();
-    private volatile Future<?> reaperFuture;
+    private volatile boolean requestTimeoutReached;
+    private volatile boolean idleConnectionTimeoutReached;
+    private volatile TimeoutsHolder timeoutsHolder;
     private final AtomicBoolean inAuth = new AtomicBoolean(false);
     private final AtomicBoolean statusReceived = new AtomicBoolean(false);
     private final AtomicLong touch = new AtomicLong(millisTime());
@@ -159,7 +162,7 @@ void setAsyncHandler(AsyncHandler<V> asyncHandler) {
      */
     /* @Override */
     public boolean cancel(boolean force) {
-        cancelReaper();
+        cancelTimeouts();
 
         if (isCancelled.get())
             return false;
@@ -189,16 +192,23 @@ 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 idleConnectionTimeoutInMs != -1 && (now - touch.get()) >= idleConnectionTimeoutInMs;
+    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;
     }
 
     /**
@@ -209,14 +219,15 @@ public V get() throws InterruptedException, ExecutionException {
         try {
             return get(requestTimeoutInMs, TimeUnit.MILLISECONDS);
         } catch (TimeoutException e) {
-            cancelReaper();
+            cancelTimeouts();
             throw new ExecutionException(e);
         }
     }
 
-    void cancelReaper() {
-        if (reaperFuture != null) {
-            reaperFuture.cancel(false);
+    public void cancelTimeouts() {
+        if (timeoutsHolder != null) {
+            timeoutsHolder.cancel();
+            timeoutsHolder = null;
         }
     }
 
@@ -251,7 +262,7 @@ public V get(long l, TimeUnit tu) throws InterruptedException, TimeoutException,
                         }
                         throw new ExecutionException(te);
                     } finally {
-                        cancelReaper();
+                        cancelTimeouts();
                     }
                 }
             }
@@ -287,7 +298,7 @@ V getContent() throws ExecutionException {
                         }
                         throw new RuntimeException(ex);
                     } finally {
-                        cancelReaper();
+                        cancelTimeouts();
                     }
                 }
             }
@@ -299,7 +310,7 @@ V getContent() throws ExecutionException {
     public final void done() {
 
         try {
-            cancelReaper();
+            cancelTimeouts();
 
             if (exEx.get() != null) {
                 return;
@@ -320,7 +331,7 @@ public final void done() {
     }
 
     public final void abort(final Throwable t) {
-        cancelReaper();
+        cancelTimeouts();
 
         if (isDone.get() || isCancelled.get())
             return;
@@ -379,11 +390,6 @@ protected int incrementAndGetCurrentRedirectCount() {
         return redirectCount.incrementAndGet();
     }
 
-    protected void setReaperFuture(Future<?> reaperFuture) {
-        cancelReaper();
-        this.reaperFuture = reaperFuture;
-    }
-
     protected boolean isInAuth() {
         return inAuth.get();
     }
@@ -407,15 +413,17 @@ public boolean getAndSetStatusReceived(boolean sr) {
     /**
      * {@inheritDoc}
      */
-    /* @Override */
     public void touch() {
         touch.set(millisTime());
     }
 
+    public long getLastTouch() {
+        return touch.get();
+    }
+
     /**
      * {@inheritDoc}
      */
-    /* @Override */
     public boolean getAndSetWriteHeaders(boolean writeHeaders) {
         boolean b = this.writeHeaders;
         this.writeHeaders = writeHeaders;
@@ -425,7 +433,6 @@ public boolean getAndSetWriteHeaders(boolean writeHeaders) {
     /**
      * {@inheritDoc}
      */
-    /* @Override */
     public boolean getAndSetWriteBody(boolean writeBody) {
         boolean b = this.writeBody;
         this.writeBody = writeBody;
@@ -512,7 +519,7 @@ public String toString() {
                 ",\n\thttpResponse=" + httpResponse + //
                 ",\n\texEx=" + exEx + //
                 ",\n\tredirectCount=" + redirectCount + //
-                ",\n\treaperFuture=" + reaperFuture + //
+                ",\n\ttimeoutsHolder=" + timeoutsHolder + //
                 ",\n\tinAuth=" + inAuth + //
                 ",\n\tstatusReceived=" + statusReceived + //
                 ",\n\ttouch=" + touch + //

src/main/java/com/ning/http/client/providers/netty/NettyResponseFuture.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2010-2013 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.http.client.providers.netty.timeout;
+
+import static com.ning.http.util.DateUtil.*;
+
+import org.jboss.netty.util.Timeout;
+
+import com.ning.http.client.providers.netty.NettyAsyncHttpProvider;
+import com.ning.http.client.providers.netty.NettyResponseFuture;
+
+public class IdleConnectionTimeoutTimerTask extends TimeoutTimerTask {
+
+    private final long idleConnectionTimeout;
+    private final long requestTimeoutInstant;
+
+    public IdleConnectionTimeoutTimerTask(NettyResponseFuture<?> nettyResponseFuture, NettyAsyncHttpProvider provider, TimeoutsHolder timeoutsHolder,
+            long requestTimeout, long idleConnectionTimeout) {
+        super(nettyResponseFuture, provider, timeoutsHolder);
+        this.idleConnectionTimeout = idleConnectionTimeout;
+        requestTimeoutInstant = requestTimeout >= 0 ? nettyResponseFuture.getStart() + requestTimeout : Long.MAX_VALUE;
+    }
+
+    public void run(Timeout timeout) throws Exception {
+        if (provider.isClose()) {
+            timeoutsHolder.cancel();
+            return;
+        }
+
+        if (!nettyResponseFuture.isDone() && !nettyResponseFuture.isCancelled()) {
+
+            long now = millisTime();
+
+            long currentIdleConnectionTimeoutInstant = idleConnectionTimeout - nettyResponseFuture.getLastTouch();
+            long durationBeforeCurrentIdleConnectionTimeout = currentIdleConnectionTimeoutInstant - now;
+
+            if (durationBeforeCurrentIdleConnectionTimeout <= 0L) {
+                // idleConnectionTimeout reached
+                long durationSinceLastTouch = now - nettyResponseFuture.getLastTouch();
+                expire("Connection reached idle timeout of " + idleConnectionTimeout + " ms after " + durationSinceLastTouch + " ms");
+                nettyResponseFuture.setIdleConnectionTimeoutReached();
+
+            } else if (currentIdleConnectionTimeoutInstant < requestTimeoutInstant) {
+                // reschedule
+                timeoutsHolder.idleConnectionTimeout = provider.newTimeoutInMs(this, durationBeforeCurrentIdleConnectionTimeout);
+
+            } else {
+                // otherwise, no need to reschedule: requestTimeout will happen sooner
+                timeoutsHolder.idleConnectionTimeout = null;
+            }
+
+        } else {
+            timeoutsHolder.cancel();
+        }
+    }
+}

src/main/java/com/ning/http/client/providers/netty/timeout/IdleConnectionTimeoutTimerTask.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2010-2013 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.http.client.providers.netty.timeout;
+
+import static com.ning.http.util.DateUtil.*;
+
+import org.jboss.netty.util.Timeout;
+
+import com.ning.http.client.providers.netty.NettyAsyncHttpProvider;
+import com.ning.http.client.providers.netty.NettyResponseFuture;
+
+public class RequestTimeoutTimerTask extends TimeoutTimerTask {
+
+    public RequestTimeoutTimerTask(NettyResponseFuture<?> nettyResponseFuture, NettyAsyncHttpProvider provider, TimeoutsHolder timeoutsHolder) {
+        super(nettyResponseFuture, provider, timeoutsHolder);
+    }
+
+    public void run(Timeout timeout) throws Exception {
+
+        // in any case, cancel possible idleConnectionTimeout
+        timeoutsHolder.cancel();
+
+        if (provider.isClose()) {
+            return;
+        }
+
+        if (!nettyResponseFuture.isDone() && !nettyResponseFuture.isCancelled()) {
+            expire("Request reached timeout of " + nettyResponseFuture.getRequestTimeoutInMs() + " ms after " + (millisTime() - nettyResponseFuture.getStart()) + " ms");
+            nettyResponseFuture.setRequestTimeoutReached();
+        }
+    }
+}

src/main/java/com/ning/http/client/providers/netty/timeout/RequestTimeoutTimerTask.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2010-2013 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.http.client.providers.netty.timeout;
+
+import java.util.concurrent.TimeoutException;
+
+import org.jboss.netty.util.TimerTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.ning.http.client.providers.netty.NettyAsyncHttpProvider;
+import com.ning.http.client.providers.netty.NettyResponseFuture;
+
+public abstract class TimeoutTimerTask implements TimerTask {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutTimerTask.class);
+
+    protected final NettyResponseFuture<?> nettyResponseFuture;
+    protected final NettyAsyncHttpProvider provider;
+    protected final TimeoutsHolder timeoutsHolder;
+
+    public TimeoutTimerTask(NettyResponseFuture<?> nettyResponseFuture, NettyAsyncHttpProvider provider, TimeoutsHolder timeoutsHolder) {
+        this.nettyResponseFuture = nettyResponseFuture;
+        this.provider = provider;
+        this.timeoutsHolder = timeoutsHolder;
+    }
+
+    protected void expire(String message) {
+        LOGGER.debug("{} for {}", message, nettyResponseFuture);
+        provider.abort(nettyResponseFuture, new TimeoutException(message));
+    }
+}

src/main/java/com/ning/http/client/providers/netty/timeout/TimeoutTimerTask.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2010-2013 Ning, Inc.
+ *
+ * Ning licenses this file to you 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 com.ning.http.client.providers.netty.timeout;
+
+import org.jboss.netty.util.Timeout;
+
+public class TimeoutsHolder {
+
+    public volatile Timeout requestTimeout;
+    public volatile Timeout idleConnectionTimeout;
+
+    public void cancel() {
+        if (requestTimeout != null) {
+            requestTimeout.cancel();
+            requestTimeout = null;
+        }
+        if (idleConnectionTimeout != null) {
+            idleConnectionTimeout.cancel();
+            idleConnectionTimeout = null;
+        }
+    }
+}

src/main/java/com/ning/http/client/providers/netty/timeout/TimeoutsHolder.java

@@ -35,6 +35,7 @@
 import com.ning.http.client.HttpResponseBodyPart;
 import com.ning.http.client.HttpResponseBodyPartsInputStream;
 import com.ning.http.client.Part;
+import com.ning.http.client.Request;
 import com.ning.http.client.StringPart;
 import com.ning.http.multipart.ByteArrayPartSource;
 import com.ning.http.multipart.MultipartRequestEntity;
@@ -548,4 +549,8 @@ public static void checkBodyParts(int statusCode, Collection<HttpResponseBodyPar
     public static String keepAliveHeaderValue(AsyncHttpClientConfig config) {
         return config.getAllowPoolingConnection() ? "keep-alive" : "close";
     }
+    
+    public static int requestTimeout(AsyncHttpClientConfig config, Request request) {
+        return (request.getPerRequestConfig() != null && request.getPerRequestConfig().getRequestTimeoutInMs() != 0) ? request.getPerRequestConfig().getRequestTimeoutInMs() : config.getRequestTimeoutInMs();
+    }
 }

src/main/java/com/ning/http/util/AsyncHttpProviderUtils.java

@@ -12,7 +12,7 @@
  */
 package com.ning.http.client.async.netty;
 
-import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.*;
 
 import com.ning.http.client.AsyncHttpClient;
 import com.ning.http.client.AsyncHttpClientConfig;
@@ -22,7 +22,7 @@
 public class NettyPerRequestTimeoutTest extends PerRequestTimeoutTest {
 
     protected void checkTimeoutMessage(String message) {
-        assertTrue(message.startsWith("Request reached time out of 100 ms after "));
+        assertTrue(message.startsWith("Request reached timeout of 100 ms after "));
     }
 
     @Override