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.ArrayList;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

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;

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