Use field updaters for timeout guards

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

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

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

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<TimeoutsHolder> 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();

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<Object>() {
+            @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);
+    }
 }