test/loops/ConcurrentQueueLoops.java

/*
 * @test %I% %E%
 * @bug 4486658
 * @compile -source 1.5 ConcurrentQueueLoops.java
 * @run main/timeout=230 ConcurrentQueueLoops
 * @summary Checks that a set of threads can repeatedly get and modify items
 */
/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain. Use, modify, and
 * redistribute this code in any way without acknowledgement.
 */

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 even = new Integer(42);
    static final Integer odd = new Integer(17);
    static int workMask;
    static final long RUN_TIME_NANOS = 5 * 1000L * 1000L * 1000L;

    public static void main(String[] args) throws Exception {
        int maxStages = 48;
        int work = 32;
        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;
        Stage (Queue<Integer> q, CyclicBarrier b) {
            queue = q; 
            barrier = b;
        }

        static int compute127(int l) {
            l = LoopHelpers.compute1(l);
            l = LoopHelpers.compute2(l);
            l = LoopHelpers.compute3(l);
            l = LoopHelpers.compute4(l);
            l = LoopHelpers.compute5(l);
            l = LoopHelpers.compute6(l);
            return l;
        }

        static int compute(int l) {
            if (l == 0) 
                return (int)System.nanoTime();
            int nn =  l & workMask;
            while (nn-- > 0) 
                l = compute127(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;
                for (;;) {
                    l = compute(l);
                    Integer item = queue.poll();
                    if (item != null) {
                        l += item.intValue();
                        ++takes;
                    } else if ((misses++ & 255) == 0 &&
                               System.nanoTime() >= stopTime) {
                        break;
                    } else {
                        Integer a = ((l++ & 4)== 0)? even : odd;
                        queue.add(a);
                    }
                }
                return new Integer(takes);
            }
            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)));

        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 tpi = time / total;
        if (print)
            System.out.println(LoopHelpers.rightJustify(tpi) + " ns per item");
        
    }

}
Revision:	1.2
Committed:	Tue Oct 4 20:09:41 2005 UTC (18 years, 7 months ago) by dl
Branch:	MAIN
Changes since 1.1:	+62 -39 lines
Log Message:	Add collections tests
#	Content
1	/*
2	* @test %I% %E%
3	* @bug 4486658
4	* @compile -source 1.5 ConcurrentQueueLoops.java
5	* @run main/timeout=230 ConcurrentQueueLoops
6	* @summary Checks that a set of threads can repeatedly get and modify items
7	*/
8	/*
9	* Written by Doug Lea with assistance from members of JCP JSR-166
10	* Expert Group and released to the public domain. Use, modify, and
11	* redistribute this code in any way without acknowledgement.
12	*/
13
14	import java.util.*;
15	import java.util.concurrent.*;
16	import java.util.concurrent.locks.*;
17	import java.util.concurrent.atomic.*;
18
19	public class ConcurrentQueueLoops {
20	static final ExecutorService pool = Executors.newCachedThreadPool();
21	static boolean print = false;
22	static final Integer even = new Integer(42);
23	static final Integer odd = new Integer(17);
24	static int workMask;
25	static final long RUN_TIME_NANOS = 5 * 1000L * 1000L * 1000L;
26
27	public static void main(String[] args) throws Exception {
28	int maxStages = 48;
29	int work = 32;
30	Class klass = null;
31	if (args.length > 0) {
32	try {
33	klass = Class.forName(args[0]);
34	} catch(ClassNotFoundException e) {
35	throw new RuntimeException("Class " + args[0] + " not found.");
36	}
37	}
38
39	if (args.length > 1)
40	maxStages = Integer.parseInt(args[1]);
41
42	if (args.length > 2)
43	work = Integer.parseInt(args[2]);
44
45	workMask = work - 1;
46	System.out.print("Class: " + klass.getName());
47	System.out.print(" stages: " + maxStages);
48	System.out.println(" work: " + work);
49
50	print = false;
51	System.out.println("Warmup...");
52	oneRun(klass, 4);
53	Thread.sleep(100);
54	oneRun(klass, 1);
55	Thread.sleep(100);
56	print = true;
57
58	int k = 1;
59	for (int i = 1; i <= maxStages;) {
60	oneRun(klass, i);
61	if (i == k) {
62	k = i << 1;
63	i = i + (i >>> 1);
64	}
65	else
66	i = k;
67	}
68	pool.shutdown();
69	}
70
71	static final class Stage implements Callable<Integer> {
72	final Queue<Integer> queue;
73	final CyclicBarrier barrier;
74	Stage (Queue<Integer> q, CyclicBarrier b) {
75	queue = q;
76	barrier = b;
77	}
78
79	static int compute127(int l) {
80	l = LoopHelpers.compute1(l);
81	l = LoopHelpers.compute2(l);
82	l = LoopHelpers.compute3(l);
83	l = LoopHelpers.compute4(l);
84	l = LoopHelpers.compute5(l);
85	l = LoopHelpers.compute6(l);
86	return l;
87	}
88
89	static int compute(int l) {
90	if (l == 0)
91	return (int)System.nanoTime();
92	int nn = l & workMask;
93	while (nn-- > 0)
94	l = compute127(l);
95	return l;
96	}
97
98	public Integer call() {
99	try {
100	barrier.await();
101	long now = System.nanoTime();
102	long stopTime = now + RUN_TIME_NANOS;
103	int l = (int)now;
104	int takes = 0;
105	int misses = 0;
106	for (;;) {
107	l = compute(l);
108	Integer item = queue.poll();
109	if (item != null) {
110	l += item.intValue();
111	++takes;
112	} else if ((misses++ & 255) == 0 &&
113	System.nanoTime() >= stopTime) {
114	break;
115	} else {
116	Integer a = ((l++ & 4)== 0)? even : odd;
117	queue.add(a);
118	}
119	}
120	return new Integer(takes);
121	}
122	catch (Exception ie) {
123	ie.printStackTrace();
124	throw new Error("Call loop failed");
125	}
126	}
127	}
128
129	static void oneRun(Class klass, int n) throws Exception {
130	Queue<Integer> q = (Queue<Integer>)klass.newInstance();
131	LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
132	CyclicBarrier barrier = new CyclicBarrier(n + 1, timer);
133	ArrayList<Future<Integer>> results = new ArrayList<Future<Integer>>(n);
134	for (int i = 0; i < n; ++i)
135	results.add(pool.submit(new Stage(q, barrier)));
136
137	if (print)
138	System.out.print("Threads: " + n + "\t:");
139	barrier.await();
140	int total = 0;
141	for (int i = 0; i < n; ++i) {
142	Future<Integer> f = results.get(i);
143	Integer r = f.get();
144	total += r.intValue();
145	}
146	long endTime = System.nanoTime();
147	long time = endTime - timer.startTime;
148	long tpi = time / total;
149	if (print)
150	System.out.println(LoopHelpers.rightJustify(tpi) + " ns per item");
151
152	}
153
154	}