ViewVC Help

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

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

Comparing jsr166/src/test/tck/ExecutorCompletionServiceTest.java (file contents):
Revision 1.9 by jsr166, Sat Nov 21 02:33:20 2009 UTC vs.
Revision 1.24 by jsr166, Sat May 21 22:30:16 2016 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	–
10	–	import junit.framework.*;
11	–	import java.util.*;
12	–	import java.util.concurrent.*;
9		import static java.util.concurrent.TimeUnit.MILLISECONDS;
10	<	import java.util.concurrent.atomic.*;
11	<	import java.math.BigInteger;
12	<	import java.security.*;
10	>
11	>	import java.util.concurrent.ArrayBlockingQueue;
12	>	import java.util.concurrent.Callable;
13	>	import java.util.concurrent.CompletionService;
14	>	import java.util.concurrent.CountDownLatch;
15	>	import java.util.concurrent.ExecutionException;
16	>	import java.util.concurrent.ExecutorCompletionService;
17	>	import java.util.concurrent.Executor;
18	>	import java.util.concurrent.Executors;
19	>	import java.util.concurrent.ExecutorService;
20	>	import java.util.concurrent.ForkJoinPool;
21	>	import java.util.concurrent.Future;
22	>	import java.util.concurrent.FutureTask;
23	>	import java.util.concurrent.RunnableFuture;
24	>	import java.util.concurrent.ThreadPoolExecutor;
25	>	import java.util.concurrent.TimeUnit;
26	>	import java.util.concurrent.atomic.AtomicBoolean;
27	>
28	>	import junit.framework.Test;
29	>	import junit.framework.TestSuite;
30
31		public class ExecutorCompletionServiceTest extends JSR166TestCase {
32		public static void main(String[] args) {
33	<	junit.textui.TestRunner.run (suite());
33	>	main(suite(), args);
34		}
35		public static Test suite() {
36		return new TestSuite(ExecutorCompletionServiceTest.class);
37		}
38
26	–
39		/**
40	<	* Creating a new ECS with null Executor throw NPE
40	>	* new ExecutorCompletionService(null) throws NullPointerException
41		*/
42		public void testConstructorNPE() {
43		try {
44	<	ExecutorCompletionService ecs = new ExecutorCompletionService(null);
44	>	new ExecutorCompletionService(null);
45		shouldThrow();
46	<	} catch (NullPointerException success) {
35	<	}
46	>	} catch (NullPointerException success) {}
47		}
48
49		/**
50	<	* Creating a new ECS with null queue throw NPE
50	>	* new ExecutorCompletionService(e, null) throws NullPointerException
51		*/
52		public void testConstructorNPE2() {
53	+	final Executor e = ForkJoinPool.commonPool();
54		try {
55	<	ExecutorService e = Executors.newCachedThreadPool();
44	<	ExecutorCompletionService ecs = new ExecutorCompletionService(e, null);
55	>	new ExecutorCompletionService(e, null);
56		shouldThrow();
57	<	} catch (NullPointerException success) {
47	<	}
57	>	} catch (NullPointerException success) {}
58		}
59
60		/**
61	<	* Submitting a null callable throws NPE
61	>	* ecs.submit(null) throws NullPointerException
62		*/
63	<	public void testSubmitNPE() {
64	<	ExecutorService e = Executors.newCachedThreadPool();
65	<	ExecutorCompletionService ecs = new ExecutorCompletionService(e);
63	>	public void testSubmitNullCallable() {
64	>	final ExecutorCompletionService ecs =
65	>	new ExecutorCompletionService(ForkJoinPool.commonPool());
66		try {
67	<	Callable c = null;
58	<	ecs.submit(c);
67	>	ecs.submit((Callable) null);
68		shouldThrow();
69	<	} catch (NullPointerException success) {
61	<	} finally {
62	<	joinPool(e);
63	<	}
69	>	} catch (NullPointerException success) {}
70		}
71
72		/**
73	<	* Submitting a null runnable throws NPE
73	>	* ecs.submit(null, val) throws NullPointerException
74		*/
75	<	public void testSubmitNPE2() {
76	<	ExecutorService e = Executors.newCachedThreadPool();
77	<	ExecutorCompletionService ecs = new ExecutorCompletionService(e);
75	>	public void testSubmitNullRunnable() {
76	>	final ExecutorCompletionService ecs =
77	>	new ExecutorCompletionService(ForkJoinPool.commonPool());
78		try {
79	<	Runnable r = null;
74	<	ecs.submit(r, Boolean.TRUE);
79	>	ecs.submit((Runnable) null, Boolean.TRUE);
80		shouldThrow();
81	<	} catch (NullPointerException success) {
77	<	} finally {
78	<	joinPool(e);
79	<	}
81	>	} catch (NullPointerException success) {}
82		}
83
84		/**
85		* A taken submitted task is completed
86		*/
87	<	public void testTake() {
88	<	ExecutorService e = Executors.newCachedThreadPool();
89	<	ExecutorCompletionService ecs = new ExecutorCompletionService(e);
90	<	try {
91	<	Callable c = new StringTask();
92	<	ecs.submit(c);
93	<	Future f = ecs.take();
94	<	assertTrue(f.isDone());
93	<	} catch (Exception ex) {
94	<	unexpectedException();
95	<	} finally {
96	<	joinPool(e);
97	<	}
87	>	public void testTake()
88	>	throws InterruptedException, ExecutionException {
89	>	final ExecutorCompletionService ecs =
90	>	new ExecutorCompletionService(ForkJoinPool.commonPool());
91	>	ecs.submit(new StringTask());
92	>	Future f = ecs.take();
93	>	assertTrue(f.isDone());
94	>	assertSame(TEST_STRING, f.get());
95		}
96
97		/**
98		* Take returns the same future object returned by submit
99		*/
100	<	public void testTake2() {
101	<	ExecutorService e = Executors.newCachedThreadPool();
102	<	ExecutorCompletionService ecs = new ExecutorCompletionService(e);
103	<	try {
104	<	Callable c = new StringTask();
105	<	Future f1 = ecs.submit(c);
106	<	Future f2 = ecs.take();
107	<	assertSame(f1, f2);
108	<	} catch (Exception ex) {
109	<	unexpectedException();
110	<	} finally {
111	<	joinPool(e);
100	>	public void testTake2() throws InterruptedException {
101	>	final ExecutorCompletionService ecs =
102	>	new ExecutorCompletionService(ForkJoinPool.commonPool());
103	>	Future f1 = ecs.submit(new StringTask());
104	>	Future f2 = ecs.take();
105	>	assertSame(f1, f2);
106	>	}
107	>
108	>	/**
109	>	* poll returns non-null when the returned task is completed
110	>	*/
111	>	public void testPoll1()
112	>	throws InterruptedException, ExecutionException {
113	>	final ExecutorCompletionService ecs =
114	>	new ExecutorCompletionService(ForkJoinPool.commonPool());
115	>	assertNull(ecs.poll());
116	>	ecs.submit(new StringTask());
117	>
118	>	long startTime = System.nanoTime();
119	>	Future f;
120	>	while ((f = ecs.poll()) == null) {
121	>	if (millisElapsedSince(startTime) > LONG_DELAY_MS)
122	>	fail("timed out");
123	>	Thread.yield();
124		}
125	+	assertTrue(f.isDone());
126	+	assertSame(TEST_STRING, f.get());
127		}
128
129		/**
130	<	* If poll returns non-null, the returned task is completed
130	>	* timed poll returns non-null when the returned task is completed
131		*/
132	<	public void testPoll1() {
133	<	ExecutorService e = Executors.newCachedThreadPool();
134	<	ExecutorCompletionService ecs = new ExecutorCompletionService(e);
135	<	try {
136	<	assertNull(ecs.poll());
137	<	Callable c = new StringTask();
138	<	ecs.submit(c);
139	<	Thread.sleep(SHORT_DELAY_MS);
140	<	for (;;) {
141	<	Future f = ecs.poll();
142	<	if (f != null) {
143	<	assertTrue(f.isDone());
144	<	break;
132	>	public void testPoll2()
133	>	throws InterruptedException, ExecutionException {
134	>	final ExecutorCompletionService ecs =
135	>	new ExecutorCompletionService(ForkJoinPool.commonPool());
136	>	assertNull(ecs.poll());
137	>	ecs.submit(new StringTask());
138	>
139	>	long startTime = System.nanoTime();
140	>	Future f;
141	>	while ((f = ecs.poll(SHORT_DELAY_MS, MILLISECONDS)) == null) {
142	>	if (millisElapsedSince(startTime) > LONG_DELAY_MS)
143	>	fail("timed out");
144	>	Thread.yield();
145	>	}
146	>	assertTrue(f.isDone());
147	>	assertSame(TEST_STRING, f.get());
148	>	}
149	>
150	>	/**
151	>	* poll returns null before the returned task is completed
152	>	*/
153	>	public void testPollReturnsNull()
154	>	throws InterruptedException, ExecutionException {
155	>	final ExecutorCompletionService ecs =
156	>	new ExecutorCompletionService(ForkJoinPool.commonPool());
157	>	final CountDownLatch proceed = new CountDownLatch(1);
158	>	ecs.submit(new Callable() { public String call() throws Exception {
159	>	proceed.await();
160	>	return TEST_STRING;
161	>	}});
162	>	assertNull(ecs.poll());
163	>	assertNull(ecs.poll(0L, MILLISECONDS));
164	>	assertNull(ecs.poll(Long.MIN_VALUE, MILLISECONDS));
165	>	long startTime = System.nanoTime();
166	>	assertNull(ecs.poll(timeoutMillis(), MILLISECONDS));
167	>	assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
168	>	proceed.countDown();
169	>	assertSame(TEST_STRING, ecs.take().get());
170	>	}
171	>
172	>	/**
173	>	* successful and failed tasks are both returned
174	>	*/
175	>	public void testTaskAssortment()
176	>	throws InterruptedException, ExecutionException {
177	>	final ExecutorService e = Executors.newCachedThreadPool();
178	>	final CompletionService cs = new ExecutorCompletionService(e);
179	>	final ArithmeticException ex = new ArithmeticException();
180	>	try (PoolCleaner cleaner = cleaner(e)) {
181	>	for (int i = 0; i < 2; i++) {
182	>	cs.submit(new StringTask());
183	>	cs.submit(callableThrowing(ex));
184	>	cs.submit(runnableThrowing(ex), null);
185	>	}
186	>	int normalCompletions = 0;
187	>	int exceptionalCompletions = 0;
188	>	for (int i = 0; i < 3 * 2; i++) {
189	>	try {
190	>	if (cs.take().get() == TEST_STRING)
191	>	normalCompletions++;
192	>	}
193	>	catch (ExecutionException expected) {
194	>	assertTrue(expected.getCause() instanceof ArithmeticException);
195	>	exceptionalCompletions++;
196		}
197		}
198	<	} catch (Exception ex) {
199	<	unexpectedException();
200	<	} finally {
139	<	joinPool(e);
198	>	assertEquals(2 * 1, normalCompletions);
199	>	assertEquals(2 * 2, exceptionalCompletions);
200	>	assertNull(cs.poll());
201		}
202		}
203
204		/**
205	<	* If timed poll returns non-null, the returned task is completed
205	>	* Submitting to underlying AES that overrides newTaskFor(Callable)
206	>	* returns and eventually runs Future returned by newTaskFor.
207		*/
208	<	public void testPoll2() {
209	<	ExecutorService e = Executors.newCachedThreadPool();
210	<	ExecutorCompletionService ecs = new ExecutorCompletionService(e);
211	<	try {
212	<	assertNull(ecs.poll());
213	<	Callable c = new StringTask();
214	<	ecs.submit(c);
215	<	Future f = ecs.poll(SHORT_DELAY_MS, MILLISECONDS);
216	<	if (f != null)
217	<	assertTrue(f.isDone());
218	<	} catch (Exception ex) {
219	<	unexpectedException();
220	<	} finally {
221	<	joinPool(e);
222	<	}
223	<	}
224	<	/**
225	<	* Submitting to underlying AES that overrides newTaskFor(Callable)
226	<	* returns and eventually runs Future returned by newTaskFor.
227	<	*/
228	<	public void testNewTaskForCallable() {
229	<	final AtomicBoolean done = new AtomicBoolean(false);
230	<	class MyCallableFuture<V> extends FutureTask<V> {
231	<	MyCallableFuture(Callable<V> c) { super(c); }
232	<	protected void done() { done.set(true); }
171	<	}
172	<	ExecutorService e = new ThreadPoolExecutor(
173	<	1, 1, 30L, TimeUnit.SECONDS,
174	<	new ArrayBlockingQueue<Runnable>(1)) {
175	<	protected <T> RunnableFuture<T> newTaskFor(Callable<T> c) {
176	<	return new MyCallableFuture<T>(c);
177	<	}
178	<	};
179	<	ExecutorCompletionService<String> ecs =
180	<	new ExecutorCompletionService<String>(e);
181	<	try {
182	<	assertNull(ecs.poll());
183	<	Callable<String> c = new StringTask();
184	<	Future f1 = ecs.submit(c);
185	<	assertTrue("submit must return MyCallableFuture",
186	<	f1 instanceof MyCallableFuture);
187	<	Future f2 = ecs.take();
188	<	assertSame("submit and take must return same objects", f1, f2);
189	<	assertTrue("completed task must have set done", done.get());
190	<	} catch (Exception ex) {
191	<	unexpectedException();
192	<	} finally {
193	<	joinPool(e);
194	<	}
195	<	}
196	<
197	<	/**
198	<	* Submitting to underlying AES that overrides newTaskFor(Runnable,T)
199	<	* returns and eventually runs Future returned by newTaskFor.
200	<	*/
201	<	public void testNewTaskForRunnable() {
202	<	final AtomicBoolean done = new AtomicBoolean(false);
203	<	class MyRunnableFuture<V> extends FutureTask<V> {
204	<	MyRunnableFuture(Runnable t, V r) { super(t, r); }
205	<	protected void done() { done.set(true); }
206	<	}
207	<	ExecutorService e = new ThreadPoolExecutor(
208	<	1, 1, 30L, TimeUnit.SECONDS,
209	<	new ArrayBlockingQueue<Runnable>(1)) {
210	<	protected <T> RunnableFuture<T> newTaskFor(Runnable t, T r) {
211	<	return new MyRunnableFuture<T>(t, r);
212	<	}
213	<	};
214	<	ExecutorCompletionService<String> ecs =
215	<	new ExecutorCompletionService<String>(e);
216	<	try {
217	<	assertNull(ecs.poll());
218	<	Runnable r = new NoOpRunnable();
219	<	Future f1 = ecs.submit(r, null);
220	<	assertTrue("submit must return MyRunnableFuture",
221	<	f1 instanceof MyRunnableFuture);
222	<	Future f2 = ecs.take();
223	<	assertSame("submit and take must return same objects", f1, f2);
224	<	assertTrue("completed task must have set done", done.get());
225	<	} catch (Exception ex) {
226	<	unexpectedException();
227	<	} finally {
228	<	joinPool(e);
229	<	}
230	<	}
231	<
208	>	public void testNewTaskForCallable() throws InterruptedException {
209	>	final AtomicBoolean done = new AtomicBoolean(false);
210	>	class MyCallableFuture<V> extends FutureTask<V> {
211	>	MyCallableFuture(Callable<V> c) { super(c); }
212	>	@Override protected void done() { done.set(true); }
213	>	}
214	>	final ExecutorService e =
215	>	new ThreadPoolExecutor(1, 1,
216	>	30L, TimeUnit.SECONDS,
217	>	new ArrayBlockingQueue<Runnable>(1)) {
218	>	protected <T> RunnableFuture<T> newTaskFor(Callable<T> c) {
219	>	return new MyCallableFuture<T>(c);
220	>	}};
221	>	CompletionService<String> cs = new ExecutorCompletionService<>(e);
222	>	try (PoolCleaner cleaner = cleaner(e)) {
223	>	assertNull(cs.poll());
224	>	Callable<String> c = new StringTask();
225	>	Future f1 = cs.submit(c);
226	>	assertTrue("submit must return MyCallableFuture",
227	>	f1 instanceof MyCallableFuture);
228	>	Future f2 = cs.take();
229	>	assertSame("submit and take must return same objects", f1, f2);
230	>	assertTrue("completed task must have set done", done.get());
231	>	}
232	>	}
233
234	+	/**
235	+	* Submitting to underlying AES that overrides newTaskFor(Runnable,T)
236	+	* returns and eventually runs Future returned by newTaskFor.
237	+	*/
238	+	public void testNewTaskForRunnable() throws InterruptedException {
239	+	final AtomicBoolean done = new AtomicBoolean(false);
240	+	class MyRunnableFuture<V> extends FutureTask<V> {
241	+	MyRunnableFuture(Runnable t, V r) { super(t, r); }
242	+	@Override protected void done() { done.set(true); }
243	+	}
244	+	final ExecutorService e =
245	+	new ThreadPoolExecutor(1, 1,
246	+	30L, TimeUnit.SECONDS,
247	+	new ArrayBlockingQueue<Runnable>(1)) {
248	+	protected <T> RunnableFuture<T> newTaskFor(Runnable t, T r) {
249	+	return new MyRunnableFuture<T>(t, r);
250	+	}};
251	+	CompletionService<String> cs = new ExecutorCompletionService<>(e);
252	+	try (PoolCleaner cleaner = cleaner(e)) {
253	+	assertNull(cs.poll());
254	+	Runnable r = new NoOpRunnable();
255	+	Future f1 = cs.submit(r, null);
256	+	assertTrue("submit must return MyRunnableFuture",
257	+	f1 instanceof MyRunnableFuture);
258	+	Future f2 = cs.take();
259	+	assertSame("submit and take must return same objects", f1, f2);
260	+	assertTrue("completed task must have set done", done.get());
261	+	}
262	+	}
263
264		}

Diff Legend

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