Dispatcher to use caller thread instead of dedicated scheduler thread

pivovarit · Oct 10, 2023 · e9ef721 · e9ef721
1 parent 15126c4
commit e9ef721
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 75 deletions.
diff --git a/src/main/java/com/pivovarit/collectors/AsyncParallelCollector.java b/src/main/java/com/pivovarit/collectors/AsyncParallelCollector.java
@@ -55,18 +55,13 @@ public BinaryOperator<List<CompletableFuture<R>>> combiner() {
 
     @Override
     public BiConsumer<List<CompletableFuture<R>>, T> accumulator() {
-        return (acc, e) -> {
-            if (!dispatcher.isRunning()) {
-                dispatcher.start();
-            }
-            acc.add(dispatcher.enqueue(() -> mapper.apply(e)));
-        };
+        return (acc, e) -> acc.add(dispatcher.enqueue(() -> mapper.apply(e)));
     }
 
     @Override
     public Function<List<CompletableFuture<R>>, CompletableFuture<C>> finisher() {
         return futures -> {
-            dispatcher.stop();
+            dispatcher.complete();
 
             return combine(futures).thenApply(processor);
         };

diff --git a/src/main/java/com/pivovarit/collectors/Dispatcher.java b/src/main/java/com/pivovarit/collectors/Dispatcher.java
@@ -1,17 +1,11 @@
 package com.pivovarit.collectors;
 
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.FutureTask;
-import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.Semaphore;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -20,17 +14,11 @@
  */
 final class Dispatcher<T> {
 
-    private static final Runnable POISON_PILL = () -> System.out.println("Why so serious?");
-
     private final CompletableFuture<Void> completionSignaller = new CompletableFuture<>();
-
-    private final BlockingQueue<Runnable> workingQueue = new LinkedBlockingQueue<>();
-
-    private final ExecutorService dispatcher = newLazySingleThreadExecutor();
     private final Executor executor;
     private final Semaphore limiter;
 
-    private final AtomicBoolean started = new AtomicBoolean(false);
+    private volatile boolean completed = false;
 
     private volatile boolean shortCircuited = false;
 
@@ -52,51 +40,13 @@ static <T> Dispatcher<T> virtual(int permits) {
         return new Dispatcher<>(permits);
     }
 
-    void start() {
-        if (!started.getAndSet(true)) {
-            dispatcher.execute(() -> {
-                try {
-                    while (true) {
-                        Runnable task;
-                        if ((task = workingQueue.take()) != POISON_PILL) {
-                            executor.execute(() -> {
-                                try {
-                                    limiter.acquire();
-                                    task.run();
-                                } catch (InterruptedException e) {
-                                    handle(e);
-                                } finally {
-                                    limiter.release();
-                                }
-                            });
-                        } else {
-                            break;
-                        }
-                    }
-                } catch (Throwable e) {
-                    handle(e);
-                }
-            });
-        }
-    }
-
-    void stop() {
-        try {
-            workingQueue.put(POISON_PILL);
-        } catch (InterruptedException e) {
-            completionSignaller.completeExceptionally(e);
-        } finally {
-            dispatcher.shutdown();
-        }
-    }
-
-    boolean isRunning() {
-        return started.get();
+    void complete() {
+        completed = true;
     }
 
     CompletableFuture<T> enqueue(Supplier<T> supplier) {
         InterruptibleCompletableFuture<T> future = new InterruptibleCompletableFuture<>();
-        workingQueue.add(completionTask(supplier, future));
+        executor.execute(completionTask(supplier, future));
         completionSignaller.exceptionally(shortcircuit(future));
         return future;
     }
@@ -105,7 +55,12 @@ private FutureTask<Void> completionTask(Supplier<T> supplier, InterruptibleCompl
         FutureTask<Void> task = new FutureTask<>(() -> {
             try {
                 if (!shortCircuited) {
-                    future.complete(supplier.get());
+                    try {
+                        limiter.acquire();
+                        future.complete(supplier.get());
+                    } finally {
+                        limiter.release();
+                    }
                 }
             } catch (Throwable e) {
                 handle(e);
@@ -118,7 +73,6 @@ private FutureTask<Void> completionTask(Supplier<T> supplier, InterruptibleCompl
     private void handle(Throwable e) {
         shortCircuited = true;
         completionSignaller.completeExceptionally(e);
-        dispatcher.shutdownNow();
     }
 
     private static Function<Throwable, Void> shortcircuit(InterruptibleCompletableFuture<?> future) {
@@ -129,19 +83,10 @@ private static Function<Throwable, Void> shortcircuit(InterruptibleCompletableFu
         };
     }
 
-    private static ThreadPoolExecutor newLazySingleThreadExecutor() {
-        return new ThreadPoolExecutor(1, 1,
-          0L, TimeUnit.MILLISECONDS,
-          new SynchronousQueue<>(), // dispatcher always executes a single task
-          Thread.ofPlatform()
-            .name("parallel-collectors-dispatcher-", 0)
-            .daemon(false)
-            .factory());
-    }
-
     static final class InterruptibleCompletableFuture<T> extends CompletableFuture<T> {
 
         private volatile FutureTask<?> backingTask;
+
         private void completedBy(FutureTask<Void> task) {
             backingTask = task;
         }
@@ -153,8 +98,8 @@ public boolean cancel(boolean mayInterruptIfRunning) {
             }
             return super.cancel(mayInterruptIfRunning);
         }
-
     }
+
     private static ExecutorService defaultExecutorService() {
         return Executors.newVirtualThreadPerTaskExecutor();
     }

diff --git a/src/main/java/com/pivovarit/collectors/ParallelStreamCollector.java b/src/main/java/com/pivovarit/collectors/ParallelStreamCollector.java
@@ -74,7 +74,7 @@ public BinaryOperator<List<CompletableFuture<R>>> combiner() {
     @Override
     public Function<List<CompletableFuture<R>>, Stream<R>> finisher() {
         return acc -> {
-            dispatcher.stop();
+            dispatcher.complete();
             return completionStrategy.apply(acc);
         };
     }