Add ExecutorPollutionTest (#874)

Author: pivovarit

src/test/java/com/pivovarit/collectors/functional/ExecutorPollutionTest.java

@@ -0,0 +1,62 @@
+package com.pivovarit.collectors.functional;
+
+import com.pivovarit.collectors.ParallelCollectors;
+import org.junit.jupiter.api.DynamicTest;
+import org.junit.jupiter.api.TestFactory;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collector;
+import java.util.stream.Stream;
+
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Stream.of;
+
+class ExecutorPollutionTest {
+
+    @TestFactory
+    Stream<DynamicTest> shouldStartProcessingElementsTests() {
+        return of(
+          shouldNotSubmitMoreTasksThanParallelism(ParallelCollectors::parallel, "parallel#1"),
+          shouldNotSubmitMoreTasksThanParallelism((f, e, p) -> ParallelCollectors.parallel(f, toList(), e, p), "parallel#2"),
+          shouldNotSubmitMoreTasksThanParallelism(ParallelCollectors::parallelToStream, "parallelToStream"),
+          shouldNotSubmitMoreTasksThanParallelism(ParallelCollectors::parallelToOrderedStream, "parallelToOrderedStream"),
+          shouldNotSubmitMoreTasksThanParallelism(ParallelCollectors.Batching::parallel, "parallel#1 (batching)"),
+          shouldNotSubmitMoreTasksThanParallelism((f, e, p) -> ParallelCollectors.Batching.parallel(f, toList(), e, p), "parallel#2 (batching)"),
+          shouldNotSubmitMoreTasksThanParallelism(ParallelCollectors.Batching::parallelToStream, "parallelToStream (batching)"),
+          shouldNotSubmitMoreTasksThanParallelism(ParallelCollectors.Batching::parallelToOrderedStream, "parallelToOrderedStream (batching)")
+        );
+    }
+
+    private static DynamicTest shouldNotSubmitMoreTasksThanParallelism(CollectorFactory<Integer> collector, String name) {
+        return DynamicTest.dynamicTest(name, () -> {
+            try (var e = warmedUp(new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(2)))) {
+
+                var result = Stream.generate(() -> 42)
+                  .limit(1000)
+                  .collect(collector.apply(i -> i, e, 2));
+
+                switch (result) {
+                    case CompletableFuture<?> cf -> cf.join();
+                    case Stream<?> s -> s.forEach((__) -> {});
+                    default -> throw new IllegalStateException("can't happen");
+                }
+            }
+        });
+    }
+
+    interface CollectorFactory<T> {
+        Collector<T, ?, ?> apply(Function<T, ?> function, Executor executorService, int parallelism);
+    }
+
+    private static ThreadPoolExecutor warmedUp(ThreadPoolExecutor e) {
+        for (int i = 0; i < e.getCorePoolSize(); i++) {
+            e.submit(() -> {});
+        }
+        return e;
+    }
+}

src/test/java/com/pivovarit/collectors/ExecutorValidationTest.java renamed to src/test/java/com/pivovarit/collectors/functional/ExecutorValidationTest.java

@@ -1,5 +1,6 @@
-package com.pivovarit.collectors;
+package com.pivovarit.collectors.functional;
 
+import com.pivovarit.collectors.ParallelCollectors;
 import org.junit.jupiter.api.DynamicTest;
 import org.junit.jupiter.api.TestFactory;
 

src/test/java/com/pivovarit/collectors/ToStreamTest.java renamed to src/test/java/com/pivovarit/collectors/functional/ImmediateStreamProcessingTest.java

@@ -1,5 +1,6 @@
-package com.pivovarit.collectors;
+package com.pivovarit.collectors.functional;
 
+import com.pivovarit.collectors.ParallelCollectors;
 import org.junit.jupiter.api.DynamicTest;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestFactory;
@@ -15,7 +16,7 @@
 import static java.util.stream.Stream.of;
 import static org.awaitility.Awaitility.await;
 
-class ToStreamTest {
+class ImmediateStreamProcessingTest {
 
     @TestFactory
     Stream<DynamicTest> shouldStartProcessingElementsTests() {
@@ -24,25 +25,24 @@ Stream<DynamicTest> shouldStartProcessingElementsTests() {
           shouldStartProcessingElements(f -> ParallelCollectors.parallelToOrderedStream(f, Executors.newVirtualThreadPerTaskExecutor(), 2), "parallelToStream, parallelism: 2, vthreads"),
           shouldStartProcessingElements(f -> ParallelCollectors.parallelToOrderedStream(f, Executors.newCachedThreadPool(), 2), "parallelToOrderedStream, parallelism: 2, os threads"),
           shouldStartProcessingElements(f -> ParallelCollectors.parallelToOrderedStream(f, Executors.newVirtualThreadPerTaskExecutor(), 2), "parallelToOrderedStream, parallelism: 2, vthreads")
-          );
+        );
     }
 
     private static DynamicTest shouldStartProcessingElements(Function<Function<Integer, Integer>, Collector<Integer, ?, Stream<Integer>>> collector, String name) {
         return DynamicTest.dynamicTest(name, () -> {
             var counter = new AtomicInteger();
-            Thread.ofPlatform().start(() -> {
-                Stream.iterate(0, i -> i + 1)
-                  .limit(100)
-                  .collect(collector.apply(i -> {
-                      try {
-                          Thread.sleep(100);
-                      } catch (InterruptedException ex) {
-                          Thread.currentThread().interrupt();
-                      }
-                      return i;
-                  }))
-                  .forEach(c -> counter.incrementAndGet());
-            });
+            Thread.ofPlatform()
+              .start(() -> Stream.iterate(0, i -> i + 1)
+                .limit(100)
+                .collect(collector.apply(i -> {
+                    try {
+                        Thread.sleep(100);
+                    } catch (InterruptedException ex) {
+                        Thread.currentThread().interrupt();
+                    }
+                    return i;
+                }))
+                .forEach(c -> counter.incrementAndGet()));
             await()
               .atMost(ofSeconds(1))
               .until(() -> counter.get() > 0);