diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java index 8b0d4373a1..d337b4505a 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java @@ -33,7 +33,7 @@ public class ReadTimeoutTimerTask extends TimeoutTimerTask { @Override public void run(Timeout timeout) { - if (done.getAndSet(true) || requestSender.isClosed()) { + if (!markDone() || requestSender.isClosed()) { return; } @@ -58,7 +58,7 @@ public void run(Timeout timeout) { timeoutsHolder.cancel(); } else { - done.set(false); + markPending(); timeoutsHolder.startReadTimeout(this); } } diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java index 74c5d0197a..61fa1ec0f1 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java @@ -36,7 +36,7 @@ public class RequestTimeoutTimerTask extends TimeoutTimerTask { @Override public void run(Timeout timeout) { - if (done.getAndSet(true) || requestSender.isClosed()) { + if (!markDone() || requestSender.isClosed()) { return; } diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java index b7e678fa84..3d0d627dbc 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java @@ -23,13 +23,16 @@ import java.net.InetSocketAddress; import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; public abstract class TimeoutTimerTask implements TimerTask { + private static final AtomicIntegerFieldUpdater DONE_UPDATER = + AtomicIntegerFieldUpdater.newUpdater(TimeoutTimerTask.class, "done"); private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutTimerTask.class); - protected final AtomicBoolean done = new AtomicBoolean(); + @SuppressWarnings("unused") + private volatile int done; protected final NettyRequestSender requestSender; final TimeoutsHolder timeoutsHolder; volatile NettyResponseFuture nettyResponseFuture; @@ -45,12 +48,20 @@ void expire(String message, long time) { requestSender.abort(nettyResponseFuture.channel(), nettyResponseFuture, new TimeoutException(message)); } + boolean markDone() { + return DONE_UPDATER.getAndSet(this, 1) == 0; + } + + void markPending() { + done = 0; + } + /** * When the timeout is cancelled, it could still be referenced for quite some time in the Timer. 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_UPDATER.compareAndSet(this, 0, 1)) { nettyResponseFuture = null; } } diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java index 93f6b26a26..c4f77ddd90 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -25,14 +25,18 @@ import java.net.InetSocketAddress; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime; public class TimeoutsHolder { + private static final AtomicIntegerFieldUpdater CANCELLED_UPDATER = + AtomicIntegerFieldUpdater.newUpdater(TimeoutsHolder.class, "cancelled"); + private final Timeout requestTimeout; - private final AtomicBoolean cancelled = new AtomicBoolean(); + @SuppressWarnings("unused") + private volatile int cancelled; private final Timer nettyTimer; private final NettyRequestSender requestSender; private final long requestTimeoutMillisTime; @@ -97,7 +101,7 @@ void startReadTimeout(ReadTimeoutTimerTask task) { } public void cancel() { - if (cancelled.compareAndSet(false, true)) { + if (CANCELLED_UPDATER.compareAndSet(this, 0, 1)) { if (requestTimeout != null) { requestTimeout.cancel(); ((TimeoutTimerTask) requestTimeout.task()).clean(); diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutTimerTaskTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutTimerTaskTest.java index 2a5f5e2059..6ff74def92 100644 --- a/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutTimerTaskTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutTimerTaskTest.java @@ -25,6 +25,7 @@ import java.net.InetSocketAddress; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; public class TimeoutTimerTaskTest { @@ -84,4 +85,30 @@ public void run(io.netty.util.Timeout timeout) { task.appendRemoteAddress(sb); assertTrue(sb.toString().contains(":8080"), sb.toString()); } + + @Test + public void cleanShouldClearFutureOnlyOnce() { + Request request = new RequestBuilder().setUrl("http://example.com:12345").build(); + NettyResponseFuture future = new NettyResponseFuture<>(request, new AsyncCompletionHandler() { + @Override + public Object onCompleted(org.asynchttpclient.Response response) throws Exception { + return null; + } + }, null, + 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null); + + TimeoutsHolder timeoutsHolder = new TimeoutsHolder(null, future, null, new DefaultAsyncHttpClientConfig.Builder().build(), null); + + TimeoutTimerTask task = new TimeoutTimerTask(future, null, timeoutsHolder) { + @Override + public void run(io.netty.util.Timeout timeout) { + // no-op + } + }; + + task.clean(); + task.clean(); + + assertNull(task.nettyResponseFuture); + } }