package rsc.publisher;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
import org.openjdk.jmh.annotations.Measurement;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.OutputTimeUnit;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.Warmup;
import org.openjdk.jmh.infra.Blackhole;

import rsc.scheduler.ParallelScheduler;
import rsc.scheduler.Scheduler;
import rsc.util.PerfAsyncSubscriber;

/**
 * Benchmark parallelization approaches.
 * <p>
 * gradle jmh -Pjmh='ParallelApproachesPerf'
 */
@BenchmarkMode(Mode.Throughput)
@Warmup(iterations = 5)
@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
@Fork(value = 1)
@OutputTimeUnit(TimeUnit.SECONDS)
@State(Scope.Thread)
public class ParallelApproachesPerf {

    @Param({"10000"})
    public int count;
    
    @Param({"1", "10", "100", "1000", "10000"})
    public int compute;
    
    @Param({"1", "2", "3", "4"})
    public int parallelism;
    
    Px<Integer> groupBy;
    
    Px<Integer> window;
    
    Px<Integer> flatMap;

    Px<Integer> flatMapLimit;

    Scheduler scheduler;
    
    @Setup
    public void setup(Blackhole bh) {
        
        scheduler = new ParallelScheduler(parallelism);
        
        Integer[] values = new Integer[count];
        Arrays.fill(values, 0);
        
        Px<Integer> source = Px.fromArray(values);
        
        int[] index = { 0 };
        Function<Integer, Integer> gf = v -> {
            int g = index[0]++;
            if (g + 1 == parallelism) {
                index[0] = 0;
            }
            return g;
        };
        
        Function<Integer, Integer> work = v -> { Blackhole.consumeCPU(compute); return v; };
        
        groupBy = source.groupBy(gf).flatMap(g -> g.observeOn(scheduler).map(work), false, Px.BUFFER_SIZE);
        
        window = source.window(Px.BUFFER_SIZE).flatMap(w -> w.subscribeOn(scheduler).map(work), false, Px.BUFFER_SIZE);
        
        flatMap = source.flatMap(v -> Px.just(v).observeOn(scheduler).map(work), false, Px.BUFFER_SIZE);

        flatMapLimit = source.flatMap(v -> Px.just(v).observeOn(scheduler).map(work), false, parallelism);
    }

    @Benchmark
    public void groupBy(Blackhole bh) {
        PerfAsyncSubscriber s = new PerfAsyncSubscriber(bh);
        groupBy.subscribe(s);
        s.await(count);
    }

    @Benchmark
    public void window(Blackhole bh) {
        PerfAsyncSubscriber s = new PerfAsyncSubscriber(bh);
        window.subscribe(s);
        s.await(count);
    }
    
    @Benchmark
    public void flatMap(Blackhole bh) {
        PerfAsyncSubscriber s = new PerfAsyncSubscriber(bh);
        flatMap.subscribe(s);
        s.await(count);
    }

    @Benchmark
    public void flatMapLimit(Blackhole bh) {
        PerfAsyncSubscriber s = new PerfAsyncSubscriber(bh);
        flatMapLimit.subscribe(s);
        s.await(count);
    }


}
@@ -63,63 +63,196 @@
 til.
-concurrent.TimeUnit;%0Aimport java.util.function.Function
+List;%0Aimport java.util.concurrent.ForkJoinPool;%0Aimport java.util.concurrent.TimeUnit;%0Aimport java.util.function.Function;%0Aimport java.util.function.Supplier;%0Aimport java.util.stream.Stream
 ;%0A%0Ai
@@ -1559,24 +1559,93 @@
 duler;%0A    %0A
+    Supplier%3CStream%3CInteger%3E%3E stream;%0A    %0A    ForkJoinPool fj;%0A    %0A
     @Setup%0A 
@@ -1681,25 +1681,16 @@
 e bh) %7B%0A
-        %0A
         
@@ -2685,24 +2685,591 @@
 rallelism);%0A
+        %0A        List%3CInteger%3E list = Arrays.asList(values);%0A        %0A        fj = new ForkJoinPool(parallelism);%0A        %0A        stream = () -%3E %7B%0A            return list.parallelStream().unordered().map(work);%0A        %7D;%0A    %7D%0A%0A    @Benchmark%0A    public void stream(Blackhole bh) %7B%0A        stream.get().iterator().forEachRemaining(bh::consume);%0A    %7D%0A%0A    @Benchmark%0A    public void streamCustom(Blackhole bh) throws Exception %7B%0A        fj.submit(() -%3E %7B%0A            stream.get().iterator().forEachRemaining(bh::consume);%0A            return null;%0A        %7D).get();%0A
     %7D%0A%0A    @

Add parallelStream benchmark