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 |
/* |
8 |
* @test |
9 |
* @bug 6785442 |
10 |
* @summary Checks race between poll and remove(Object), while |
11 |
* occasionally moonlighting as a microbenchmark. |
12 |
* @run main RemovePollRace 1234 |
13 |
*/ |
14 |
|
15 |
import java.util.concurrent.ArrayBlockingQueue; |
16 |
import java.util.concurrent.ConcurrentHashMap; |
17 |
import java.util.concurrent.ConcurrentLinkedDeque; |
18 |
import java.util.concurrent.ConcurrentLinkedQueue; |
19 |
import java.util.concurrent.CountDownLatch; |
20 |
import java.util.concurrent.LinkedBlockingDeque; |
21 |
import java.util.concurrent.LinkedBlockingQueue; |
22 |
import java.util.concurrent.LinkedTransferQueue; |
23 |
import java.util.concurrent.atomic.AtomicLong; |
24 |
import java.util.ArrayList; |
25 |
import java.util.Collection; |
26 |
import java.util.Collections; |
27 |
import java.util.List; |
28 |
import java.util.Queue; |
29 |
import java.util.Map; |
30 |
|
31 |
public class RemovePollRace { |
32 |
// Suitable for benchmarking. Overridden by args[0] for testing. |
33 |
int count = 1024 * 1024; |
34 |
|
35 |
final Map<String,String> results = new ConcurrentHashMap<>(); |
36 |
|
37 |
Collection<Queue<Boolean>> concurrentQueues() { |
38 |
List<Queue<Boolean>> queues = new ArrayList<>(); |
39 |
queues.add(new ConcurrentLinkedDeque<Boolean>()); |
40 |
queues.add(new ConcurrentLinkedQueue<Boolean>()); |
41 |
queues.add(new ArrayBlockingQueue<Boolean>(count, false)); |
42 |
queues.add(new ArrayBlockingQueue<Boolean>(count, true)); |
43 |
queues.add(new LinkedBlockingQueue<Boolean>()); |
44 |
queues.add(new LinkedBlockingDeque<Boolean>()); |
45 |
queues.add(new LinkedTransferQueue<Boolean>()); |
46 |
|
47 |
// Following additional implementations are available from: |
48 |
// http://gee.cs.oswego.edu/dl/concurrency-interest/index.html |
49 |
// queues.add(new SynchronizedLinkedListQueue<Boolean>()); |
50 |
|
51 |
// Avoid "first fast, second slow" benchmark effect. |
52 |
Collections.shuffle(queues); |
53 |
return queues; |
54 |
} |
55 |
|
56 |
void prettyPrintResults() { |
57 |
List<String> classNames = new ArrayList<>(results.keySet()); |
58 |
Collections.sort(classNames); |
59 |
int maxClassNameLength = 0; |
60 |
int maxNanosLength = 0; |
61 |
for (String name : classNames) { |
62 |
if (maxClassNameLength < name.length()) |
63 |
maxClassNameLength = name.length(); |
64 |
if (maxNanosLength < results.get(name).length()) |
65 |
maxNanosLength = results.get(name).length(); |
66 |
} |
67 |
String format = String.format("%%%ds %%%ds nanos/item%%n", |
68 |
maxClassNameLength, maxNanosLength); |
69 |
for (String name : classNames) |
70 |
System.out.printf(format, name, results.get(name)); |
71 |
} |
72 |
|
73 |
void test(String[] args) throws Throwable { |
74 |
if (args.length > 0) |
75 |
count = Integer.valueOf(args[0]); |
76 |
|
77 |
// Warmup |
78 |
for (Queue<Boolean> queue : concurrentQueues()) |
79 |
test(queue); |
80 |
results.clear(); |
81 |
for (Queue<Boolean> queue : concurrentQueues()) |
82 |
test(queue); |
83 |
|
84 |
prettyPrintResults(); |
85 |
} |
86 |
|
87 |
void await(CountDownLatch latch) { |
88 |
try { latch.await(); } |
89 |
catch (InterruptedException e) { unexpected(e); } |
90 |
} |
91 |
|
92 |
void test(final Queue<Boolean> q) throws Throwable { |
93 |
long t0 = System.nanoTime(); |
94 |
final int SPINS = 5; |
95 |
final AtomicLong removes = new AtomicLong(0); |
96 |
final AtomicLong polls = new AtomicLong(0); |
97 |
|
98 |
// We need at least 3 threads, but we don't want to use too |
99 |
// many on massively multi-core systems. |
100 |
final int cpus = Runtime.getRuntime().availableProcessors(); |
101 |
final int threadsToUse = Math.max(3, Math.min(cpus, 16)); |
102 |
final int adderCount = threadsToUse / 3; |
103 |
final int removerCount = adderCount; |
104 |
final int pollerCount = removerCount; |
105 |
final int threadCount = adderCount + removerCount + pollerCount; |
106 |
|
107 |
final CountDownLatch startingGate = new CountDownLatch(1); |
108 |
final CountDownLatch addersDone = new CountDownLatch(adderCount); |
109 |
final Runnable remover = new Runnable() { |
110 |
public void run() { |
111 |
await(startingGate); |
112 |
int spins = 0; |
113 |
for (;;) { |
114 |
boolean quittingTime = (addersDone.getCount() == 0); |
115 |
if (q.remove(Boolean.TRUE)) |
116 |
removes.getAndIncrement(); |
117 |
else if (quittingTime) |
118 |
break; |
119 |
else if (++spins > SPINS) { |
120 |
Thread.yield(); |
121 |
spins = 0; |
122 |
}}}}; |
123 |
final Runnable poller = new Runnable() { |
124 |
public void run() { |
125 |
await(startingGate); |
126 |
int spins = 0; |
127 |
for (;;) { |
128 |
boolean quittingTime = (addersDone.getCount() == 0); |
129 |
if (q.poll() == Boolean.TRUE) |
130 |
polls.getAndIncrement(); |
131 |
else if (quittingTime) |
132 |
break; |
133 |
else if (++spins > SPINS) { |
134 |
Thread.yield(); |
135 |
spins = 0; |
136 |
}}}}; |
137 |
final Runnable adder = new Runnable() { |
138 |
public void run() { |
139 |
await(startingGate); |
140 |
for (int i = 0; i < count; i++) { |
141 |
for (;;) { |
142 |
try { q.add(Boolean.TRUE); break; } |
143 |
catch (IllegalStateException e) { Thread.yield(); } |
144 |
} |
145 |
} |
146 |
addersDone.countDown(); |
147 |
}}; |
148 |
|
149 |
final List<Thread> adders = new ArrayList<>(); |
150 |
final List<Thread> removers = new ArrayList<>(); |
151 |
final List<Thread> pollers = new ArrayList<>(); |
152 |
for (int i = 0; i < adderCount; i++) |
153 |
adders.add(checkedThread(adder)); |
154 |
for (int i = 0; i < removerCount; i++) |
155 |
removers.add(checkedThread(remover)); |
156 |
for (int i = 0; i < pollerCount; i++) |
157 |
pollers.add(checkedThread(poller)); |
158 |
|
159 |
final List<Thread> allThreads = new ArrayList<>(); |
160 |
allThreads.addAll(removers); |
161 |
allThreads.addAll(pollers); |
162 |
allThreads.addAll(adders); |
163 |
|
164 |
for (Thread t : allThreads) |
165 |
t.start(); |
166 |
startingGate.countDown(); |
167 |
for (Thread t : allThreads) |
168 |
t.join(); |
169 |
|
170 |
String className = q.getClass().getSimpleName(); |
171 |
long elapsed = System.nanoTime() - t0; |
172 |
int nanos = (int) ((double) elapsed / (adderCount * count)); |
173 |
results.put(className, String.valueOf(nanos)); |
174 |
if (removes.get() + polls.get() != adderCount * count) { |
175 |
String msg = String.format |
176 |
("class=%s removes=%s polls=%d count=%d", |
177 |
className, removes.get(), polls.get(), count); |
178 |
fail(msg); |
179 |
} |
180 |
} |
181 |
|
182 |
//--------------------- Infrastructure --------------------------- |
183 |
volatile int passed = 0, failed = 0; |
184 |
void pass() {passed++;} |
185 |
void fail() {failed++; Thread.dumpStack();} |
186 |
void fail(String msg) {System.err.println(msg); fail();} |
187 |
void unexpected(Throwable t) {failed++; t.printStackTrace();} |
188 |
void check(boolean cond) {if (cond) pass(); else fail();} |
189 |
void equal(Object x, Object y) { |
190 |
if (x == null ? y == null : x.equals(y)) pass(); |
191 |
else fail(x + " not equal to " + y);} |
192 |
public static void main(String[] args) throws Throwable { |
193 |
new RemovePollRace().instanceMain(args);} |
194 |
public void instanceMain(String[] args) throws Throwable { |
195 |
try {test(args);} catch (Throwable t) {unexpected(t);} |
196 |
System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed); |
197 |
if (failed > 0) throw new AssertionError("Some tests failed");} |
198 |
Thread checkedThread(final Runnable r) { |
199 |
return new Thread() {public void run() { |
200 |
try {r.run();} catch (Throwable t) {unexpected(t);}}};} |
201 |
} |