Future (#1286)

* simplify AbstractListenableFuture.executionList initialization

Just use CAS loop instead of synchronized and two boolean variables.

* Allow null executor in ListenableFuture

It allows execution of future callback in completion thread, and
makes `ListenableFuture` closer to `CompletableFuture` (which allows
`null` executor).

It also saves a little memory in `toCompletableFuture` implementation.

Author: stepancheg

client/src/main/java/org/asynchttpclient/ListenableFuture.java

@@ -67,6 +67,9 @@ public interface ListenableFuture<V> extends Future<V> {
      * to the executor} for execution when the {@code Future}'s computation is
      * {@linkplain Future#isDone() complete}.
      * <br>
+     * Executor can be <code>null</code>, in that case executor will be executed
+     * in the thread where completion happens.
+     * <br>
      * There is no guaranteed ordering of execution of listeners, they may get
      * called in the order they were added and they may get called out of order,
      * but any listener added through this method is guaranteed to be called once
@@ -131,7 +134,11 @@ public void touch() {
 
         @Override
         public ListenableFuture<T> addListener(Runnable listener, Executor exec) {
-            exec.execute(listener);
+            if (exec != null) {
+                exec.execute(listener);
+            } else {
+                listener.run();
+            }
             return this;
         }
 

client/src/main/java/org/asynchttpclient/future/AbstractListenableFuture.java

@@ -29,6 +29,7 @@
 package org.asynchttpclient.future;
 
 import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 
 import org.asynchttpclient.ListenableFuture;
 
@@ -43,41 +44,44 @@
  */
 public abstract class AbstractListenableFuture<V> implements ListenableFuture<V> {
 
-    private volatile boolean hasRun;
-    private volatile boolean executionListInitialized;
     private volatile ExecutionList executionList;
 
+    private static final AtomicReferenceFieldUpdater<AbstractListenableFuture, ExecutionList> executionListField =
+            AtomicReferenceFieldUpdater.newUpdater(AbstractListenableFuture.class, ExecutionList.class, "executionList");
+
     private ExecutionList executionList() {
-        ExecutionList localExecutionList = executionList;
-        if (localExecutionList == null) {
-            synchronized (this) {
-                localExecutionList = executionList;
-                if (localExecutionList == null) {
-                    localExecutionList = new ExecutionList();
-                    executionList = localExecutionList;
-                    executionListInitialized = true;
-                }
-            }
+        ExecutionList executionListLocal = this.executionList;
+        if (executionListLocal != null) {
+            return executionListLocal;
+        }
+
+        ExecutionList r = new ExecutionList();
+        if (executionListField.compareAndSet(this, null, r)) {
+            return r;
+        } else {
+            return this.executionList;
         }
-        return localExecutionList;
     }
 
     @Override
     public ListenableFuture<V> addListener(Runnable listener, Executor exec) {
-        executionList().add(listener, exec);
-        if (hasRun) {
-            runListeners();
+        if (executionList == null) {
+            ExecutionList.executeListener(listener, exec);
+            return this;
         }
+
+        executionList().add(listener, exec);
         return this;
     }
 
     /**
      * Execute the execution list.
      */
     protected void runListeners() {
-        hasRun = true;
-        if (executionListInitialized) {
-            executionList().execute();
+        if (executionList == null) {
+            return;
         }
+
+        executionList().execute();
     }
 }

client/src/main/java/org/asynchttpclient/future/ExecutionList.java

@@ -56,7 +56,6 @@ public void add(Runnable runnable, Executor executor) {
         // Fail fast on a null. We throw NPE here because the contract of Executor states that it
         // throws NPE on null listener, so we propagate that contract up into the add method as well.
         assertNotNull(runnable, "runnable");
-        assertNotNull(executor, "executor");
 
         // Lock while we check state. We must maintain the lock while adding the new pair so that
         // another thread can't run the list out from under us. We only add to the list if we have not
@@ -120,9 +119,13 @@ public void execute() {
     /**
      * Submits the given runnable to the given {@link Executor} catching and logging all {@linkplain RuntimeException runtime exceptions} thrown by the executor.
      */
-    private static void executeListener(Runnable runnable, Executor executor) {
+    static void executeListener(Runnable runnable, Executor executor) {
         try {
-            executor.execute(runnable);
+            if (executor != null) {
+                executor.execute(runnable);
+            } else {
+                runnable.run();
+            }
         } catch (RuntimeException e) {
             // Log it and keep going, bad runnable and/or executor. Don't punish the other runnables if
             // we're given a bad one. We only catch RuntimeException because we want Errors to propagate

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

@@ -271,12 +271,7 @@ public void run() {
                     completable.complete((V) CONTENT_UPDATER.get(NettyResponseFuture.this));
             }
 
-        }, new Executor() {
-            @Override
-            public void execute(Runnable command) {
-                command.run();
-            }
-        });
+        }, null);
 
         return completable;
     }