Replace synchronization by Lock in FutureImpl

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

@@ -25,7 +25,9 @@
 
 import java.util.Objects;
 import java.util.concurrent.*;
+import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.LockSupport;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Consumer;
 
 /**
@@ -45,7 +47,7 @@ final class FutureImpl<T> implements Future<T> {
     /**
      * Used to synchronize state changes.
      */
-    private final Object lock = new Object();
+    private final Lock lock;
 
     /**
      * Indicates if this Future is cancelled
@@ -87,8 +89,10 @@ final class FutureImpl<T> implements Future<T> {
 
     // single constructor
     private FutureImpl(Executor executor, Option<Try<T>> value, Queue<Consumer<Try<T>>> actions, Queue<Thread> waiters, Computation<T> computation) {
+        this.lock = new ReentrantLock();
         this.executor = executor;
-        synchronized (lock) {
+        lock.lock();
+        try {
             this.cancelled = false;
             this.value = value;
             this.actions = actions;
@@ -98,6 +102,8 @@ private FutureImpl(Executor executor, Option<Try<T>> value, Queue<Consumer<Try<T
             } catch (Throwable x) {
                 tryComplete(Try.failure(x));
             }
+        } finally {
+            lock.unlock();
         }
     }
 
@@ -219,8 +225,11 @@ private void _await(long start, long timeout, TimeUnit unit) {
                 public boolean block() {
                     try {
                         if (!threadEnqueued) {
-                            synchronized (lock) {
+                            lock.lock();
+                            try {
                                 waiters = waiters.enqueue(waitingThread);
+                            } finally {
+                                lock.unlock();
                             }
                             threadEnqueued = true;
                         }
@@ -256,14 +265,17 @@ public boolean isReleasable() {
     @Override
     public boolean cancel(boolean mayInterruptIfRunning) {
         if (!isCompleted()) {
-            synchronized (lock) {
+            lock.lock();
+            try {
                 if (!isCompleted()) {
                     if (mayInterruptIfRunning && this.thread != null) {
                         this.thread.interrupt();
                     }
                     this.cancelled = tryComplete(Try.failure(new CancellationException()));
                     return this.cancelled;
                 }
+            } finally {
+                lock.unlock();
             }
         }
         return false;
@@ -272,7 +284,8 @@ public boolean cancel(boolean mayInterruptIfRunning) {
     private void updateThread() {
         // cancellation may have been initiated by a different thread before this.thread is set by the worker thread
         if (!isCompleted()) {
-            synchronized (lock) {
+            lock.lock();
+            try {
                 if (!isCompleted()) {
                     this.thread = Thread.currentThread();
                     try {
@@ -281,6 +294,8 @@ private void updateThread() {
                         // we are not allowed to set the uncaught exception handler of the worker thread ¯\_(ツ)_/¯
                     }
                 }
+            } finally {
+                lock.unlock();
             }
         }
     }
@@ -322,12 +337,15 @@ public Future<T> onComplete(Consumer<? super Try<T>> action) {
         if (isCompleted()) {
             perform(action);
         } else {
-            synchronized (lock) {
+            lock.lock();
+            try {
                 if (isCompleted()) {
                     perform(action);
                 } else {
                     actions = actions.enqueue((Consumer<Try<T>>) action);
                 }
+            } finally {
+                lock.unlock();
             }
         }
         return this;
@@ -362,7 +380,8 @@ boolean tryComplete(Try<? extends T> value) {
             final Queue<Consumer<Try<T>>> actions;
             final Queue<Thread> waiters;
             // it is essential to make the completed state public *before* performing the actions
-            synchronized (lock) {
+            lock.lock();
+            try {
                 if (isCompleted()) {
                     actions = null;
                     waiters = null;
@@ -374,6 +393,8 @@ boolean tryComplete(Try<? extends T> value) {
                     this.waiters = null;
                     this.thread = null;
                 }
+            } finally {
+                lock.unlock();
             }
             if (waiters != null) {
                 waiters.forEach(this::unlock);