test/tck/ExecutorCompletionServiceTest.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/publicdomain/zero/1.0/
 * Other contributors include Andrew Wright, Jeffrey Hayes,
 * Pat Fisher, Mike Judd.
 */

import static java.util.concurrent.TimeUnit.MILLISECONDS;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import junit.framework.Test;
import junit.framework.TestSuite;

public class ExecutorCompletionServiceTest extends JSR166TestCase {
    public static void main(String[] args) {
        main(suite(), args);
    }
    public static Test suite() {
        return new TestSuite(ExecutorCompletionServiceTest.class);
    }

    /**
     * new ExecutorCompletionService(null) throws NullPointerException
     */
    public void testConstructorNPE() {
        try {
            new ExecutorCompletionService<Item>(null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }

    /**
     * new ExecutorCompletionService(e, null) throws NullPointerException
     */
    public void testConstructorNPE2() {
        try {
            new ExecutorCompletionService<Item>(cachedThreadPool, null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }

    /**
     * ecs.submit(null) throws NullPointerException
     */
    public void testSubmitNullCallable() {
        CompletionService<Item> cs = new ExecutorCompletionService<>(cachedThreadPool);
        try {
            cs.submit((Callable<Item>) null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }

    /**
     * ecs.submit(null, val) throws NullPointerException
     */
    public void testSubmitNullRunnable() {
        CompletionService<Boolean> cs = new ExecutorCompletionService<>(cachedThreadPool);
        try {
            cs.submit((Runnable) null, Boolean.TRUE);
            shouldThrow();
        } catch (NullPointerException success) {}
    }

    /**
     * A taken submitted task is completed
     */
    public void testTake() throws Exception {
        CompletionService<String> cs = new ExecutorCompletionService<>(cachedThreadPool);
        cs.submit(new StringTask());
        Future<?> f = cs.take();
        assertTrue(f.isDone());
        assertSame(TEST_STRING, f.get());
    }

    /**
     * Take returns the same future object returned by submit
     */
    public void testTake2() throws InterruptedException {
        CompletionService<String> cs = new ExecutorCompletionService<>(cachedThreadPool);
        Future<?> f1 = cs.submit(new StringTask());
        Future<?> f2 = cs.take();
        assertSame(f1, f2);
    }

    /**
     * poll returns non-null when the returned task is completed
     */
    public void testPoll1() throws Exception {
        CompletionService<String> cs = new ExecutorCompletionService<>(cachedThreadPool);
        assertNull(cs.poll());
        cs.submit(new StringTask());

        long startTime = System.nanoTime();
        Future<?> f;
        while ((f = cs.poll()) == null) {
            if (millisElapsedSince(startTime) > LONG_DELAY_MS)
                fail("timed out");
            Thread.yield();
        }
        assertTrue(f.isDone());
        assertSame(TEST_STRING, f.get());
    }

    /**
     * timed poll returns non-null when the returned task is completed
     */
    public void testPoll2() throws Exception {
        CompletionService<String> cs = new ExecutorCompletionService<>(cachedThreadPool);
        assertNull(cs.poll());
        cs.submit(new StringTask());

        long startTime = System.nanoTime();
        Future<?> f;
        while ((f = cs.poll(timeoutMillis(), MILLISECONDS)) == null) {
            assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
            if (millisElapsedSince(startTime) > LONG_DELAY_MS)
                fail("timed out");
            Thread.yield();
        }
        assertTrue(f.isDone());
        assertSame(TEST_STRING, f.get());
    }

    /**
     * poll returns null before the returned task is completed
     */
    public void testPollReturnsNullBeforeCompletion() throws Exception {
        CompletionService<String> cs = new ExecutorCompletionService<>(cachedThreadPool);
        final CountDownLatch proceed = new CountDownLatch(1);
        cs.submit(new Callable<String>() { public String call() throws Exception {
            await(proceed);
            return TEST_STRING;
        }});
        assertNull(cs.poll());
        assertNull(cs.poll(0L, MILLISECONDS));
        assertNull(cs.poll(Long.MIN_VALUE, MILLISECONDS));
        long startTime = System.nanoTime();
        assertNull(cs.poll(timeoutMillis(), MILLISECONDS));
        assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
        proceed.countDown();
        assertSame(TEST_STRING, cs.take().get());
    }

    /**
     * successful and failed tasks are both returned
     */
    public void testTaskAssortment() throws Exception {
        CompletionService<String> cs = new ExecutorCompletionService<>(cachedThreadPool);
        ArithmeticException ex = new ArithmeticException();
        final int rounds = 2;
        for (int i = rounds; i--> 0; ) {
            cs.submit(new StringTask());
            cs.submit(callableThrowing(ex));
            cs.submit(runnableThrowing(ex), null);
        }
        int normalCompletions = 0;
        int exceptionalCompletions = 0;
        for (int i = 3 * rounds; i--> 0; ) {
            try {
                assertSame(TEST_STRING, cs.take().get());
                normalCompletions++;
            } catch (ExecutionException expected) {
                assertSame(ex, expected.getCause());
                exceptionalCompletions++;
            }
        }
        assertEquals(1 * rounds, normalCompletions);
        assertEquals(2 * rounds, exceptionalCompletions);
        assertNull(cs.poll());
    }

    /**
     * Submitting to underlying AES that overrides newTaskFor(Callable)
     * returns and eventually runs Future returned by newTaskFor.
     */
    public void testNewTaskForCallable() throws InterruptedException {
        final AtomicBoolean done = new AtomicBoolean(false);
        class MyCallableFuture<V> extends FutureTask<V> {
            MyCallableFuture(Callable<V> c) { super(c); }
            @Override protected void done() { done.set(true); }
        }
        final ExecutorService e =
            new ThreadPoolExecutor(1, 1,
                                   30L, TimeUnit.SECONDS,
                                   new ArrayBlockingQueue<Runnable>(1)) {
                protected <T> RunnableFuture<T> newTaskFor(Callable<T> c) {
                    return new MyCallableFuture<T>(c);
                }};
        CompletionService<String> cs = new ExecutorCompletionService<>(e);
        try (PoolCleaner cleaner = cleaner(e)) {
            assertNull(cs.poll());
            Callable<String> c = new StringTask();
            Future<?> f1 = cs.submit(c);
            assertTrue("submit must return MyCallableFuture",
                       f1 instanceof MyCallableFuture);
            Future<?> f2 = cs.take();
            assertSame("submit and take must return same objects", f1, f2);
            assertTrue("completed task must have set done", done.get());
        }
    }

    /**
     * Submitting to underlying AES that overrides newTaskFor(Runnable,T)
     * returns and eventually runs Future returned by newTaskFor.
     */
    public void testNewTaskForRunnable() throws InterruptedException {
        final AtomicBoolean done = new AtomicBoolean(false);
        class MyRunnableFuture<V> extends FutureTask<V> {
            MyRunnableFuture(Runnable t, V r) { super(t, r); }
            @Override protected void done() { done.set(true); }
        }
        final ExecutorService e =
            new ThreadPoolExecutor(1, 1,
                                   30L, TimeUnit.SECONDS,
                                   new ArrayBlockingQueue<Runnable>(1)) {
                protected <T> RunnableFuture<T> newTaskFor(Runnable t, T r) {
                    return new MyRunnableFuture<T>(t, r);
                }};
        CompletionService<String> cs = new ExecutorCompletionService<>(e);
        try (PoolCleaner cleaner = cleaner(e)) {
            assertNull(cs.poll());
            Runnable r = new NoOpRunnable();
            Future<?> f1 = cs.submit(r, null);
            assertTrue("submit must return MyRunnableFuture",
                       f1 instanceof MyRunnableFuture);
            Future<?> f2 = cs.take();
            assertSame("submit and take must return same objects", f1, f2);
            assertTrue("completed task must have set done", done.get());
        }
    }

}
Revision:	1.31
Committed:	Wed Jan 27 01:57:24 2021 UTC (3 years, 3 months ago) by jsr166
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.30:	+8 -8 lines
Log Message:	use diamond <> pervasively
#	Content
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	* 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 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	main(suite(), args);
31	}
32	public static Test suite() {
33	return new TestSuite(ExecutorCompletionServiceTest.class);
34	}
35
36	/**
37	* new ExecutorCompletionService(null) throws NullPointerException
38	*/
39	public void testConstructorNPE() {
40	try {
41	new ExecutorCompletionService<Item>(null);
42	shouldThrow();
43	} catch (NullPointerException success) {}
44	}
45
46	/**
47	* new ExecutorCompletionService(e, null) throws NullPointerException
48	*/
49	public void testConstructorNPE2() {
50	try {
51	new ExecutorCompletionService<Item>(cachedThreadPool, null);
52	shouldThrow();
53	} catch (NullPointerException success) {}
54	}
55
56	/**
57	* ecs.submit(null) throws NullPointerException
58	*/
59	public void testSubmitNullCallable() {
60	CompletionService<Item> cs = new ExecutorCompletionService<>(cachedThreadPool);
61	try {
62	cs.submit((Callable<Item>) null);
63	shouldThrow();
64	} catch (NullPointerException success) {}
65	}
66
67	/**
68	* ecs.submit(null, val) throws NullPointerException
69	*/
70	public void testSubmitNullRunnable() {
71	CompletionService<Boolean> cs = new ExecutorCompletionService<>(cachedThreadPool);
72	try {
73	cs.submit((Runnable) null, Boolean.TRUE);
74	shouldThrow();
75	} catch (NullPointerException success) {}
76	}
77
78	/**
79	* A taken submitted task is completed
80	*/
81	public void testTake() throws Exception {
82	CompletionService<String> 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() throws InterruptedException {
93	CompletionService<String> 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<String> 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<String> 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<String> cs = new ExecutorCompletionService<>(cachedThreadPool);
143	final CountDownLatch proceed = new CountDownLatch(1);
144	cs.submit(new Callable<String>() { 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<String> 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	* 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	* 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	}