1 /*
2 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
3 *
4 * This code is free software; you can redistribute it and/or modify it
5 * under the terms of the GNU General Public License version 2 only, as
6 * published by the Free Software Foundation.
7 *
8 * This code is distributed in the hope that it will be useful, but WITHOUT
9 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
11 * version 2 for more details (a copy is included in the LICENSE file that
12 * accompanied this code).
13 *
14 * You should have received a copy of the GNU General Public License version
15 * 2 along with this work; if not, write to the Free Software Foundation,
16 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
17 *
18 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
19 * or visit www.oracle.com if you need additional information or have any
20 * questions.
21 */
22
23 /*
24 * This file is available under and governed by the GNU General Public
25 * License version 2 only, as published by the Free Software Foundation.
26 * However, the following notice accompanied the original version of this
27 * file:
28 *
29 * Written by Doug Lea with assistance from members of JCP JSR-166
30 * Expert Group and released to the public domain, as explained at
31 * http://creativecommons.org/publicdomain/zero/1.0/
32 */
33
34 /*
35 * @test
36 * @bug 4486658 8031651
37 * @compile SingleProducerMultipleConsumerLoops.java
38 * @run main/timeout=600 SingleProducerMultipleConsumerLoops
39 * @summary check ordering for blocking queues with 1 producer and multiple consumers
40 */
41
42 import java.util.concurrent.*;
43
44 public class SingleProducerMultipleConsumerLoops {
45 static final int CAPACITY = 100;
46
47 static final ExecutorService pool = Executors.newCachedThreadPool();
48 static boolean print = false;
49
50 public static void main(String[] args) throws Exception {
51 int maxConsumers = 5;
52 int iters = 10000;
53
54 if (args.length > 0)
55 maxConsumers = Integer.parseInt(args[0]);
56
57 print = false;
58 System.out.println("Warmup...");
59 oneTest(1, 10000);
60 Thread.sleep(100);
61 oneTest(2, 10000);
62 Thread.sleep(100);
63 print = true;
64
65 for (int i = 1; i <= maxConsumers; i += (i+1) >>> 1) {
66 System.out.println("----------------------------------------");
67 System.out.println("Consumers: " + i);
68 oneTest(i, iters);
69 Thread.sleep(100);
70 }
71 pool.shutdown();
72 if (! pool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS))
73 throw new Error();
74 }
75
76 static void oneTest(int consumers, int iters) throws Exception {
77 oneRun(new ArrayBlockingQueue<Integer>(CAPACITY), consumers, iters);
78 oneRun(new LinkedBlockingQueue<Integer>(CAPACITY), consumers, iters);
79 oneRun(new LinkedBlockingDeque<Integer>(CAPACITY), consumers, iters);
80 oneRun(new LinkedTransferQueue<Integer>(), consumers, iters);
81 oneRun(new PriorityBlockingQueue<Integer>(), consumers, iters);
82 oneRun(new SynchronousQueue<Integer>(), consumers, iters);
83 if (print)
84 System.out.println("fair implementations:");
85 oneRun(new SynchronousQueue<Integer>(true), consumers, iters);
86 oneRun(new ArrayBlockingQueue<Integer>(CAPACITY, true), consumers, iters);
87 }
88
89 abstract static class Stage implements Runnable {
90 final int iters;
91 final BlockingQueue<Integer> queue;
92 final CyclicBarrier barrier;
93 volatile int result;
94 Stage(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
95 queue = q;
96 barrier = b;
97 this.iters = iters;
98 }
99 }
100
101 static class Producer extends Stage {
102 Producer(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
103 super(q, b, iters);
104 }
105
106 public void run() {
107 try {
108 barrier.await();
109 for (int i = 0; i < iters; ++i) {
110 queue.put(new Integer(i));
111 }
112 barrier.await();
113 result = 432;
114 }
115 catch (Exception ie) {
116 ie.printStackTrace();
117 return;
118 }
119 }
120 }
121
122 static class Consumer extends Stage {
123 Consumer(BlockingQueue<Integer> q, CyclicBarrier b, int iters) {
124 super(q, b, iters);
125 }
126
127 public void run() {
128 try {
129 barrier.await();
130 int l = 0;
131 int s = 0;
132 int last = -1;
133 for (int i = 0; i < iters; ++i) {
134 Integer item = queue.take();
135 int v = item.intValue();
136 if (v < last)
137 throw new Error("Out-of-Order transfer");
138 last = v;
139 l = LoopHelpers.compute1(v);
140 s += l;
141 }
142 barrier.await();
143 result = s;
144 }
145 catch (Exception ie) {
146 ie.printStackTrace();
147 return;
148 }
149 }
150
151 }
152
153 static void oneRun(BlockingQueue<Integer> q, int nconsumers, int iters) throws Exception {
154 if (print)
155 System.out.printf("%-18s", q.getClass().getSimpleName());
156 LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
157 CyclicBarrier barrier = new CyclicBarrier(nconsumers + 2, timer);
158 pool.execute(new Producer(q, barrier, iters * nconsumers));
159 for (int i = 0; i < nconsumers; ++i) {
160 pool.execute(new Consumer(q, barrier, iters));
161 }
162 barrier.await();
163 barrier.await();
164 long time = timer.getTime();
165 if (print)
166 System.out.println("\t: " + LoopHelpers.rightJustify(time / (iters * nconsumers)) + " ns per transfer");
167 }
168
169 }