test/loops/ConcurrentQueueLoops.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.*;
import java.util.concurrent.locks.*;
import java.util.concurrent.atomic.*;

public class ConcurrentQueueLoops {
    static final ExecutorService pool = Executors.newCachedThreadPool();
    static boolean print = false;
    static final Integer zero = new Integer(0);
    static final Integer one = new Integer(1);
    static int workMask;
    static final long RUN_TIME_NANOS = 5 * 1000L * 1000L * 1000L;
    static final int BATCH_SIZE = 8;

    public static void main(String[] args) throws Exception {
        int maxStages = 100;
        int work = 1024;
        Class klass = null;
        if (args.length > 0) {
            try {
                klass = Class.forName(args[0]);
            } catch (ClassNotFoundException e) {
                throw new RuntimeException("Class " + args[0] + " not found.");
            }
        }

        if (args.length > 1)
            maxStages = Integer.parseInt(args[1]);

        if (args.length > 2)
            work = Integer.parseInt(args[2]);

        workMask = work - 1;
        System.out.print("Class: " + klass.getName());
        System.out.print(" stages: " + maxStages);
        System.out.println(" work: " + work);

        print = false;
        System.out.println("Warmup...");
        //        oneRun(klass, 4);
        //
        Thread.sleep(100);
        oneRun(klass, 1);
        Thread.sleep(100);
        print = true;

        int k = 1;
        for (int i = 1; i <= maxStages;) {
            oneRun(klass, i);
            if (i == k) {
                k = i << 1;
                i = i + (i >>> 1);
            }
            else
                i = k;
        }
        pool.shutdown();
    }

    static final class Stage implements Callable<Integer> {
        final Queue<Integer> queue;
        final CyclicBarrier barrier;
        final int nthreads;
        Stage(Queue<Integer> q, CyclicBarrier b, int nthreads) {
            queue = q;
            barrier = b;
            this.nthreads = nthreads;
        }

        static int compute(int l) {
            if (l == 0)
                return (int) System.nanoTime();
            int nn =  (l >>> 7) & workMask;
            while (nn-- > 0)
                l = LoopHelpers.compute6(l);
            return l;
        }

        public Integer call() {
            try {
                barrier.await();
                long now = System.nanoTime();
                long stopTime = now + RUN_TIME_NANOS;
                int l = (int) now;
                int takes = 0;
                int misses = 0;
                int lmask = 1;
                for (;;) {
                    l = compute(l);
                    Integer item = queue.poll();
                    if (item != null) {
                        ++takes;
                        if (item == one)
                            l = LoopHelpers.compute6(l);
                    } else if ((misses++ & 255) == 0 &&
                               System.nanoTime() >= stopTime) {
                        return Integer.valueOf(takes);
                    } else {
                        for (int i = 0; i < BATCH_SIZE; ++i) {
                            queue.offer(((l & lmask)== 0) ? zero : one);
                            if ((lmask <<= 1) == 0) lmask = 1;
                            if (i != 0) l = compute(l);
                        }
                    }
                }
            }
            catch (Exception ie) {
                ie.printStackTrace();
                throw new Error("Call loop failed");
            }
        }
    }

    static void oneRun(Class klass, int n) throws Exception {
        Queue<Integer> q = (Queue<Integer>) klass.newInstance();
        LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
        CyclicBarrier barrier = new CyclicBarrier(n + 1, timer);
        ArrayList<Future<Integer>> results = new ArrayList<Future<Integer>>(n);
        for (int i = 0; i < n; ++i)
            results.add(pool.submit(new Stage(q, barrier, n)));

        if (print)
            System.out.print("Threads: " + n + "\t:");
        barrier.await();
        int total = 0;
        for (int i = 0; i < n; ++i) {
            Future<Integer> f = results.get(i);
            Integer r = f.get();
            total += r.intValue();
        }
        long endTime = System.nanoTime();
        long time = endTime - timer.startTime;
        long ips = 1000000000L * total / time;

        if (print)
            System.out.print(LoopHelpers.rightJustify(ips) + " items per sec");
        if (print)
            System.out.println();
    }

}
Revision:	1.13
Committed:	Thu Apr 14 23:16:10 2011 UTC (13 years, 1 month ago) by jsr166
Branch:	MAIN
CVS Tags:	release-1_7_0
Changes since 1.12:	+1 -1 lines
Log Message:	whitespace
#	Content
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	import java.util.concurrent.locks.*;
10	import java.util.concurrent.atomic.*;
11
12	public class ConcurrentQueueLoops {
13	static final ExecutorService pool = Executors.newCachedThreadPool();
14	static boolean print = false;
15	static final Integer zero = new Integer(0);
16	static final Integer one = new Integer(1);
17	static int workMask;
18	static final long RUN_TIME_NANOS = 5 * 1000L * 1000L * 1000L;
19	static final int BATCH_SIZE = 8;
20
21	public static void main(String[] args) throws Exception {
22	int maxStages = 100;
23	int work = 1024;
24	Class klass = null;
25	if (args.length > 0) {
26	try {
27	klass = Class.forName(args[0]);
28	} catch (ClassNotFoundException e) {
29	throw new RuntimeException("Class " + args[0] + " not found.");
30	}
31	}
32
33	if (args.length > 1)
34	maxStages = Integer.parseInt(args[1]);
35
36	if (args.length > 2)
37	work = Integer.parseInt(args[2]);
38
39	workMask = work - 1;
40	System.out.print("Class: " + klass.getName());
41	System.out.print(" stages: " + maxStages);
42	System.out.println(" work: " + work);
43
44	print = false;
45	System.out.println("Warmup...");
46	// oneRun(klass, 4);
47	//
48	Thread.sleep(100);
49	oneRun(klass, 1);
50	Thread.sleep(100);
51	print = true;
52
53	int k = 1;
54	for (int i = 1; i <= maxStages;) {
55	oneRun(klass, i);
56	if (i == k) {
57	k = i << 1;
58	i = i + (i >>> 1);
59	}
60	else
61	i = k;
62	}
63	pool.shutdown();
64	}
65
66	static final class Stage implements Callable<Integer> {
67	final Queue<Integer> queue;
68	final CyclicBarrier barrier;
69	final int nthreads;
70	Stage(Queue<Integer> q, CyclicBarrier b, int nthreads) {
71	queue = q;
72	barrier = b;
73	this.nthreads = nthreads;
74	}
75
76	static int compute(int l) {
77	if (l == 0)
78	return (int) System.nanoTime();
79	int nn = (l >>> 7) & workMask;
80	while (nn-- > 0)
81	l = LoopHelpers.compute6(l);
82	return l;
83	}
84
85	public Integer call() {
86	try {
87	barrier.await();
88	long now = System.nanoTime();
89	long stopTime = now + RUN_TIME_NANOS;
90	int l = (int) now;
91	int takes = 0;
92	int misses = 0;
93	int lmask = 1;
94	for (;;) {
95	l = compute(l);
96	Integer item = queue.poll();
97	if (item != null) {
98	++takes;
99	if (item == one)
100	l = LoopHelpers.compute6(l);
101	} else if ((misses++ & 255) == 0 &&
102	System.nanoTime() >= stopTime) {
103	return Integer.valueOf(takes);
104	} else {
105	for (int i = 0; i < BATCH_SIZE; ++i) {
106	queue.offer(((l & lmask)== 0) ? zero : one);
107	if ((lmask <<= 1) == 0) lmask = 1;
108	if (i != 0) l = compute(l);
109	}
110	}
111	}
112	}
113	catch (Exception ie) {
114	ie.printStackTrace();
115	throw new Error("Call loop failed");
116	}
117	}
118	}
119
120	static void oneRun(Class klass, int n) throws Exception {
121	Queue<Integer> q = (Queue<Integer>) klass.newInstance();
122	LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
123	CyclicBarrier barrier = new CyclicBarrier(n + 1, timer);
124	ArrayList<Future<Integer>> results = new ArrayList<Future<Integer>>(n);
125	for (int i = 0; i < n; ++i)
126	results.add(pool.submit(new Stage(q, barrier, n)));
127
128	if (print)
129	System.out.print("Threads: " + n + "\t:");
130	barrier.await();
131	int total = 0;
132	for (int i = 0; i < n; ++i) {
133	Future<Integer> f = results.get(i);
134	Integer r = f.get();
135	total += r.intValue();
136	}
137	long endTime = System.nanoTime();
138	long time = endTime - timer.startTime;
139	long ips = 1000000000L * total / time;
140
141	if (print)
142	System.out.print(LoopHelpers.rightJustify(ips) + " items per sec");
143	if (print)
144	System.out.println();
145	}
146
147	}