/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/publicdomain/zero/1.0/
 */

import java.util.*;
import java.util.concurrent.*;

/**
 * Creates PRODUCERS publishers each with 1 subscriber,
 * sent ITEMS items; repeats REPS times
 */
public class SubmissionPublisherLoops5 {
    static final int ITEMS      = 1 << 25;
    static final int PRODUCERS  = 32;
    static final int REPS = 9;

    static final Phaser phaser = new Phaser(PRODUCERS + 1);

    public static void main(String[] args) throws Exception {
        int reps = REPS;
        if (args.length > 0)
            reps = Integer.parseInt(args[0]);

        System.out.println("ITEMS: " + ITEMS +
                           " PRODUCERS: " + PRODUCERS +
                           " ITEMS: " + ITEMS);
        for (int rep = 0; rep < reps; ++rep) {
            oneRun();
            Thread.sleep(1000);
        }
    }

    static void oneRun() throws Exception {
        long nitems = (long)ITEMS * PRODUCERS;
        long startTime = System.nanoTime();
        for (int i = 0; i < PRODUCERS; ++i)
            new Pub().fork();
        phaser.arriveAndAwaitAdvance();
        long elapsed = System.nanoTime() - startTime;
        double secs = ((double)elapsed) / (1000L * 1000 * 1000);
        double ips = nitems / secs;
        System.out.printf("Time: %7.2f", secs);
        System.out.printf(" items per sec: %14.2f\n", ips);
        System.out.println(ForkJoinPool.commonPool());
    }

    static final class Sub implements Flow.Subscriber<Boolean> {
        int count;
        Flow.Subscription subscription;
        public void onSubscribe(Flow.Subscription s) {
            (subscription = s).request(ITEMS);
        }
        public void onNext(Boolean b) {
            if (b) ++count;
        }
        public void onComplete() {
            if (count != ITEMS)
                System.out.println("Error: remaining " + (ITEMS - count));
            phaser.arrive();
        }
        public void onError(Throwable t) { t.printStackTrace(); }
    }

    static final class Pub extends RecursiveAction {
        public void compute() {
            SubmissionPublisher<Boolean> p =
                new SubmissionPublisher<Boolean>();
            p.subscribe(new Sub());
            for (int i = 0; i < ITEMS; ++i)
                p.submit(Boolean.TRUE);
            p.close();
        }
    }

}