Not started Future is now cancellable (#2750)

Author: Egor Tsinko

src/main/java/io/vavr/concurrent/FutureImpl.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.*;
 import java.util.concurrent.locks.LockSupport;
 import java.util.function.Consumer;
+import java.util.function.Supplier;
 
 /**
  * <strong>INTERNAL API - This class is subject to change.</strong>
@@ -100,7 +101,12 @@ private FutureImpl(Executor executor, Option<Try<T>> value, Queue<Consumer<Try<T
             this.actions = actions;
             this.waiters = waiters;
             try {
-                computation.execute(this::tryComplete, this::updateThread);
+                computation.execute(this::tryComplete, this::updateThread, () -> {
+                    // Synchronize as the future could be in the process of cancelling
+                    synchronized (lock) {
+                        return isCompleted();
+                    }
+                });
             } catch (Throwable x) {
                 tryComplete(Try.failure(x));
             }
@@ -115,7 +121,7 @@ private FutureImpl(Executor executor, Option<Try<T>> value, Queue<Consumer<Try<T
      * @return a new {@code FutureImpl} instance
      */
     static <T> FutureImpl<T> of(Executor executor) {
-        return new FutureImpl<>(executor, Option.none(), Queue.empty(), Queue.empty(), (complete, updateThread) -> {});
+        return new FutureImpl<>(executor, Option.none(), Queue.empty(), Queue.empty(), (complete, updateThread, isCompleted) -> {});
     }
 
     /**
@@ -127,7 +133,7 @@ static <T> FutureImpl<T> of(Executor executor) {
      * @return a new {@code FutureImpl} instance
      */
     static <T> FutureImpl<T> of(Executor executor, Try<? extends T> value) {
-        return new FutureImpl<>(executor, Option.some(Try.narrow(value)), null, null, (complete, updateThread) -> {});
+        return new FutureImpl<>(executor, Option.some(Try.narrow(value)), null, null, (complete, updateThread, isCompleted) -> {});
     }
 
     /**
@@ -140,7 +146,7 @@ static <T> FutureImpl<T> of(Executor executor, Try<? extends T> value) {
      * @return a new {@code FutureImpl} instance
      */
     static <T> FutureImpl<T> sync(Executor executor, Task<? extends T> task) {
-        return new FutureImpl<>(executor, Option.none(), Queue.empty(), Queue.empty(), (complete, updateThread) ->
+        return new FutureImpl<>(executor, Option.none(), Queue.empty(), Queue.empty(), (complete, updateThread, isCompleted) ->
             task.run(complete::with)
         );
     }
@@ -156,8 +162,12 @@ static <T> FutureImpl<T> sync(Executor executor, Task<? extends T> task) {
      */
     static <T> FutureImpl<T> async(Executor executor, Task<? extends T> task) {
         // In a single-threaded context this Future may already have been completed during initialization.
-        return new FutureImpl<>(executor, Option.none(), Queue.empty(), Queue.empty(), (complete, updateThread) ->
+        return new FutureImpl<>(executor, Option.none(), Queue.empty(), Queue.empty(), (complete, updateThread, isCompleted) ->
                 executor.execute(() -> {
+                    // Avoid performing work, if future is already complete (normally by cancellation)
+                    if (isCompleted.get()) {
+                        return;
+                    }
                     updateThread.run();
                     try {
                         task.run(complete::with);
@@ -414,6 +424,6 @@ private void handleUncaughtException(Throwable x) {
     }
 
     private interface Computation<T> {
-        void execute(Task.Complete<T> complete, Runnable updateThread) throws Throwable;
+        void execute(Task.Complete<T> complete, Runnable updateThread, Supplier<Boolean> isCompleted) throws Throwable;
     }
 }

src/test/java/io/vavr/concurrent/FutureTest.java

@@ -53,6 +53,7 @@
 import static io.vavr.concurrent.Concurrent.waitUntil;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static io.vavr.concurrent.Concurrent.zZz;
+import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.assertj.core.api.Assertions.fail;
 
 @SuppressWarnings("deprecation")
@@ -143,6 +144,31 @@ protected int getPeekNonNilPerformingAnAction() {
 
     // -- static failed()
 
+
+    @Test
+    public void shouldNotExecuteFutureThatHasBeenCancelledBeforeItStarted() throws InterruptedException {
+        ExecutorService es = Executors.newSingleThreadExecutor();
+
+        AtomicBoolean future2Executed = new AtomicBoolean(false);
+
+        // Submit f1 to the executor first
+        Future<Void> f = Future.run(es, () -> Thread.sleep(1000));
+        // Submit f2 next, it will have to wait to be executed
+        Future<Void> f2 = Future.run(es, () -> {
+                    // Should never run this
+                    future2Executed.set(true);
+                });
+
+        // Cancel f2 BEFORE it runs on the executor
+        f2.cancel(true);
+        f.cancel(true);
+        es.shutdown();
+        boolean terminated = es.awaitTermination(2, SECONDS);
+        assertThat(terminated).isTrue();
+        // f2 should never have run
+        assertThat(future2Executed.get()).isFalse();
+    }
+
     @Test
     public void shouldCreateFailureThatFailsWithRuntimeException() {
         final Future<Object> failed = Future.failed(new RuntimeException("ooops")).await();