test/loops/SubmissionPublisherLoops3.java

/*
 * 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 CONSUMERS subscribers,
 * each sent ITEMS items, with CAP buffering; repeats REPS times
 */
public class SubmissionPublisherLoops3 {
    static final int ITEMS      = 1 << 20;
    static final int PRODUCERS  = 32;
    static final int CONSUMERS  = 32;
    static final int CAP        = Flow.defaultBufferSize();
    static final int REPS = 9;

    static final Phaser phaser = new Phaser(PRODUCERS * CONSUMERS + 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 +
                           " CONSUMERS: " + CONSUMERS +
                           " CAP: " + CAP);
        for (int rep = 0; rep < reps; ++rep) {
            oneRun();
            Thread.sleep(1000);
        }
    }

    static void oneRun() throws Exception {
        long nitems = (long)ITEMS * PRODUCERS * CONSUMERS;
        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(CAP);
        }
        public void onNext(Boolean b) {
            if (b && (++count & ((CAP >>> 1) - 1)) == 0)
                subscription.request(CAP >>> 1);
        }
        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 {
        final SubmissionPublisher<Boolean> pub =
            new SubmissionPublisher<Boolean>(ForkJoinPool.commonPool(), CAP);
        public void compute() {
            SubmissionPublisher<Boolean> p = pub;
            for (int i = 0; i < CONSUMERS; ++i)
                p.subscribe(new Sub());
            for (int i = 0; i < ITEMS; ++i)
                p.submit(Boolean.TRUE);
            p.close();
        }
    }

}
Revision:	1.4
Committed:	Sun Oct 11 01:02:20 2015 UTC (8 years, 7 months ago) by jsr166
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.3:	+2 -2 lines
Log Message:	javadoc style
#	User	Rev	Content
1	dl	1.1	/*
2			* Written by Doug Lea with assistance from members of JCP JSR-166
3			* Expert Group and released to the public domain, as explained at
4			* http://creativecommons.org/publicdomain/zero/1.0/
5			*/
6
7			import java.util.*;
8			import java.util.concurrent.*;
9
10			/**
11	jsr166	1.4	* Creates PRODUCERS publishers each with CONSUMERS subscribers,
12			* each sent ITEMS items, with CAP buffering; repeats REPS times
13	dl	1.1	*/
14			public class SubmissionPublisherLoops3 {
15			static final int ITEMS = 1 << 20;
16			static final int PRODUCERS = 32;
17			static final int CONSUMERS = 32;
18			static final int CAP = Flow.defaultBufferSize();
19			static final int REPS = 9;
20
21			static final Phaser phaser = new Phaser(PRODUCERS * CONSUMERS + 1);
22
23			public static void main(String[] args) throws Exception {
24			int reps = REPS;
25			if (args.length > 0)
26			reps = Integer.parseInt(args[0]);
27	jsr166	1.2
28			System.out.println("ITEMS: " + ITEMS +
29			" PRODUCERS: " + PRODUCERS +
30	dl	1.1	" CONSUMERS: " + CONSUMERS +
31			" CAP: " + CAP);
32			for (int rep = 0; rep < reps; ++rep) {
33	dl	1.3	oneRun();
34			Thread.sleep(1000);
35	dl	1.1	}
36			}
37
38	dl	1.3	static void oneRun() throws Exception {
39			long nitems = (long)ITEMS * PRODUCERS * CONSUMERS;
40			long startTime = System.nanoTime();
41			for (int i = 0; i < PRODUCERS; ++i)
42			new Pub().fork();
43			phaser.arriveAndAwaitAdvance();
44			long elapsed = System.nanoTime() - startTime;
45			double secs = ((double)elapsed) / (1000L * 1000 * 1000);
46			double ips = nitems / secs;
47			System.out.printf("Time: %7.2f", secs);
48			System.out.printf(" items per sec: %14.2f\n", ips);
49			System.out.println(ForkJoinPool.commonPool());
50			}
51
52	dl	1.1	static final class Sub implements Flow.Subscriber<Boolean> {
53			int count;
54			Flow.Subscription subscription;
55	jsr166	1.2	public void onSubscribe(Flow.Subscription s) {
56			(subscription = s).request(CAP);
57	dl	1.1	}
58	jsr166	1.2	public void onNext(Boolean b) {
59	dl	1.1	if (b && (++count & ((CAP >>> 1) - 1)) == 0)
60			subscription.request(CAP >>> 1);
61			}
62	jsr166	1.2	public void onComplete() {
63	dl	1.1	if (count != ITEMS)
64			System.out.println("Error: remaining " + (ITEMS - count));
65	jsr166	1.2	phaser.arrive();
66	dl	1.1	}
67			public void onError(Throwable t) { t.printStackTrace(); }
68			}
69
70			static final class Pub extends RecursiveAction {
71	jsr166	1.2	final SubmissionPublisher<Boolean> pub =
72	dl	1.1	new SubmissionPublisher<Boolean>(ForkJoinPool.commonPool(), CAP);
73			public void compute() {
74			SubmissionPublisher<Boolean> p = pub;
75			for (int i = 0; i < CONSUMERS; ++i)
76			p.subscribe(new Sub());
77	jsr166	1.2	for (int i = 0; i < ITEMS; ++i)
78	dl	1.1	p.submit(Boolean.TRUE);
79			p.close();
80			}
81			}
82
83			}