Remove Future.Computation and streamline task execution/cancellation (#3107)

This change removes the internal `Computation` indirection in `FutureImpl` and replaces it with a simplified, clearer execution model based directly on `Task`. 

Internalizing various execution modes inside `FutureImpl` enabled better cancellation handling in cases when futures were cancelled before running.

Author: pivovarit

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

@@ -22,7 +22,13 @@
 import io.vavr.control.Option;
 import io.vavr.control.Try;
 import java.util.Objects;
-import java.util.concurrent.*;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.LockSupport;
 import java.util.concurrent.locks.ReentrantLock;
@@ -32,7 +38,7 @@
  * <strong>INTERNAL API - This class is subject to change.</strong>
  *
  * @param <T> Result of the computation.
- * @author Daniel Dietrich
+ * @author Daniel Dietrich, Grzegorz Piwowarek
  */
 @SuppressWarnings("deprecation")
 final class FutureImpl<T> implements Future<T> {
@@ -84,9 +90,8 @@ final class FutureImpl<T> implements Future<T> {
      */
     private Thread thread;
 
-
     // single constructor
-    private FutureImpl(Executor executor, Option<Try<T>> value, Queue<Consumer<Try<T>>> actions, Queue<Thread> waiters, Computation<T> computation) {
+    private FutureImpl(Executor executor, Option<Try<T>> value, Queue<Consumer<Try<T>>> actions, Queue<Thread> waiters, Task<? extends T> task) {
         this.lock = new ReentrantLock();
         this.executor = executor;
         lock.lock();
@@ -96,7 +101,22 @@ 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);
+                if (task == null) {
+                    // no need to do anything
+                } else if (task instanceof Task.SyncTask) {
+                    task.run(this::tryComplete);
+                } else {
+                    executor.execute(() -> {
+                        if (!isCancelled()) {
+                            updateThread();
+                            try {
+                                task.run(this::tryComplete);
+                            } catch (Throwable x) {
+                                tryComplete(Try.failure(x));
+                            }
+                        }
+                    });
+                }
             } catch (Throwable x) {
                 tryComplete(Try.failure(x));
             }
@@ -113,7 +133,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(), null);
     }
 
     /**
@@ -125,7 +145,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, null);
     }
 
     /**
@@ -138,9 +158,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) ->
-            task.run(complete::with)
-        );
+        return new FutureImpl<>(executor, Option.none(), Queue.empty(), Queue.empty(), (Task.SyncTask<T>) complete -> task.run(complete::with));
     }
 
     /**
@@ -154,16 +172,7 @@ 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) ->
-                executor.execute(() -> {
-                    updateThread.run();
-                    try {
-                        task.run(complete::with);
-                    } catch (Throwable x) {
-                        complete.with(Try.failure(x));
-                    }
-                })
-        );
+        return new FutureImpl<>(executor, Option.none(), Queue.empty(), Queue.empty(), task);
     }
 
     @Override
@@ -425,8 +434,4 @@ private void unlock(Thread waiter) {
     private void handleUncaughtException(Throwable x) {
         tryComplete(Try.failure(x));
     }
-
-    private interface Computation<T> {
-        void execute(Task.Complete<T> complete, Runnable updateThread) throws Throwable;
-    }
 }

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

@@ -29,6 +29,10 @@ public interface Task<T> {
      */
     void run(Complete<T> complete) throws Throwable;
 
+    @FunctionalInterface
+    interface SyncTask<T> extends Task<T> {
+    }
+
     /**
      * Completes a task.
      *

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

@@ -40,6 +40,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.SynchronousQueue;
@@ -50,6 +51,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import org.assertj.core.api.IterableAssert;
+import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.AfterEachCallback;
 import org.junit.jupiter.api.extension.BeforeEachCallback;
@@ -59,6 +61,7 @@
 import static io.vavr.concurrent.Concurrent.waitUntil;
 import static io.vavr.concurrent.Concurrent.zZz;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 @SuppressWarnings("deprecation")
@@ -112,7 +115,8 @@ protected <T> IterableAssert<T> assertThat(Iterable<T> actual) {
             @Override
             public IterableAssert<T> isEqualTo(Object expected) {
                 if (actual instanceof Future && expected instanceof Future) {
-                    FutureTest.super.assertThat(((Future<T>) actual).getValue()).isEqualTo(((Future<T>) expected).getValue());
+                    FutureTest.super.assertThat(((Future<T>) actual).getValue())
+                      .isEqualTo(((Future<T>) expected).getValue());
                     return this;
                 } else {
                     return super.isEqualTo(expected);
@@ -487,7 +491,7 @@ public void shouldReduceSequenceOfFutures() {
     public void shouldReduceWithErrorIfSequenceOfFuturesContainsOneError() {
         final Future<Integer> future = Future.reduce(
           List.of(Future.of(zZz(13)), Future.of(zZz(new Error()))),
-          (i1, i2) -> i1 + i2
+          Integer::sum
         ).await();
         assertFailed(future, Error.class);
     }
@@ -692,46 +696,29 @@ public void shouldFallbackToThisFailure() {
     @Test
     public void shouldFoldEmptyIterable() {
         final Seq<Future<Integer>> futures = Stream.empty();
-        final Future<Integer> testee = Future.fold(futures, 0, (a, b) -> a + b).await();
+        final Future<Integer> testee = Future.fold(futures, 0, Integer::sum).await();
         assertThat(testee.getValue().get()).isEqualTo(Try.success(0));
     }
 
     @Test
     public void shouldFoldNonEmptyIterableOfSucceedingFutures() {
         final Seq<Future<Integer>> futures = Stream.from(1).map(i -> Future.of(zZz(i))).take(5);
-        final Future<Integer> testee = Future.fold(futures, 0, (a, b) -> a + b).await();
+        final Future<Integer> testee = Future.fold(futures, 0, Integer::sum).await();
         assertThat(testee.getValue().get()).isEqualTo(Try.success(15));
     }
 
     @Test
     public void shouldFoldNonEmptyIterableOfFailingFutures() {
         final Seq<Future<Integer>> futures = Stream.from(1).map(i -> Future.<Integer>of(zZz(new Error()))).take(5);
-        final Future<Integer> testee = Future.fold(futures, 0, (a, b) -> a + b).await();
+        final Future<Integer> testee = Future.fold(futures, 0, Integer::sum).await();
         assertFailed(testee, Error.class);
     }
 
-    // -- cancel()
-
-    @Test
-    public void shouldInterruptLockedFuture() {
-        final Object monitor = new Object();
-        final AtomicBoolean running = new AtomicBoolean(false);
-        final Future<?> future = blocking(() -> {
-            synchronized (monitor) {
-                running.set(true);
-                monitor.wait(); // wait forever
-            }
-        });
-        waitUntil(running::get);
-        synchronized (monitor) {
-            future.cancel();
-        }
-        assertThat(future.isCancelled()).isTrue();
-    }
-
-    @Test
-    public void shouldThrowOnGetAfterCancellation() {
-        assertThrows(CancellationException.class, () -> {
+    @Nested
+    public class CancellationTests {
+        
+        @Test
+        public void shouldInterruptLockedFuture() {
             final Object monitor = new Object();
             final AtomicBoolean running = new AtomicBoolean(false);
             final Future<?> future = blocking(() -> {
@@ -745,21 +732,67 @@ public void shouldThrowOnGetAfterCancellation() {
                 future.cancel();
             }
             assertThat(future.isCancelled()).isTrue();
-            future.get();
-        });
-    }
+        }
 
-    @Test
-    public void shouldCancelFutureThatNeverCompletes() {
-        @SuppressWarnings("deprecation") final Future<?> future = Future.run(complete -> {
-            // we break our promise, the Future is never completed
-        });
+        @Test
+        public void shouldThrowOnGetAfterCancellation() {
+            assertThrows(CancellationException.class, () -> {
+                final Object monitor = new Object();
+                final AtomicBoolean running = new AtomicBoolean(false);
+                final Future<?> future = blocking(() -> {
+                    synchronized (monitor) {
+                        running.set(true);
+                        monitor.wait(); // wait forever
+                    }
+                });
+                waitUntil(running::get);
+                synchronized (monitor) {
+                    future.cancel();
+                }
+                assertThat(future.isCancelled()).isTrue();
+                future.get();
+            });
+        }
 
-        assertThat(future.isCompleted()).isFalse();
-        assertThat(future.isCancelled()).isFalse();
+        @Test
+        public void shouldCancelFutureThatNeverCompletes() {
+            @SuppressWarnings("deprecation") final Future<?> future = Future.run(complete -> {
+                // we break our promise, the Future is never completed
+            });
 
-        assertThat(future.cancel()).isTrue();
-        assertThat(future.isCompleted()).isTrue();
+            assertThat(future.isCompleted()).isFalse();
+            assertThat(future.isCancelled()).isFalse();
+
+            assertThat(future.cancel()).isTrue();
+            assertThat(future.isCompleted()).isTrue();
+        }
+
+        @Test
+        void shouldNotRunCancelledFuture() {
+            ExecutorService es = Executors.newSingleThreadExecutor();
+
+            AtomicBoolean f1Executed = new AtomicBoolean(false);
+            AtomicBoolean f2Executed = new AtomicBoolean(false);
+
+            Future<Void> f1 = Future.run(es, () -> {Thread.sleep(10000);f1Executed.set(true);});
+            Future<Void> f2 = Future.run(es, () -> {f2Executed.set(true);Thread.sleep(10000);});
+
+            f2.cancel(true);
+            f1.cancel(true);
+
+            f1.await();
+            f2.await();
+
+            assertThat(f1.isCancelled()).isTrue();
+            assertThat(f2.isCancelled()).isTrue();
+
+            assertThat(f1.isFailure()).isTrue();
+            assertThat(f2.isFailure()).isTrue();
+            assertThatThrownBy(f1::get).isInstanceOf(CancellationException.class);
+            assertThatThrownBy(f2::get).isInstanceOf(CancellationException.class);
+            assertThat(f1Executed.get()).isFalse();
+            assertThat(f2Executed.get()).isFalse();
+        }
     }
 
     // -- collect()