test/loops/SubmissionPublisherLoops2.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.*;

/**
 * One FJ publisher, many subscribers
 */
public class SubmissionPublisherLoops2 {
    static final int ITEMS = 1 << 20;
    static final int CONSUMERS  = 64;
    static final int CAP        = Flow.defaultBufferSize();
    static final int REPS = 9;

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

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

    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();
        }
    }

    static final long NPS = (1000L * 1000 * 1000);

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

    static void oneRun() throws Exception {
        long startTime = System.nanoTime();
        new Pub().fork();
        phaser.arriveAndAwaitAdvance();
        long elapsed = System.nanoTime() - startTime;
        double secs = ((double)elapsed) / NPS;
        System.out.printf("\tTime: %7.3f\n", secs);
    }

}
Revision:	1.4
Committed:	Sat Sep 12 20:30:46 2015 UTC (8 years, 8 months ago) by dl
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.3:	+13 -8 lines
Log Message:	Use common conventions
#	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			* One FJ publisher, many subscribers
12			*/
13			public class SubmissionPublisherLoops2 {
14			static final int ITEMS = 1 << 20;
15			static final int CONSUMERS = 64;
16			static final int CAP = Flow.defaultBufferSize();
17			static final int REPS = 9;
18
19			static final Phaser phaser = new Phaser(CONSUMERS + 1);
20
21	dl	1.4	static final class Sub implements Flow.Subscriber<Boolean> {
22	dl	1.1	Flow.Subscription sn;
23			int count;
24	jsr166	1.2	public void onSubscribe(Flow.Subscription s) {
25			(sn = s).request(CAP);
26	dl	1.1	}
27	jsr166	1.2	public void onNext(Boolean t) {
28	dl	1.1	if ((++count & (CAP - 1)) == (CAP >>> 1))
29			sn.request(CAP);
30			}
31			public void onError(Throwable t) { t.printStackTrace(); }
32	jsr166	1.2	public void onComplete() {
33	dl	1.1	if (count != ITEMS)
34			System.out.println("Error: remaining " + (ITEMS - count));
35	jsr166	1.2	phaser.arrive();
36	dl	1.1	}
37			}
38
39			static final class Pub extends RecursiveAction {
40	jsr166	1.2	final SubmissionPublisher<Boolean> pub =
41	dl	1.1	new SubmissionPublisher<Boolean>(ForkJoinPool.commonPool(), CAP);
42			public void compute() {
43			SubmissionPublisher<Boolean> p = pub;
44			for (int i = 0; i < CONSUMERS; ++i)
45			p.subscribe(new Sub());
46	jsr166	1.2	for (int i = 0; i < ITEMS; ++i)
47	dl	1.1	p.submit(Boolean.TRUE);
48			p.close();
49			}
50			}
51
52			static final long NPS = (1000L * 1000 * 1000);
53
54			public static void main(String[] args) throws Exception {
55			int reps = REPS;
56			if (args.length > 0)
57			reps = Integer.parseInt(args[0]);
58	jsr166	1.2
59			System.out.println("ITEMS: " + ITEMS +
60	dl	1.1	" CONSUMERS: " + CONSUMERS +
61			" CAP: " + CAP);
62			for (int rep = 0; rep < reps; ++rep) {
63	dl	1.4	oneRun();
64			Thread.sleep(1000);
65	dl	1.1	}
66			}
67	dl	1.4
68			static void oneRun() throws Exception {
69			long startTime = System.nanoTime();
70			new Pub().fork();
71			phaser.arriveAndAwaitAdvance();
72			long elapsed = System.nanoTime() - startTime;
73			double secs = ((double)elapsed) / NPS;
74			System.out.printf("\tTime: %7.3f\n", secs);
75			}
76
77	dl	1.1	}