Fix cancellation in AsyncLoop

Author: vsilaev

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

@@ -181,7 +181,7 @@ void onComplete(int idx, T result, Throwable error) {
         }
     }
 
-    Promise<List<R>> postConstruct() {
+    Promise<List<R>> start() {
         int i = 0;
         for (CompletionStage<? extends T> promise : promises) {
             int idx = i++;

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

@@ -35,10 +35,7 @@ class AsyncLoop<T> extends CompletableFutureWrapper<T> {
     @Override
     public boolean cancel(boolean mayInterruptIfRunning) {
         if (super.cancel(mayInterruptIfRunning)) {
-            CompletionStage<?> dependent = currentStage;
-            if (null != dependent) {
-                SharedFunctions.cancelPromise(dependent, mayInterruptIfRunning);
-            }
+            cancelCurrentStage(mayInterruptIfRunning);
             return true;
         } else {
             return false;
@@ -49,13 +46,13 @@ void run(T initialValue) {
         run(initialValue, null, null);
     }
 
-    private void run(T resolvedValue, Thread initialThread, AsyncLoop.IterationState<T> initialState) {
+    private void run(T resolvedValue, Thread initialThread, IterationState<T> initialState) {
         Thread currentThread = Thread.currentThread();
         if (currentThread.equals(initialThread) && initialState.running) {
             initialState.put(resolvedValue);
             return;
         }
-        AsyncLoop.IterationState<T> currentState = new AsyncLoop.IterationState<>();
+        IterationState<T> currentState = new IterationState<>();
         T currentValue = resolvedValue;
         try {
             do {
@@ -70,6 +67,11 @@ private void run(T resolvedValue, Thread initialThread, AsyncLoop.IterationState
                                 run(next, currentThread, currentState);
                             }
                         });
+                        if (isDone()) {
+                            // If race between this.cancel and this.run
+                            // Double-cancel is not an issue
+                            cancelCurrentStage(true);
+                        }
                     } else {
                         success(currentValue);
                         break;
@@ -84,6 +86,13 @@ private void run(T resolvedValue, Thread initialThread, AsyncLoop.IterationState
         }
     }
 
+    private void cancelCurrentStage(boolean mayInterruptIfRunning) {
+        CompletionStage<?> dependent = currentStage;
+        if (null != dependent) {
+            SharedFunctions.cancelPromise(dependent, mayInterruptIfRunning);
+        }        
+    }
+    
     private static final class IterationState<T> {
         static final Object END = new Object();
 

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

@@ -990,7 +990,7 @@ private static <T, R> Promise<List<R>> atLeast(int minResultsCount, int maxError
             return transform(stage, singleResultMapper, Promises::wrapMultitargetException);
         } else {
             return ctr.create(minResultsCount, maxErrorsCount, cancelRemaining, promises)
-                      .postConstruct();
+                      .start();
         }
     }