ViewVC Help

View File | Revision Log | Show Annotations | Download File | Root Listing

root/jsr166/jsr166/src/test/tck/CyclicBarrierTest.java

Comparing jsr166/src/test/tck/CyclicBarrierTest.java (file contents):
Revision 1.15 by jsr166, Wed Aug 25 00:07:03 2010 UTC vs.
Revision 1.33 by jsr166, Mon Sep 9 00:03:07 2019 UTC

#	Line 1 | Line 1
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/licenses/publicdomain
4	>	* http://creativecommons.org/publicdomain/zero/1.0/
5		* Other contributors include Andrew Wright, Jeffrey Hayes,
6		* Pat Fisher, Mike Judd.
7		*/
8
9	–	import junit.framework.*;
10	–	import java.util.*;
11	–	import java.util.concurrent.*;
12	–	import java.util.concurrent.locks.*;
13	–	import java.util.concurrent.atomic.*;
9		import static java.util.concurrent.TimeUnit.MILLISECONDS;
10
11	+	import java.util.concurrent.BrokenBarrierException;
12	+	import java.util.concurrent.CountDownLatch;
13	+	import java.util.concurrent.CyclicBarrier;
14	+	import java.util.concurrent.ExecutorService;
15	+	import java.util.concurrent.Executors;
16	+	import java.util.concurrent.ThreadLocalRandom;
17	+	import java.util.concurrent.TimeoutException;
18	+	import java.util.concurrent.atomic.AtomicBoolean;
19	+	import java.util.concurrent.atomic.AtomicInteger;
20	+
21	+	import junit.framework.Test;
22	+	import junit.framework.TestSuite;
23	+
24		public class CyclicBarrierTest extends JSR166TestCase {
25		public static void main(String[] args) {
26	<	junit.textui.TestRunner.run(suite());
26	>	main(suite(), args);
27		}
28		public static Test suite() {
29		return new TestSuite(CyclicBarrierTest.class);
30		}
31
32	<	private volatile int countAction;
33	<	private class MyAction implements Runnable {
34	<	public void run() { ++countAction; }
32	>	/**
33	>	* Spin-waits till the number of waiters == numberOfWaiters.
34	>	*/
35	>	void awaitNumberWaiting(CyclicBarrier barrier, int numberOfWaiters) {
36	>	long startTime = System.nanoTime();
37	>	while (barrier.getNumberWaiting() != numberOfWaiters) {
38	>	if (millisElapsedSince(startTime) > LONG_DELAY_MS)
39	>	fail("timed out");
40	>	Thread.yield();
41	>	}
42		}
43
44		/**
45	<	* Creating with negative parties throws IAE
45	>	* Creating with negative parties throws IllegalArgumentException
46		*/
47		public void testConstructor1() {
48		try {
#	Line 37 | Line 52 | public class CyclicBarrierTest extends J
52		}
53
54		/**
55	<	* Creating with negative parties and no action throws IAE
55	>	* Creating with negative parties and no action throws
56	>	* IllegalArgumentException
57		*/
58		public void testConstructor2() {
59		try {
#	Line 71 | Line 87 | public class CyclicBarrierTest extends J
87		* The supplied barrier action is run at barrier
88		*/
89		public void testBarrierAction() throws Exception {
90	<	countAction = 0;
91	<	CyclicBarrier b = new CyclicBarrier(1, new MyAction());
90	>	final AtomicInteger count = new AtomicInteger(0);
91	>	final Runnable incCount = new Runnable() { public void run() {
92	>	count.getAndIncrement(); }};
93	>	CyclicBarrier b = new CyclicBarrier(1, incCount);
94		assertEquals(1, b.getParties());
95		assertEquals(0, b.getNumberWaiting());
96		b.await();
97		b.await();
98		assertEquals(0, b.getNumberWaiting());
99	<	assertEquals(countAction, 2);
99	>	assertEquals(2, count.get());
100		}
101
102		/**
#	Line 86 | Line 104 | public class CyclicBarrierTest extends J
104		*/
105		public void testTwoParties() throws Exception {
106		final CyclicBarrier b = new CyclicBarrier(2);
107	<	Thread t = new Thread(new CheckedRunnable() {
107	>	Thread t = newStartedThread(new CheckedRunnable() {
108		public void realRun() throws Exception {
109		b.await();
110		b.await();
#	Line 94 | Line 112 | public class CyclicBarrierTest extends J
112		b.await();
113		}});
114
97	–	t.start();
115		b.await();
116		b.await();
117		b.await();
118		b.await();
119	<	t.join();
119	>	awaitTermination(t);
120		}
121
105	–
122		/**
123		* An interruption in one party causes others waiting in await to
124		* throw BrokenBarrierException
125		*/
126	<	public void testAwait1_Interrupted_BrokenBarrier() throws Exception {
126	>	public void testAwait1_Interrupted_BrokenBarrier() {
127		final CyclicBarrier c = new CyclicBarrier(3);
128	+	final CountDownLatch pleaseInterrupt = new CountDownLatch(2);
129		Thread t1 = new ThreadShouldThrow(InterruptedException.class) {
130		public void realRun() throws Exception {
131	+	pleaseInterrupt.countDown();
132		c.await();
133		}};
134		Thread t2 = new ThreadShouldThrow(BrokenBarrierException.class) {
135		public void realRun() throws Exception {
136	+	pleaseInterrupt.countDown();
137		c.await();
138		}};
139
140		t1.start();
141		t2.start();
142	<	Thread.sleep(SHORT_DELAY_MS);
142	>	await(pleaseInterrupt);
143		t1.interrupt();
144	<	t1.join();
145	<	t2.join();
144	>	awaitTermination(t1);
145	>	awaitTermination(t2);
146		}
147
148		/**
#	Line 132 | Line 151 | public class CyclicBarrierTest extends J
151		*/
152		public void testAwait2_Interrupted_BrokenBarrier() throws Exception {
153		final CyclicBarrier c = new CyclicBarrier(3);
154	+	final CountDownLatch pleaseInterrupt = new CountDownLatch(2);
155		Thread t1 = new ThreadShouldThrow(InterruptedException.class) {
156		public void realRun() throws Exception {
157	+	pleaseInterrupt.countDown();
158		c.await(LONG_DELAY_MS, MILLISECONDS);
159		}};
160		Thread t2 = new ThreadShouldThrow(BrokenBarrierException.class) {
161		public void realRun() throws Exception {
162	+	pleaseInterrupt.countDown();
163		c.await(LONG_DELAY_MS, MILLISECONDS);
164		}};
165
166		t1.start();
167		t2.start();
168	<	Thread.sleep(SHORT_DELAY_MS);
168	>	await(pleaseInterrupt);
169		t1.interrupt();
170	<	t1.join();
171	<	t2.join();
170	>	awaitTermination(t1);
171	>	awaitTermination(t2);
172		}
173
174		/**
175		* A timeout in timed await throws TimeoutException
176		*/
177	<	public void testAwait3_TimeOutException() throws InterruptedException {
177	>	public void testAwait3_TimeoutException() throws InterruptedException {
178		final CyclicBarrier c = new CyclicBarrier(2);
179	<	Thread t = new ThreadShouldThrow(TimeoutException.class) {
179	>	Thread t = newStartedThread(new CheckedRunnable() {
180		public void realRun() throws Exception {
181	<	c.await(SHORT_DELAY_MS, MILLISECONDS);
182	<	}};
181	>	long startTime = System.nanoTime();
182	>	try {
183	>	c.await(timeoutMillis(), MILLISECONDS);
184	>	shouldThrow();
185	>	} catch (TimeoutException success) {}
186	>	assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
187	>	}});
188
189	<	t.start();
163	<	t.join();
189	>	awaitTermination(t);
190		}
191
192		/**
#	Line 169 | Line 195 | public class CyclicBarrierTest extends J
195		*/
196		public void testAwait4_Timeout_BrokenBarrier() throws InterruptedException {
197		final CyclicBarrier c = new CyclicBarrier(3);
198	<	Thread t1 = new ThreadShouldThrow(TimeoutException.class) {
198	>	Thread t1 = newStartedThread(new CheckedRunnable() {
199		public void realRun() throws Exception {
200	<	c.await(SHORT_DELAY_MS, MILLISECONDS);
201	<	}};
202	<	Thread t2 = new ThreadShouldThrow(BrokenBarrierException.class) {
200	>	try {
201	>	c.await(LONG_DELAY_MS, MILLISECONDS);
202	>	shouldThrow();
203	>	} catch (BrokenBarrierException success) {}
204	>	}});
205	>	Thread t2 = newStartedThread(new CheckedRunnable() {
206		public void realRun() throws Exception {
207	<	c.await(MEDIUM_DELAY_MS, MILLISECONDS);
208	<	}};
207	>	awaitNumberWaiting(c, 1);
208	>	long startTime = System.nanoTime();
209	>	try {
210	>	c.await(timeoutMillis(), MILLISECONDS);
211	>	shouldThrow();
212	>	} catch (TimeoutException success) {}
213	>	assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
214	>	}});
215
216	<	t1.start();
217	<	t2.start();
183	<	t1.join();
184	<	t2.join();
216	>	awaitTermination(t1);
217	>	awaitTermination(t2);
218		}
219
220		/**
#	Line 190 | Line 223 | public class CyclicBarrierTest extends J
223		*/
224		public void testAwait5_Timeout_BrokenBarrier() throws InterruptedException {
225		final CyclicBarrier c = new CyclicBarrier(3);
226	<	Thread t1 = new ThreadShouldThrow(TimeoutException.class) {
226	>	Thread t1 = newStartedThread(new CheckedRunnable() {
227		public void realRun() throws Exception {
228	<	c.await(SHORT_DELAY_MS, MILLISECONDS);
229	<	}};
230	<	Thread t2 = new ThreadShouldThrow(BrokenBarrierException.class) {
228	>	try {
229	>	c.await();
230	>	shouldThrow();
231	>	} catch (BrokenBarrierException success) {}
232	>	}});
233	>	Thread t2 = newStartedThread(new CheckedRunnable() {
234		public void realRun() throws Exception {
235	<	c.await();
236	<	}};
235	>	awaitNumberWaiting(c, 1);
236	>	long startTime = System.nanoTime();
237	>	try {
238	>	c.await(timeoutMillis(), MILLISECONDS);
239	>	shouldThrow();
240	>	} catch (TimeoutException success) {}
241	>	assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
242	>	}});
243
244	<	t1.start();
245	<	t2.start();
204	<	t1.join();
205	<	t2.join();
244	>	awaitTermination(t1);
245	>	awaitTermination(t2);
246		}
247
248		/**
#	Line 211 | Line 251 | public class CyclicBarrierTest extends J
251		*/
252		public void testReset_BrokenBarrier() throws InterruptedException {
253		final CyclicBarrier c = new CyclicBarrier(3);
254	+	final CountDownLatch pleaseReset = new CountDownLatch(2);
255		Thread t1 = new ThreadShouldThrow(BrokenBarrierException.class) {
256		public void realRun() throws Exception {
257	+	pleaseReset.countDown();
258		c.await();
259		}};
260		Thread t2 = new ThreadShouldThrow(BrokenBarrierException.class) {
261		public void realRun() throws Exception {
262	+	pleaseReset.countDown();
263		c.await();
264		}};
265
266		t1.start();
267		t2.start();
268	<	Thread.sleep(SHORT_DELAY_MS);
268	>	await(pleaseReset);
269	>
270	>	awaitNumberWaiting(c, 2);
271		c.reset();
272	<	t1.join();
273	<	t2.join();
272	>	awaitTermination(t1);
273	>	awaitTermination(t2);
274		}
275
276		/**
#	Line 234 | Line 279 | public class CyclicBarrierTest extends J
279		*/
280		public void testReset_NoBrokenBarrier() throws Exception {
281		final CyclicBarrier c = new CyclicBarrier(3);
282	<	Thread t1 = new Thread(new CheckedRunnable() {
282	>	c.reset();
283	>
284	>	Thread t1 = newStartedThread(new CheckedRunnable() {
285		public void realRun() throws Exception {
286		c.await();
287		}});
288	<	Thread t2 = new Thread(new CheckedRunnable() {
288	>	Thread t2 = newStartedThread(new CheckedRunnable() {
289		public void realRun() throws Exception {
290		c.await();
291		}});
292
246	–	c.reset();
247	–	t1.start();
248	–	t2.start();
293		c.await();
294	<	t1.join();
295	<	t2.join();
252	<	}
253	<
254	<	/**
255	<	* All threads block while a barrier is broken.
256	<	*/
257	<	public void testReset_Leakage() throws InterruptedException {
258	<	final CyclicBarrier c = new CyclicBarrier(2);
259	<	final AtomicBoolean done = new AtomicBoolean();
260	<	Thread t = new Thread() {
261	<	public void run() {
262	<	while (!done.get()) {
263	<	try {
264	<	while (c.isBroken())
265	<	c.reset();
266	<
267	<	c.await();
268	<	threadFail("await should not return");
269	<	}
270	<	catch (BrokenBarrierException e) {
271	<	}
272	<	catch (InterruptedException ie) {
273	<	}
274	<	}
275	<	}
276	<	};
277	<
278	<	t.start();
279	<	for (int i = 0; i < 4; i++) {
280	<	Thread.sleep(SHORT_DELAY_MS);
281	<	t.interrupt();
282	<	}
283	<	done.set(true);
284	<	t.interrupt();
285	<	t.join();
294	>	awaitTermination(t1);
295	>	awaitTermination(t2);
296		}
297
298		/**
299		* Reset of a non-broken barrier does not break barrier
300		*/
301		public void testResetWithoutBreakage() throws Exception {
292	–	final CyclicBarrier start = new CyclicBarrier(3);
302		final CyclicBarrier barrier = new CyclicBarrier(3);
303		for (int i = 0; i < 3; i++) {
304	<	Thread t1 = new Thread(new CheckedRunnable() {
304	>	final CyclicBarrier start = new CyclicBarrier(3);
305	>	Thread t1 = newStartedThread(new CheckedRunnable() {
306		public void realRun() throws Exception {
307		start.await();
308		barrier.await();
309		}});
310
311	<	Thread t2 = new Thread(new CheckedRunnable() {
311	>	Thread t2 = newStartedThread(new CheckedRunnable() {
312		public void realRun() throws Exception {
313		start.await();
314		barrier.await();
315		}});
316
307	–	t1.start();
308	–	t2.start();
317		start.await();
318		barrier.await();
319	<	t1.join();
320	<	t2.join();
319	>	awaitTermination(t1);
320	>	awaitTermination(t2);
321		assertFalse(barrier.isBroken());
322		assertEquals(0, barrier.getNumberWaiting());
323		if (i == 1) barrier.reset();
#	Line 322 | Line 330 | public class CyclicBarrierTest extends J
330		* Reset of a barrier after interruption reinitializes it.
331		*/
332		public void testResetAfterInterrupt() throws Exception {
325	–	final CyclicBarrier start = new CyclicBarrier(3);
333		final CyclicBarrier barrier = new CyclicBarrier(3);
334		for (int i = 0; i < 2; i++) {
335	+	final CyclicBarrier start = new CyclicBarrier(3);
336		Thread t1 = new ThreadShouldThrow(InterruptedException.class) {
337		public void realRun() throws Exception {
338		start.await();
#	Line 341 | Line 349 | public class CyclicBarrierTest extends J
349		t2.start();
350		start.await();
351		t1.interrupt();
352	<	t1.join();
353	<	t2.join();
352	>	awaitTermination(t1);
353	>	awaitTermination(t2);
354		assertTrue(barrier.isBroken());
355		assertEquals(0, barrier.getNumberWaiting());
356		barrier.reset();
#	Line 355 | Line 363 | public class CyclicBarrierTest extends J
363		* Reset of a barrier after timeout reinitializes it.
364		*/
365		public void testResetAfterTimeout() throws Exception {
358	–	final CyclicBarrier start = new CyclicBarrier(3);
366		final CyclicBarrier barrier = new CyclicBarrier(3);
367		for (int i = 0; i < 2; i++) {
368	<	Thread t1 = new ThreadShouldThrow(TimeoutException.class) {
369	<	public void realRun() throws Exception {
363	<	start.await();
364	<	barrier.await(MEDIUM_DELAY_MS, MILLISECONDS);
365	<	}};
366	<
367	<	Thread t2 = new ThreadShouldThrow(BrokenBarrierException.class) {
368	>	assertEquals(0, barrier.getNumberWaiting());
369	>	Thread t1 = newStartedThread(new CheckedRunnable() {
370		public void realRun() throws Exception {
371	<	start.await();
372	<	barrier.await();
373	<	}};
371	>	try {
372	>	barrier.await();
373	>	shouldThrow();
374	>	} catch (BrokenBarrierException success) {}
375	>	}});
376	>	Thread t2 = newStartedThread(new CheckedRunnable() {
377	>	public void realRun() throws Exception {
378	>	awaitNumberWaiting(barrier, 1);
379	>	long startTime = System.nanoTime();
380	>	try {
381	>	barrier.await(timeoutMillis(), MILLISECONDS);
382	>	shouldThrow();
383	>	} catch (TimeoutException success) {}
384	>	assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
385	>	}});
386
387	<	t1.start();
388	<	t2.start();
389	<	start.await();
376	<	t1.join();
377	<	t2.join();
387	>	awaitTermination(t1);
388	>	awaitTermination(t2);
389	>	assertEquals(0, barrier.getNumberWaiting());
390		assertTrue(barrier.isBroken());
391		assertEquals(0, barrier.getNumberWaiting());
392		barrier.reset();
#	Line 383 | Line 395 | public class CyclicBarrierTest extends J
395		}
396		}
397
386	–
398		/**
399		* Reset of a barrier after a failed command reinitializes it.
400		*/
401		public void testResetAfterCommandException() throws Exception {
391	–	final CyclicBarrier start = new CyclicBarrier(3);
402		final CyclicBarrier barrier =
403		new CyclicBarrier(3, new Runnable() {
404		public void run() {
405		throw new NullPointerException(); }});
406		for (int i = 0; i < 2; i++) {
407	+	final CyclicBarrier start = new CyclicBarrier(3);
408		Thread t1 = new ThreadShouldThrow(BrokenBarrierException.class) {
409		public void realRun() throws Exception {
410		start.await();
#	Line 409 | Line 420 | public class CyclicBarrierTest extends J
420		t1.start();
421		t2.start();
422		start.await();
423	<	while (barrier.getNumberWaiting() < 2) { Thread.yield(); }
423	>	awaitNumberWaiting(barrier, 2);
424		try {
425		barrier.await();
426		shouldThrow();
427		} catch (NullPointerException success) {}
428	<	t1.join();
429	<	t2.join();
428	>	awaitTermination(t1);
429	>	awaitTermination(t2);
430		assertTrue(barrier.isBroken());
431		assertEquals(0, barrier.getNumberWaiting());
432		barrier.reset();
#	Line 423 | Line 434 | public class CyclicBarrierTest extends J
434		assertEquals(0, barrier.getNumberWaiting());
435		}
436		}
437	+
438	+	/**
439	+	* There can be more threads calling await() than parties, as long as each
440	+	* task only calls await once and the task count is a multiple of parties.
441	+	*/
442	+	public void testMoreTasksThanParties() throws Exception {
443	+	final ThreadLocalRandom rnd = ThreadLocalRandom.current();
444	+	final int parties = rnd.nextInt(1, 5);
445	+	final int nTasks = rnd.nextInt(1, 5) * parties;
446	+	final AtomicInteger tripCount = new AtomicInteger(0);
447	+	final AtomicInteger awaitCount = new AtomicInteger(0);
448	+	final CyclicBarrier barrier =
449	+	new CyclicBarrier(parties, () -> tripCount.getAndIncrement());
450	+	final ExecutorService e = Executors.newFixedThreadPool(nTasks);
451	+	final Runnable awaiter = () -> {
452	+	try {
453	+	if (randomBoolean())
454	+	barrier.await();
455	+	else
456	+	barrier.await(LONG_DELAY_MS, MILLISECONDS);
457	+	awaitCount.getAndIncrement();
458	+	} catch (Throwable fail) { threadUnexpectedException(fail); }};
459	+	try (PoolCleaner cleaner = cleaner(e)) {
460	+	for (int i = nTasks; i--> 0; )
461	+	e.execute(awaiter);
462	+	}
463	+	assertEquals(nTasks / parties, tripCount.get());
464	+	assertEquals(nTasks, awaitCount.get());
465	+	assertEquals(0, barrier.getNumberWaiting());
466	+	}
467		}

Diff Legend

–	Removed lines
+	Added lines
<	Changed lines
>	Changed lines