Default to Virtual Threads thread-per-task executor (#778)

pivovarit · Oct 3, 2023 · 5220c26 · 5220c26
1 parent 6d17bcd
commit 5220c26
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 25 deletions.
diff --git a/src/main/java/com/pivovarit/collectors/AsyncParallelCollector.java b/src/main/java/com/pivovarit/collectors/AsyncParallelCollector.java
@@ -17,7 +17,6 @@
 
 import static com.pivovarit.collectors.BatchingSpliterator.batching;
 import static com.pivovarit.collectors.BatchingSpliterator.partitioned;
-import static com.pivovarit.collectors.Dispatcher.getDefaultParallelism;
 import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.CompletableFuture.allOf;
 import static java.util.concurrent.CompletableFuture.supplyAsync;
@@ -94,10 +93,6 @@ private static <T> CompletableFuture<Stream<T>> combine(List<CompletableFuture<T
         return combined;
     }
 
-    static <T, R> Collector<T, ?, CompletableFuture<Stream<R>>> collectingToStream(Function<T, R> mapper, Executor executor) {
-        return collectingToStream(mapper, executor, getDefaultParallelism());
-    }
-
     static <T, R> Collector<T, ?, CompletableFuture<Stream<R>>> collectingToStream(Function<T, R> mapper, Executor executor, int parallelism) {
         requireNonNull(executor, "executor can't be null");
         requireNonNull(mapper, "mapper can't be null");
@@ -108,10 +103,6 @@ private static <T> CompletableFuture<Stream<T>> combine(List<CompletableFuture<T
           : new AsyncParallelCollector<>(mapper, Dispatcher.of(executor, parallelism), t -> t);
     }
 
-    static <T, R, RR> Collector<T, ?, CompletableFuture<RR>> collectingWithCollector(Collector<R, ?, RR> collector, Function<T, R> mapper, Executor executor) {
-        return collectingWithCollector(collector, mapper, executor, getDefaultParallelism());
-    }
-
     static <T, R, RR> Collector<T, ?, CompletableFuture<RR>> collectingWithCollector(Collector<R, ?, RR> collector, Function<T, R> mapper, Executor executor, int parallelism) {
         requireNonNull(collector, "collector can't be null");
         requireNonNull(executor, "executor can't be null");

diff --git a/src/main/java/com/pivovarit/collectors/Dispatcher.java b/src/main/java/com/pivovarit/collectors/Dispatcher.java
@@ -15,8 +15,6 @@
 import java.util.function.Function;
 import java.util.function.Supplier;
 
-import static java.lang.Runtime.getRuntime;
-
 /**
  * @author Grzegorz Piwowarek
  */
@@ -36,6 +34,11 @@ final class Dispatcher<T> {
 
     private volatile boolean shortCircuited = false;
 
+    private Dispatcher(int permits) {
+        this.executor = defaultExecutorService();
+        this.limiter = new Semaphore(permits);
+    }
+
     private Dispatcher(Executor executor, int permits) {
         this.executor = executor;
         this.limiter = new Semaphore(permits);
@@ -124,10 +127,6 @@ private static Runnable withFinally(Runnable task, Runnable finisher) {
         };
     }
 
-    static int getDefaultParallelism() {
-        return Math.max(getRuntime().availableProcessors() - 1, 4);
-    }
-
     private static ThreadPoolExecutor newLazySingleThreadExecutor() {
         return new ThreadPoolExecutor(0, 1,
           0L, TimeUnit.MILLISECONDS,
@@ -141,8 +140,8 @@ private static ThreadPoolExecutor newLazySingleThreadExecutor() {
     }
 
     static final class InterruptibleCompletableFuture<T> extends CompletableFuture<T> {
-        private volatile FutureTask<?> backingTask;
 
+        private volatile FutureTask<?> backingTask;
         private void completedBy(FutureTask<Void> task) {
             backingTask = task;
         }
@@ -154,5 +153,9 @@ 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
@@ -19,7 +19,6 @@
 import static com.pivovarit.collectors.BatchingSpliterator.partitioned;
 import static com.pivovarit.collectors.CompletionStrategy.ordered;
 import static com.pivovarit.collectors.CompletionStrategy.unordered;
-import static com.pivovarit.collectors.Dispatcher.getDefaultParallelism;
 import static java.util.Collections.emptySet;
 import static java.util.Objects.requireNonNull;
 import static java.util.stream.Collectors.collectingAndThen;
@@ -85,10 +84,6 @@ public Set<Characteristics> characteristics() {
         return characteristics;
     }
 
-    static <T, R> Collector<T, ?, Stream<R>> streaming(Function<T, R> mapper, Executor executor) {
-        return streaming(mapper, executor, getDefaultParallelism());
-    }
-
     static <T, R> Collector<T, ?, Stream<R>> streaming(Function<T, R> mapper, Executor executor, int parallelism) {
         requireNonNull(executor, "executor can't be null");
         requireNonNull(mapper, "mapper can't be null");
@@ -97,10 +92,6 @@ public Set<Characteristics> characteristics() {
         return new ParallelStreamCollector<>(mapper, unordered(), UNORDERED, Dispatcher.of(executor, parallelism));
     }
 
-    static <T, R> Collector<T, ?, Stream<R>> streamingOrdered(Function<T, R> mapper, Executor executor) {
-        return streamingOrdered(mapper, executor, getDefaultParallelism());
-    }
-
     static <T, R> Collector<T, ?, Stream<R>> streamingOrdered(Function<T, R> mapper, Executor executor,
         int parallelism) {
         requireNonNull(executor, "executor can't be null");