AsyncCompletions throws CompletionException (like CompletableFuture.join) instead of sneaky throw; less shared state in Promises.partitioned impl.; fix docs

Author: vsilaev

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

@@ -292,71 +292,71 @@ public static <T, A, R> Promise<R> partitioned(Iterable<? extends T> values,
                                                    int batchSize, 
                                                    Function<? super T, CompletionStage<? extends T>> spawner, 
                                                    Collector<T, A, R> downstream) {
-        return partitioned1(values.iterator(), batchSize, spawner, downstream);
+        return partitioned1(values.iterator(), null, batchSize, spawner, downstream);
     }
 
     public static <T, A, R> Promise<R> partitioned(Iterable<? extends T> values, 
                                                    int batchSize, 
                                                    Function<? super T, CompletionStage<? extends T>> spawner, 
                                                    Collector<T, A, R> downstream,
                                                    Executor downstreamExecutor) {
-        return partitioned2(values.iterator(), batchSize, spawner, downstream, downstreamExecutor);
+        return partitioned2(values.iterator(), null, batchSize, spawner, downstream, downstreamExecutor);
     }
 
     public static <T, A, R> Promise<R> partitioned(Stream<? extends T> values, 
                                                    int batchSize, 
                                                    Function<? super T, CompletionStage<? extends T>> spawner, 
                                                    Collector<T, A, R> downstream) {
-        return partitioned1(values.iterator(), batchSize, spawner, downstream);
+        return partitioned1(values.iterator(), values, batchSize, spawner, downstream);
     }
 
     public static <T, A, R> Promise<R> partitioned(Stream<? extends T> values, 
                                                    int batchSize, 
                                                    Function<? super T, CompletionStage<? extends T>> spawner, 
                                                    Collector<T, A, R> downstream,
                                                    Executor downstreamExecutor) {
-        return partitioned2(values.iterator(), batchSize, spawner, downstream, downstreamExecutor);
+        return partitioned2(values.iterator(), values, batchSize, spawner, downstream, downstreamExecutor);
     }
 
 
     private static <T, A, R> Promise<R> partitioned1(Iterator<? extends T> values, 
-                                                    int batchSize, 
-                                                    Function<? super T, CompletionStage<? extends T>> spawner, 
-                                                    Collector<T, A, R> downstream) {
+                                                     Object source,
+                                                     int batchSize, 
+                                                     Function<? super T, CompletionStage<? extends T>> spawner, 
+                                                     Collector<T, A, R> downstream) {
         return
             parallelStep1(values, batchSize, spawner, downstream)
             .dependent()
-            .thenApply(downstream.finisher(), true)
-            .as(onCloseSource(values))
+            .thenApply(downstream.finisher().compose(IndexedStep::payload), true)
+            .as(onCloseSource(null != source? source : values))
             .unwrap();
     }
 
 
     private static <T, A, R> Promise<R> partitioned2(Iterator<? extends T> values, 
-                                                    int batchSize, 
-                                                    Function<? super T, CompletionStage<? extends T>> spawner, 
-                                                    Collector<T, A, R> downstream,
-                                                    Executor downstreamExecutor) {
+                                                     Object source,
+                                                     int batchSize, 
+                                                     Function<? super T, CompletionStage<? extends T>> spawner, 
+                                                     Collector<T, A, R> downstream,
+                                                     Executor downstreamExecutor) {
         return 
             parallelStep2(values, batchSize, spawner, downstream, downstreamExecutor)
             .dependent()
-            .thenApplyAsync(downstream.finisher(), downstreamExecutor, true)
-            .as(onCloseSource(values))
+            .thenApplyAsync(downstream.finisher().compose(IndexedStep::payload), downstreamExecutor, true)
+            .as(onCloseSource(null != source? source : values))
             .unwrap();
     }
 
-    private static <T, A, R> Promise<A> parallelStep1(Iterator<? extends T> values, 
-                                                     int batchSize,
-                                                     Function<? super T, CompletionStage<? extends T>> spawner,                                                        
-                                                     Collector<T, A, R> downstream) {
+    private static <T, A, R> Promise<IndexedStep<A>> parallelStep1(
+        Iterator<? extends T> values, int batchSize,
+        Function<? super T, CompletionStage<? extends T>> spawner,                                                        
+        Collector<T, A, R> downstream) {
 
-        int[] step = new int[1];
-        return loop(null, __ -> step[0] == 0 || values.hasNext(), current -> {
+        return loop(new IndexedStep<>(), step -> step.initial() || values.hasNext(), step -> {
             List<T> valuesBatch = drainBatch(values, batchSize);
-            boolean initial = step[0]++ == 0;
             if (valuesBatch.isEmpty()) {
                 // Over
-                return Promises.success(initial ? downstream.supplier().get() : current);
+                return Promises.success(step.initial() ? step.next(downstream.supplier().get()) : step);
             } else {
                 List<CompletionStage<? extends T>> promisesBatch = 
                     valuesBatch.stream()
@@ -366,27 +366,27 @@ private static <T, A, R> Promise<A> parallelStep1(Iterator<? extends T> values,
                 return 
                 Promises.all(promisesBatch)
                         .dependent()
-                        .thenApply(vals -> accumulate(vals, initial, current, downstream), true);
+                        .thenApply(vals -> step.next(
+                            accumulate(vals, step.initial(), step.payload(), downstream)
+                        ), true);
             }
         });
     }
 
-    private static <T, A, R> Promise<A> parallelStep2(Iterator<? extends T> values, 
-                                                     int batchSize,
-                                                     Function<? super T, CompletionStage<? extends T>> spawner,                                                        
-                                                     Collector<T, A, R> downstream,
-                                                     Executor downstreamExecutor) {
+    private static <T, A, R> Promise<IndexedStep<A>> parallelStep2(
+        Iterator<? extends T> values, int batchSize,
+        Function<? super T, CompletionStage<? extends T>> spawner,                                                        
+        Collector<T, A, R> downstream,
+        Executor downstreamExecutor) {
 
-        int[] step = new int[1];
-        return loop(null, __ -> step[0] == 0 || values.hasNext(), current -> {
+        return loop(new IndexedStep<>(), step -> step.initial() || values.hasNext(), step -> {
             List<T> valuesBatch = drainBatch(values, batchSize);
-            boolean initial = step[0]++ == 0;
             if (valuesBatch.isEmpty()) {
                 // Over
-                return initial ?
-                    CompletableTask.supplyAsync(downstream.supplier(), downstreamExecutor)
+                return step.initial() ?
+                    CompletableTask.supplyAsync(() -> step.next(downstream.supplier().get()), downstreamExecutor)
                     :
-                    Promises.success(current);
+                    Promises.success(step);
             } else {
                 List<CompletionStage<? extends T>> promisesBatch = 
                         valuesBatch.stream()
@@ -396,11 +396,40 @@ private static <T, A, R> Promise<A> parallelStep2(Iterator<? extends T> values,
                 return 
                 Promises.all(promisesBatch)
                         .dependent()
-                        .thenApplyAsync(vals -> accumulate(vals, initial, current, downstream), downstreamExecutor, true);                
+                        .thenApplyAsync(vals -> step.next(
+                            accumulate(vals, step.initial(), step.payload(), downstream)
+                        ), downstreamExecutor, true);                
             }
         });
     }
 
+    
+    private static class IndexedStep<T> {
+        private final int idx;
+        private final T payload;
+        
+        IndexedStep() {
+            this(0, null);
+        }
+        
+        private IndexedStep(int idx, T payload) {
+            this.idx     = idx;
+            this.payload = payload;
+        }
+        
+        IndexedStep<T> next(T payload) {
+            return new IndexedStep<>(idx + 1, payload);
+        }
+        
+        boolean initial() {
+            return idx == 0;
+        }
+        
+        T payload() {
+            return payload;
+        }
+    }
+    
     private static <T> List<T> drainBatch(Iterator<? extends T> values, int batchSize) {
         List<T> valuesBatch = new ArrayList<>(batchSize);
         for (int count = 0; values.hasNext() && count < batchSize; count++) {
@@ -1060,13 +1089,13 @@ private static <T extends C, C> Promise<T> retryImpl(Function<? super RetryConte
 
         class State {
             private final boolean isDone;
-            private final Promise<?> prevAsync;
+            private final DependentPromise<?> prevAsync;
 
             final RetryContext<C> ctx;
             final RetryPolicy.Verdict verdict;
 
 
-            private State(RetryContext<C> ctx, Promise<?> prevAsync, boolean isDone) {
+            private State(RetryContext<C> ctx, DependentPromise<?> prevAsync, boolean isDone) {
                 this.ctx       = ctx;
                 this.prevAsync = prevAsync;
                 this.isDone    = isDone;
@@ -1079,11 +1108,11 @@ private State(RetryContext<C> ctx, Promise<?> prevAsync, boolean isDone) {
             }
 
             // Transition to intermediate
-            State next(Throwable error, Duration d, Promise<?> prevAsync) {
+            State next(Throwable error, Duration d, DependentPromise<?> prevAsync) {
                return new State(ctx.nextRetry(d, unwrapCompletionException(error)), useAsync(prevAsync), false); 
             }
 
-            State next(T result, Duration d, Promise<?> prevAsync) {
+            State next(T result, Duration d, DependentPromise<?> prevAsync) {
                 return new State(ctx.nextRetry(d, result), useAsync(prevAsync), false);
             }
 
@@ -1102,7 +1131,7 @@ Promise<T> toPromise() {
                 return  isDone ? success((T)ctx.getLastResult()) : failure(ctx.asFailure());
             }
 
-            Promise<?> useAsync(Promise<?> prevAsync) {
+            private DependentPromise<?> useAsync(DependentPromise<?> prevAsync) {
                 return prevAsync != null ? prevAsync : this.prevAsync;
             }
 
@@ -1111,7 +1140,6 @@ DependentPromise<?> makeDelay(Duration delay) {
                     Timeouts.delay(delay).dependent()
                     :
                     prevAsync.exceptionally(SharedFunctions.nullify()) // Don't propagate own error
-                             .dependent()
                              .delay(delay, true, true);
             }
         }
@@ -1123,14 +1151,14 @@ DependentPromise<?> makeDelay(Duration delay) {
             Supplier<Promise<State>> callSupplier = () -> {
                 long startTime = System.nanoTime();
 
-                Promise<T> target = futureFactory.apply(ctx);
+                DependentPromise<T> target = futureFactory.apply(ctx).dependent();
                 DependentPromise<T> withTimeout;
 
                 Duration timeout = verdict.timeout();
                 if (DelayPolicy.isValid(timeout)) {
-                    withTimeout = target.dependent().orTimeout(timeout, true, true);
+                    withTimeout = target.orTimeout(timeout, true, true);
                 } else {
-                    withTimeout = target.dependent();
+                    withTimeout = target;
                 }
 
                 return withTimeout.handle((value, ex) -> {

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

@@ -15,6 +15,7 @@
  */
 package net.tascalate.concurrent;
 
+import java.util.concurrent.CancellationException;
 import java.util.function.BiFunction;
 
 abstract class Try<R> {
@@ -57,7 +58,16 @@ static final class Failure<R> extends Try<R> {
 
         @Override
         R done() {
+            /*
             return sneakyThrow(error);
+            */
+            if (error instanceof Error) {
+                throw (Error)error;
+            } else if (error instanceof CancellationException) {
+                throw (CancellationException)error;
+            } else {
+                throw SharedFunctions.wrapCompletionException(error); 
+            }
         }
 
         @Override
@@ -88,10 +98,12 @@ static <R> Try<R> nothing() {
         return (Try<R>)NOTHING;
     }
 
+    /*
     @SuppressWarnings("unchecked")
     static <T, E extends Throwable> T sneakyThrow(Throwable e) throws E {
         throw (E) e;
     }
+    */
 
 
     private static final Try<Object> NOTHING = success(null); 

src/main/java/net/tascalate/concurrent/locks/AbstractAsyncLock.java

@@ -39,15 +39,15 @@ public interface AbstractAsyncLock<T extends AbstractAsyncLock.Token> {
      * lock after it has been acquired and the lock-protected action has
      * completed.
      *
-     * @return A {@link CompletionStage} which will complete with a
+     * @return A {@link Promise} which will complete with a
      *         {@link Token} when the lock has been exclusively acquired
      */
     Promise<T> acquire();
 
     /**
      * A lock token indicating that the associated lock has been exclusively
      * acquired. Once the protected action is completed, the lock may be
-     * released by calling {@link Token#releaseLock()}
+     * released by calling {@link Token#release()}
      */
     interface Token extends AutoCloseable {
         /** Releases this lock, allowing others to acquire it. */

src/main/java/net/tascalate/concurrent/locks/AsyncSemaphoreLock.java

@@ -42,7 +42,7 @@ default public Promise<Token> acquire() {
     /**
      * A semaphore token indicating that necessary permit has been exclusively acquired. Once the
      * protected action is completed, permits may be released by calling
-     * {@link Token#releaseLock()}
+     * {@link Token#release()}
      */
     interface Token extends AsyncLock.Token {
         long permits();