Fix onTimeout / orTimeout handling for single-threaded executors; simplify CallbackRegistry; fix toCompletableFuture completion in wrappers

Author: vsilaev

pom.xml

@@ -6,7 +6,7 @@
 
 	<groupId>net.tascalate</groupId>
 	<artifactId>net.tascalate.concurrent</artifactId>
-	<version>0.9.3</version>
+	<version>0.9.4-SNAPSHOT</version>
 	<packaging>jar</packaging>  
 
 	<name>Tascalate Concurrent</name>

src/main/java/net/tascalate/concurrent/AbstractCompletableTask.java

@@ -144,7 +144,7 @@ public String toString() {
     @Override
     public <U> Promise<U> thenApplyAsync(Function<? super T, ? extends U> fn, Executor executor) {
         return addCallbacks(
-            internalCreateCompletionStage(executor), 
+            newSubTask(executor), 
             r -> {
                 try {
                     return fn.apply(r);
@@ -161,7 +161,7 @@ public <U> Promise<U> thenApplyAsync(Function<? super T, ? extends U> fn, Execut
     public Promise<T> exceptionallyAsync(Function<Throwable, ? extends T> fn, Executor executor) {
         // Symmetrical with thenApplyAsync
         return addCallbacks(
-            internalCreateCompletionStage(executor), 
+            newSubTask(executor), 
             Function.identity(), 
             failure -> {
                 try {
@@ -180,7 +180,7 @@ public Promise<T> exceptionallyAsync(Function<Throwable, ? extends T> fn, Execut
     @Override
     public <U> Promise<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
         return addCallbacks(
-            internalCreateCompletionStage(executor), 
+            newSubTask(executor), 
             result -> {
                 try {
                     return fn.apply(result, null);
@@ -211,8 +211,8 @@ public <U> Promise<U> handleAsync(BiFunction<? super T, Throwable, ? extends U>
     @Override
     public <U> Promise<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor) {
 
-        AbstractCompletableTask<Void> tempStage = internalCreateCompletionStage(executor);
-        AbstractCompletableTask<U> nextStage = internalCreateCompletionStage(executor);
+        AbstractCompletableTask<Void> tempStage = newSubTask(executor);
+        AbstractCompletableTask<U> nextStage = newSubTask(executor);
         // Need to enlist tempStage while it is non-visible outside
         // and may not be used to interrupt fn.apply();
         nextStage.intermediateStage = tempStage;
@@ -285,8 +285,8 @@ public <U> Promise<U> thenComposeAsync(Function<? super T, ? extends CompletionS
     public Promise<T> exceptionallyComposeAsync(Function<Throwable, ? extends CompletionStage<T>> fn, Executor executor) {
         // Symmetrical with thenComposeAsync
         // See comments for thenComposeAsync -- all are valid here, this is just a different path in Either (left vs right)
-        AbstractCompletableTask<Void> tempStage = internalCreateCompletionStage(executor);
-        AbstractCompletableTask<T> nextStage = internalCreateCompletionStage(executor);
+        AbstractCompletableTask<Void> tempStage = newSubTask(executor);
+        AbstractCompletableTask<T> nextStage = newSubTask(executor);
 
         nextStage.intermediateStage = tempStage;
 
@@ -330,7 +330,7 @@ public Promise<T> exceptionallyComposeAsync(Function<Throwable, ? extends Comple
     @Override
     public Promise<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) {
         return addCallbacks(
-            internalCreateCompletionStage(executor), 
+            newSubTask(executor), 
             result -> {
                 try {
                     action.accept(result, null);
@@ -363,26 +363,9 @@ public Promise<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> act
     @Override
     public CompletableFuture<T> toCompletableFuture() {
         CompletableFuture<T> result = new CompletableFuture<>();
-        // nextStage is CompletableFuture rather than AbstractCompletableTask
-        // so trigger completion on ad-hoc runnable rather than on
-        // nextStage.task
-        Consumer<Callable<T>> setup = c -> {
-            try {
-                c.call();
-            } catch (Throwable ex) {
-                result.completeExceptionally(ex);
-            }
-        };
-        addCallbacks(
-            setup, 
-            r -> op(result.complete(r)),
-            e -> op(result.completeExceptionally(e)), 
-            SAME_THREAD_EXECUTOR
-        );
+        whenComplete((r, e) -> iif(e == null ? result.complete(r) : result.completeExceptionally(e)));
         return result;
     }
-    
-    private static <T> T op(boolean v) { return null; }
 
     /**
      * This method exists just to reconcile generics when called from
@@ -395,7 +378,7 @@ protected <R, U> Promise<U> doApplyToEitherAsync(CompletionStage<? extends R> fi
                                                      Function<? super R, U> fn, 
                                                      Executor executor) {
 
-        AbstractCompletableTask<R> nextStage = internalCreateCompletionStage(executor);
+        AbstractCompletableTask<R> nextStage = newSubTask(executor);
 
         // Next stage is not exposed to the client, so we can
         // short-circuit its initiation - just fire callbacks
@@ -416,8 +399,8 @@ protected <R, U> Promise<U> doApplyToEitherAsync(CompletionStage<? extends R> fi
     }
 
     abstract protected <U> AbstractCompletableTask<U> createCompletionStage(Executor executor);
-
-    private <U> AbstractCompletableTask<U> internalCreateCompletionStage(Executor executor) {
+    
+    private <U> AbstractCompletableTask<U> newSubTask(Executor executor) {
         // Preserve default async executor, or use user-supplied executor as default
         // But don't let SAME_THREAD_EXECUTOR to be a default async executor
         return createCompletionStage(executor == SAME_THREAD_EXECUTOR ? getDefaultExecutor() : executor);
@@ -450,16 +433,8 @@ private <U> AbstractCompletableTask<U> addCallbacks(AbstractCompletableTask<U> t
                                                         Function<Throwable, ? extends U> failureCallback,
                                                         Executor executor) {
 
-        addCallbacks(targetStage::fireTransition, successCallback, failureCallback, executor);
+        callbackRegistry.addCallbacks(targetStage, successCallback, failureCallback, executor);
         return targetStage;
     }
 
-    private <U> void addCallbacks(Consumer<? super Callable<U>> stageTransition,
-                                  Function<? super T, ? extends U> successCallback, 
-                                  Function<Throwable, ? extends U> failureCallback,
-                                  Executor executor) {
-        
-        callbackRegistry.addCallbacks(stageTransition, successCallback, failureCallback, executor);
-    }
-
 }

src/main/java/net/tascalate/concurrent/CallbackRegistry.java

@@ -22,25 +22,24 @@
  */
 package net.tascalate.concurrent;
 
+import java.util.Collection;
 import java.util.LinkedList;
 import java.util.Objects;
-import java.util.Queue;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
 import java.util.concurrent.RejectedExecutionException;
-import java.util.function.Consumer;
+import java.util.function.BiConsumer;
 import java.util.function.Function;
 
 class CallbackRegistry<T> {
-    private State<T> state = InitialState.instance();
-
     private final Object mutex = new Object();
-
+    private State<T> state = InitialState.instance();
+    
     /**
      * Adds the given callbacks to this registry.
      */
-    <U> void addCallbacks(Consumer<? super Callable<U>> stageTransition,
+    <U> void addCallbacks(AbstractCompletableTask<U> target,
                           Function<? super T, ? extends U> successCallback, 
                           Function<Throwable, ? extends U> failureCallback,
                           Executor executor) {
@@ -49,11 +48,8 @@ <U> void addCallbacks(Consumer<? super Callable<U>> stageTransition,
         Objects.requireNonNull(failureCallback, "'failureCallback' must not be null");
         Objects.requireNonNull(executor, "'executor' must not be null");
 
-        @SuppressWarnings("unchecked")
-        Consumer<? super Callable<?>> typedTransition = (Consumer<? super Callable<?>>)stageTransition;
-
         synchronized (mutex) {
-            state = state.addCallbacks(typedTransition, successCallback, failureCallback, executor);
+            state = state.addCallbacks(target, successCallback, failureCallback, executor);
         }
     }
 
@@ -114,11 +110,11 @@ boolean isCompleted() {
      * State of the registry. All subclasses are meant to be used form a
      * synchronized block and are NOT thread safe on their own.
      */
-    private static abstract class State<S> {
-        protected abstract State<S> addCallbacks(Consumer<? super Callable<?>> stageTransition,
-                                                 Function<? super S, ?> successCallback, 
-                                                 Function<Throwable, ?> failureCallback, 
-                                                 Executor executor);
+    static abstract class State<S> {
+        protected abstract <U> State<S> addCallbacks(AbstractCompletableTask<U> target,
+                                                     Function<? super S, ? extends U> successCallback, 
+                                                     Function<Throwable, ? extends U> failureCallback, 
+                                                     Executor executor);
 
         protected State<S> getSuccessState(S result) {
             throw new IllegalStateException("success method should not be called multiple times");
@@ -147,17 +143,17 @@ protected boolean isFailure() {
      * Result is not known yet and no callbacks registered. Using shared
      * instance so we do not allocate instance where it may not be needed.
      */
-    private static class InitialState<S> extends State<S> {
+    static class InitialState<S> extends State<S> {
         private static final InitialState<Object> instance = new InitialState<>();
 
         @Override
-        protected State<S> addCallbacks(Consumer<? super Callable<?>> stageTransition,
-                                        Function<? super S, ?> successCallback, 
-                                        Function<Throwable, ?> failureCallback, 
-                                        Executor executor) {
+        protected <U> State<S> addCallbacks(AbstractCompletableTask<U> target,
+                                            Function<? super S, ? extends U> successCallback, 
+                                            Function<Throwable, ? extends U> failureCallback, 
+                                            Executor executor) {
 
             IntermediateState<S> intermediateState = new IntermediateState<>();
-            intermediateState.addCallbacks(stageTransition, successCallback, failureCallback, executor);
+            intermediateState.addCallbacks(target, successCallback, failureCallback, executor);
             return intermediateState;
         }
 
@@ -177,24 +173,30 @@ protected boolean isCompleted() {
         }
 
         @SuppressWarnings("unchecked")
-        private static <T> State<T> instance() {
+        static <T> State<T> instance() {
             return (State<T>) instance;
         }
     }
 
     /**
      * Result is not known yet.
      */
-    private static class IntermediateState<S> extends State<S> {
-        private final Queue<CallbackHolder<? super S>> callbacks = new LinkedList<>();
+    static class IntermediateState<S> extends State<S> {
+        private final Collection<BiConsumer<? super S, ? super Throwable>> callbacks = new LinkedList<>();
 
         @Override
-        protected State<S> addCallbacks(Consumer<? super Callable<?>> stageTransition,
-                                        Function<? super S, ?> successCallback, 
-                                        Function<Throwable, ?> failureCallback, 
-                                        Executor executor) {
+        protected <U> State<S> addCallbacks(AbstractCompletableTask<U> target,
+                                            Function<? super S, ? extends U> successCallback, 
+                                            Function<Throwable, ? extends U> failureCallback, 
+                                            Executor executor) {
 
-            callbacks.add(new CallbackHolder<>(stageTransition, successCallback, failureCallback, executor));
+            callbacks.add((r, e) -> {
+                if (null == e) {
+                    callCallback(target, successCallback, r, executor);
+                } else {
+                    callCallback(target, failureCallback, e, executor);
+                }
+            });
             return this;
         }
 
@@ -205,10 +207,8 @@ protected State<S> getSuccessState(S result) {
 
         @Override
         protected void callSuccessCallbacks(S result) {
-            // no need to remove callbacks from the queue, this instance will be
-            // thrown away at once
-            for (CallbackHolder<? super S> callback : callbacks) {
-                callback.callSuccessCallback(result);
+            for (BiConsumer<? super S, ? super Throwable> callback : callbacks) {
+                callback.accept(result, null);
             }
         }
 
@@ -219,10 +219,8 @@ protected State<S> getFailureState(Throwable failure) {
 
         @Override
         protected void callFailureCallbacks(Throwable failure) {
-            // no need to remove callbacks from the queue, this instance will be
-            // thrown away at once
-            for (CallbackHolder<? super S> callback : callbacks) {
-                callback.callFailureCallback(failure);
+            for (BiConsumer<? super S, ? super Throwable> callback : callbacks) {
+                callback.accept(null, failure);
             }
         }
 
@@ -235,41 +233,39 @@ protected boolean isCompleted() {
     /**
      * Holds the result.
      */
-    private static final class SuccessState<S> extends State<S> {
+    static final class SuccessState<S> extends State<S> {
         private final S result;
 
-        private SuccessState(S result) {
+        SuccessState(S result) {
             this.result = result;
         }
 
         @Override
-        protected State<S> addCallbacks(Consumer<? super Callable<?>> stageTransition,
-                                        Function<? super S, ?> successCallback, 
-                                        Function<Throwable, ?> failureCallback, 
-                                        Executor executor) {
-            
-            callCallback(stageTransition, successCallback, result, executor);
+        protected <U> State<S> addCallbacks(AbstractCompletableTask<U> target,
+                                            Function<? super S, ? extends U> successCallback, 
+                                            Function<Throwable, ? extends U> failureCallback, 
+                                            Executor executor) {
+            callCallback(target, successCallback, result, executor);
             return this;
         }
     }
 
     /**
      * Holds the failure.
      */
-    private static final class FailureState<S> extends State<S> {
+    static final class FailureState<S> extends State<S> {
         private final Throwable failure;
 
-        private FailureState(Throwable failure) {
+        FailureState(Throwable failure) {
             this.failure = failure;
         }
 
         @Override
-        protected State<S> addCallbacks(Consumer<? super Callable<?>> stageTransition,
-                                        Function<? super S, ?> successCallback, 
-                                        Function<Throwable, ?> failureCallback, 
-                                        Executor executor) {
-            
-            callCallback(stageTransition, failureCallback, failure, executor);
+        protected <U> State<S> addCallbacks(AbstractCompletableTask<U> target,
+                                            Function<? super S, ? extends U> successCallback, 
+                                            Function<Throwable, ? extends U> failureCallback, 
+                                            Executor executor) {
+            callCallback(target, failureCallback, failure, executor);
             return this;
         }
 
@@ -278,47 +274,21 @@ protected boolean isFailure() {
         }
     }
 
-    private static final class CallbackHolder<S> {
-        private final Consumer<? super Callable<?>> stageTransition;
-        private final Function<? super S, ?> successCallback;
-        private final Function<Throwable, ?> failureCallback;
-        private final Executor executor;
-
-        private CallbackHolder(Consumer<? super Callable<?>> stageTransition,
-                               Function<? super S, ?> successCallback, 
-                               Function<Throwable, ?> failureCallback, 
-                               Executor executor) {
-
-            this.stageTransition = stageTransition;
-            this.successCallback = successCallback;
-            this.failureCallback = failureCallback;
-            this.executor = executor;
-        }
-
-        void callSuccessCallback(S result) {
-            callCallback(stageTransition, successCallback, result, executor);
-        }
-
-        void callFailureCallback(Throwable failure) {
-            callCallback(stageTransition, failureCallback, failure, executor);
-        }
-    }
-
-    private static <S, U> void callCallback(Consumer<? super Callable<?>> stageTransition,
-                                            Function<? super S, ? extends U> callback, 
-                                            S value,  
-                                            Executor executor) {
+    static <T, U> void callCallback(AbstractCompletableTask<U> target,
+                                    Function<? super T, ? extends U> callback, 
+                                    T value,  
+                                    Executor executor) {
 
         Callable<U> callable = () -> callback.apply(value);
         try {
-            executor.execute( (AsyncTask)() -> stageTransition.accept(callable) );
+            executor.execute( (AsyncTask)() -> target.fireTransition(callable) );
         } catch (RejectedExecutionException ex) {
             // Propagate error in-place
             Callable<U> propagateError = () -> { throw ex; };
-            stageTransition.accept(propagateError);
+            target.fireTransition(propagateError);
         }
     }
-    
+
     @FunctionalInterface
     static interface AsyncTask extends Runnable, CompletableFuture.AsynchronousCompletionTask {}