Limit parallelism at thread-level instead of dispatcher-level (#784)

pivovarit · Oct 9, 2023 · b6a029a · b6a029a
1 parent 04ce124
commit b6a029a
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 24 deletions.
diff --git a/src/main/java/com/pivovarit/collectors/AsyncParallelCollector.java b/src/main/java/com/pivovarit/collectors/AsyncParallelCollector.java
@@ -99,7 +99,7 @@ private static <T> CompletableFuture<Stream<T>> combine(List<CompletableFuture<T
 
         return parallelism == 1
           ? asyncCollector(mapper, executor, i -> i)
-          : new AsyncParallelCollector<>(mapper, Dispatcher.of(executor, parallelism), t -> t);
+          : new AsyncParallelCollector<>(mapper, Dispatcher.from(executor, parallelism), Function.identity());
     }
 
     static <T, R, RR> Collector<T, ?, CompletableFuture<RR>> collectingWithCollector(Collector<R, ?, RR> collector, Function<T, R> mapper, Executor executor, int parallelism) {
@@ -110,7 +110,7 @@ private static <T> CompletableFuture<Stream<T>> combine(List<CompletableFuture<T
 
         return parallelism == 1
           ? asyncCollector(mapper, executor, s -> s.collect(collector))
-          : new AsyncParallelCollector<>(mapper, Dispatcher.of(executor, parallelism), s -> s.collect(collector));
+          : new AsyncParallelCollector<>(mapper, Dispatcher.from(executor, parallelism), s -> s.collect(collector));
     }
 
     static void requireValidParallelism(int parallelism) {
@@ -166,13 +166,13 @@ private BatchingCollectors() {
                       return list.stream()
                         .collect(new AsyncParallelCollector<>(
                           mapper,
-                          Dispatcher.of(executor, parallelism),
+                          Dispatcher.from(executor, parallelism),
                           finisher));
                   } else {
                       return partitioned(list, parallelism)
                         .collect(new AsyncParallelCollector<>(
                           batching(mapper),
-                          Dispatcher.of(executor, parallelism),
+                          Dispatcher.from(executor, parallelism),
                           listStream -> finisher.apply(listStream.flatMap(Collection::stream))));
                   }
               });

diff --git a/src/main/java/com/pivovarit/collectors/BatchingSpliterator.java b/src/main/java/com/pivovarit/collectors/BatchingSpliterator.java
@@ -1,7 +1,6 @@
 package com.pivovarit.collectors;
 
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Spliterator;
 import java.util.function.Consumer;

diff --git a/src/main/java/com/pivovarit/collectors/Dispatcher.java b/src/main/java/com/pivovarit/collectors/Dispatcher.java
@@ -44,19 +44,32 @@ private Dispatcher(Executor executor, int permits) {
         this.limiter = new Semaphore(permits);
     }
 
-    static <T> Dispatcher<T> of(Executor executor, int permits) {
+    static <T> Dispatcher<T> from(Executor executor, int permits) {
         return new Dispatcher<>(executor, permits);
     }
 
+    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) {
-                            limiter.acquire();
-                            executor.execute(withFinally(task, limiter::release));
+                            executor.execute(() -> {
+                                try {
+                                    limiter.acquire();
+                                    task.run();
+                                } catch (InterruptedException e) {
+                                    Thread.currentThread().interrupt();
+                                    throw new RuntimeException(e);
+                                } finally {
+                                    limiter.release();
+                                }
+                            });
                         } else {
                             break;
                         }
@@ -117,20 +130,10 @@ private static Function<Throwable, Void> shortcircuit(InterruptibleCompletableFu
         };
     }
 
-    private static Runnable withFinally(Runnable task, Runnable finisher) {
-        return () -> {
-            try {
-                task.run();
-            } finally {
-                finisher.run();
-            }
-        };
-    }
-
     private static ThreadPoolExecutor newLazySingleThreadExecutor() {
-        return new ThreadPoolExecutor(0, 1,
+        return new ThreadPoolExecutor(1, 1,
           0L, TimeUnit.MILLISECONDS,
-          new LinkedBlockingQueue<>(),
+          new SynchronousQueue<>(), // dispatcher always executes a single task
           Thread.ofPlatform()
             .name("parallel-collectors-dispatcher-", 0)
             .daemon(false)

diff --git a/src/main/java/com/pivovarit/collectors/ParallelStreamCollector.java b/src/main/java/com/pivovarit/collectors/ParallelStreamCollector.java
@@ -89,7 +89,7 @@ public Set<Characteristics> characteristics() {
         requireNonNull(mapper, "mapper can't be null");
         requireValidParallelism(parallelism);
 
-        return new ParallelStreamCollector<>(mapper, unordered(), UNORDERED, Dispatcher.of(executor, parallelism));
+        return new ParallelStreamCollector<>(mapper, unordered(), UNORDERED, Dispatcher.from(executor, parallelism));
     }
 
     static <T, R> Collector<T, ?, Stream<R>> streamingOrdered(Function<T, R> mapper, Executor executor,
@@ -98,7 +98,7 @@ public Set<Characteristics> characteristics() {
         requireNonNull(mapper, "mapper can't be null");
         requireValidParallelism(parallelism);
 
-        return new ParallelStreamCollector<>(mapper, ordered(), emptySet(), Dispatcher.of(executor, parallelism));
+        return new ParallelStreamCollector<>(mapper, ordered(), emptySet(), Dispatcher.from(executor, parallelism));
     }
 
     static final class BatchingCollectors {
@@ -140,15 +140,15 @@ private BatchingCollectors() {
                                 mapper,
                                 ordered(),
                                 emptySet(),
-                                Dispatcher.of(executor, parallelism)));
+                                Dispatcher.from(executor, parallelism)));
                     }
                     else {
                         return partitioned(list, parallelism)
                             .collect(collectingAndThen(new ParallelStreamCollector<>(
                                     batching(mapper),
                                     ordered(),
                                     emptySet(),
-                                    Dispatcher.of(executor, parallelism)),
+                                    Dispatcher.from(executor, parallelism)),
                                 s -> s.flatMap(Collection::stream)));
                     }
                 });