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

Diff Legend

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