second attempt

Author: stIncMale

driver-core/src/main/com/mongodb/internal/async/function/AsyncCallbackLoop.java

@@ -61,7 +61,6 @@ public AsyncCallbackLoop(final boolean optimized, final LoopState state, final A
     @Override
     public void run(final SingleResultCallback<Void> callback) {
         body.run(new LoopingCallback(callback));
-        CallbackChain.run(chain);
     }
 
     /**
@@ -88,7 +87,7 @@ public void onResult(@Nullable final Void result, @Nullable final Throwable t) {
                     return;
                 }
                 if (continueLooping) {
-                    CallbackChain.addOrRun(chain, () -> body.run(this));
+                    CallbackChain.execute(chain, () -> body.run(this));
                 } else {
                     wrapped.onResult(result, null);
                 }

driver-core/src/main/com/mongodb/internal/async/function/CallbackChain.java

@@ -17,40 +17,50 @@
 
 import com.mongodb.lang.Nullable;
 
+import java.util.concurrent.atomic.AtomicReference;
+
 import static com.mongodb.assertions.Assertions.assertNotNull;
 import static com.mongodb.assertions.Assertions.assertNull;
 
-public final class CallbackChain {
+final class CallbackChain {
     @Nullable
     private Runnable next;
+    private int runEnteredCounter;
+    private final AtomicReference<Thread> thread;
 
-    public CallbackChain() {
+    CallbackChain() {
+        runEnteredCounter = 0;
+        thread = new AtomicReference<>();
     }
 
-    public static void addOrRun(@Nullable final CallbackChain chain, final Runnable next) {
+    static void execute(@Nullable final CallbackChain chain, final Runnable next) {
         if (chain != null) {
-            chain.add(next);
+            chain.execute(next);
         } else {
             next.run();
         }
     }
 
-    public static void run(final @Nullable CallbackChain chain) {
-        if (chain != null) {
-            chain.run();
-        }
-    }
-
-    private void add(final Runnable next) {
+    // VAKOTODO figure out thread safety
+    private void execute(final Runnable next) {
         assertNotNull(next);
         assertNull(this.next);
         this.next = next;
-    }
 
-    private void run() {
-        for (Runnable localNext = next; localNext != null; localNext = next) {
-            next = null;
-            localNext.run();
+//        if (!thread.compareAndSet(null, Thread.currentThread())) {
+//            assertTrue(Thread.currentThread() == thread.get());
+//        }
+        boolean recursive = ++runEnteredCounter > 1;
+        try {
+            if (recursive) {
+                return;
+            }
+            for (Runnable localNext = next; localNext != null; localNext = this.next) {
+                this.next = null;
+                localNext.run();
+            }
+        } finally {
+            runEnteredCounter--;
         }
     }
 }

driver-core/src/test/unit/com/mongodb/internal/async/AsyncFunctionsAbstractTest.java

@@ -990,4 +990,23 @@ void testDerivation() {
                     }).finish(callback);
                 });
     }
+
+    @Test
+    void testThenRunDoWhileLoop() {
+        assertBehavesSameVariations(8,
+                () -> {
+                    int i = 0;
+                    do {
+                        i++;
+                        sync(i);
+                    } while (i < 3 && plainTest(i));
+                },
+                (callback) -> {
+                    final int[] i = new int[1];
+                    beginAsync().thenRunDoWhileLoop((c) -> {
+                        i[0]++;
+                        async(i[0], c);
+                    }, () -> i[0] < 3 && plainTest(i[0])).finish(callback);
+                });
+    }
 }

driver-core/src/test/unit/com/mongodb/internal/async/VakoTest.java

@@ -17,24 +17,27 @@
 
 import com.mongodb.internal.async.function.AsyncCallbackLoop;
 import com.mongodb.internal.async.function.LoopState;
+import com.mongodb.internal.time.StartTime;
 import com.mongodb.lang.Nullable;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
-import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.TimeUnit;
 
 import static com.mongodb.internal.async.AsyncRunnable.beginAsync;
 
 class VakoTest {
     @ParameterizedTest
-    @CsvSource({"false, 20", "true, 20"})
-    void asyncCallbackLoop(final boolean optimized, final int iterations) {
+    @CsvSource({"false, 30", "true, 30"})
+    void asyncCallbackLoop(final boolean optimized, final int iterations) throws Exception {
         System.err.printf("baselineStackDepth=%d%n", Thread.currentThread().getStackTrace().length);
+        CompletableFuture<Void> join = new CompletableFuture<>();
         LoopState loopState = new LoopState();
         new AsyncCallbackLoop(optimized, loopState, c -> {
             int iteration = loopState.iteration();
@@ -45,127 +48,83 @@ void asyncCallbackLoop(final boolean optimized, final int iterations) {
         }).run((r, t) -> {
             System.err.printf("test callback completed callStackDepth=%d, r=%s, t=%s%n",
                     Thread.currentThread().getStackTrace().length, r, exceptionToString(t));
+            if (t != null) {
+                join.completeExceptionally(t);
+            } else {
+                join.complete(r);
+            }
         });
+        join.get();
     }
 
     @ParameterizedTest
-    @CsvSource({"false, 20", "true, 20"})
-    void testA(final boolean optimized, final int counterValue) {
+    @CsvSource({
+            "false, false, 30", "false, true, 30",
+            "true, false, 30", "true, true, 30"})
+    void testThenRunDoWhileLoop(final boolean optimized, final boolean separateThread, final int counterInitialValue) throws Exception {
+        StartTime start = StartTime.now();
         System.err.printf("baselineStackDepth=%d%n", Thread.currentThread().getStackTrace().length);
-        asyncMethod1A(optimized, new Counter(counterValue), (r, t) -> {
+        CompletableFuture<Void> join = new CompletableFuture<>();
+        asyncMethod1(optimized, separateThread, new Counter(counterInitialValue), (r, t) -> {
             System.err.printf("test callback completed callStackDepth=%s, r=%s, t=%s%n",
                     Thread.currentThread().getStackTrace().length, r, exceptionToString(t));
+            if (t != null) {
+                join.completeExceptionally(t);
+            } else {
+                join.complete(r);
+            }
         });
+        System.err.printf("asyncMethod1 executed in %s%n", start.elapsed());
+        join.get();
     }
 
-    private static void asyncMethod1A(final boolean optimized, final Counter counter, final SingleResultCallback<Void> callback) {
+    private static void asyncMethod1(final boolean optimized, final boolean separateThread,
+            final Counter counter, final SingleResultCallback<Void> callback) {
         beginAsync().thenRunDoWhileLoop(optimized, c -> {
-            asyncMethod2A(counter, c);
+            asyncMethod2(separateThread, counter, c);
         }, () -> !counter.done()).finish(callback);
     }
 
-    private static void asyncMethod2A(final Counter counter, final SingleResultCallback<Void> callback) {
-        counter.countDown();
-        callback.complete(callback);
-    }
-
-    @ParameterizedTest
-    @ValueSource(ints = {10})
-    void testB(final int counterValue) {
-        AtomicInteger stackDepthUnoptimized = new AtomicInteger();
-        AtomicInteger stackDepthOptimized = new AtomicInteger();
-        asyncMethod1B(false, new Counter(counterValue), (r, t) -> {
-            stackDepthUnoptimized.set(Thread.currentThread().getStackTrace().length);
-            System.err.printf("test callback completed callStackDepth=%s, r=%s, t=%s%n",
-                    stackDepthUnoptimized, r, exceptionToString(t));
-        });
-        asyncMethod1B(true, new Counter(counterValue), (r, t) -> {
-            stackDepthOptimized.set(Thread.currentThread().getStackTrace().length);
-            System.err.printf("test callback completed callStackDepth=%s, r=%s, t=%s%n",
-                    stackDepthOptimized, r, exceptionToString(t));
-        });
-        System.err.printf("test completed baselineStackDepth=%d, stackDepthUnoptimized=%s, stackDepthOptimized=%s%n",
-                Thread.currentThread().getStackTrace().length, stackDepthOptimized, stackDepthUnoptimized);
-    }
-
-    private static void asyncMethod1B(final boolean optimized, final Counter counter, final SingleResultCallback<Void> callback) {
-        asyncMethod2B(counter, new Callback(optimized, counter, callback));
-    }
-
-    private static void asyncMethod2B(final Counter counter, final SingleResultCallback<Void> callback) {
-        counter.countDown();
-        callback.complete(callback);
-    }
-
-    private static final class Callback implements SingleResultCallback<Void> {
-        private final boolean optimized;
-        private Counter counter;
-        private SingleResultCallback<Void> callback;
-
-        Callback(final boolean optimized, final Counter counter, final SingleResultCallback<Void> callback) {
-            this.optimized = optimized;
-            this.counter = counter;
-            this.callback = callback;
-        }
-
-        Counter takeCounter() {
-            Counter localCounter = com.mongodb.assertions.Assertions.assertNotNull(counter);
-            counter = null;
-            return localCounter;
-        }
-
-        void setCounter(final Counter counter) {
-            com.mongodb.assertions.Assertions.assertNull(this.counter);
-            this.counter = counter;
-        }
-
-        SingleResultCallback<Void> takeCallback() {
-            SingleResultCallback<Void> localCallback = com.mongodb.assertions.Assertions.assertNotNull(callback);
-            callback = null;
-            return localCallback;
-        }
-
-        void setCallback(final SingleResultCallback<Void> callback) {
-            com.mongodb.assertions.Assertions.assertNull(this.callback);
-            this.callback = callback;
-        }
-
-        @Override
-        public void onResult(final Void result, final Throwable t) {
-            SingleResultCallback<Void> localCallback = takeCallback();
-            beginAsync().thenRun((c) -> {
-                System.err.printf("thenRun%n");
-                Counter localCounter = takeCounter();
-                if (t != null) {
-                    System.err.printf("exception t=%s%n", exceptionToString(t));
-                    c.completeExceptionally(t);
-                } else if (localCounter.done()) {
-                    c.complete(c);
-                } else {
-                    asyncMethod2B(localCounter, new Callback(optimized, localCounter, localCallback));
-                }
-            }).finish((r, t2) -> {
-                System.err.printf("finish r=%s, t=%s%n", r, exceptionToString(t2));
-                localCallback.onResult(r, t);
-            });
+    private static void asyncMethod2(final boolean separateThread, final Counter counter, final SingleResultCallback<Void> callback) {
+        Runnable action = () -> {
+            try {
+                Thread.sleep(TimeUnit.SECONDS.toMillis(1) / counter.initial());
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new RuntimeException(e);
+            }
+            counter.countDown();
+            callback.complete(callback);
+        };
+        if (separateThread) {
+            ForkJoinPool.commonPool().execute(action);
+        } else {
+            action.run();
         }
     }
 
     private static final class Counter {
-        private int v;
+        private final int initial;
+        private int current;
+
+        Counter(final int initial) {
+            this.initial = initial;
+            this.current = initial;
+        }
 
-        Counter(final int v) {
-            this.v = v;
+        int initial() {
+            return initial;
         }
 
         void countDown() {
-            com.mongodb.assertions.Assertions.assertTrue(v > 0);
-            v--;
-            System.err.printf("counted %d->%d callStackDepth=%d %n", v + 1, v, Thread.currentThread().getStackTrace().length);
+            com.mongodb.assertions.Assertions.assertTrue(current > 0);
+            int previous = current;
+            int decremented = --current;
+            System.err.printf("counted %d->%d callStackDepth=%d %n", previous, decremented, Thread.currentThread().getStackTrace().length);
         }
 
         boolean done() {
-            if (v == 0) {
+            if (current == 0) {
                 System.err.printf("counting done callStackDepth=%d %n", Thread.currentThread().getStackTrace().length);
                 return true;
             }
@@ -179,36 +138,12 @@ private static String exceptionToString(@Nullable final Throwable t) {
         }
         try (StringWriter sw = new StringWriter();
              PrintWriter pw = new PrintWriter(sw)) {
-            t.printStackTrace(pw);
+//            t.printStackTrace(pw);
+            pw.println(t);
             pw.flush();
             return sw.toString();
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
     }
 }
-
-/*
-
-        c3.complete{
-            //c3...
-            c2.complete{
-                //c2...
-                c1.complete{
-                    //c1...
-                }
-        }
-
-        c3.complete{
-            //c3...
-            chain.add(c2)
-        }
-        chain.run() -> c2.complete{
-            //c2...
-            chain.add(c1)
-        }
-        chain.run() -> c1.complete{
-            //c1...
-        }
-
-         */