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
#	User	Rev	Content
1	dl	1.1	/*
2	dl	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	jsr166	1.13	* http://creativecommons.org/publicdomain/zero/1.0/
5	jsr166	1.7	* Other contributors include Andrew Wright, Jeffrey Hayes,
6			* Pat Fisher, Mike Judd.
7	dl	1.1	*/
8
9	jsr166	1.19	import static java.util.concurrent.TimeUnit.MILLISECONDS;
10
11	jsr166	1.18	import java.util.concurrent.ArrayBlockingQueue;
12			import java.util.concurrent.Callable;
13	jsr166	1.23	import java.util.concurrent.CompletionService;
14	jsr166	1.24	import java.util.concurrent.CountDownLatch;
15			import java.util.concurrent.ExecutionException;
16	jsr166	1.18	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	jsr166	1.19
25			import junit.framework.Test;
26			import junit.framework.TestSuite;
27	dl	1.1
28	jsr166	1.8	public class ExecutorCompletionServiceTest extends JSR166TestCase {
29	dl	1.1	public static void main(String[] args) {
30	jsr166	1.21	main(suite(), args);
31	dl	1.1	}
32			public static Test suite() {
33			return new TestSuite(ExecutorCompletionServiceTest.class);
34			}
35
36			/**
37	jsr166	1.24	* new ExecutorCompletionService(null) throws NullPointerException
38	jsr166	1.7	*/
39	dl	1.1	public void testConstructorNPE() {
40			try {
41	dl	1.30	new ExecutorCompletionService<Item>(null);
42	dl	1.1	shouldThrow();
43	jsr166	1.10	} catch (NullPointerException success) {}
44	dl	1.1	}
45
46			/**
47	jsr166	1.24	* new ExecutorCompletionService(e, null) throws NullPointerException
48	jsr166	1.7	*/
49	dl	1.1	public void testConstructorNPE2() {
50			try {
51	dl	1.30	new ExecutorCompletionService<Item>(cachedThreadPool, null);
52	dl	1.1	shouldThrow();
53	jsr166	1.10	} catch (NullPointerException success) {}
54	dl	1.1	}
55
56			/**
57	jsr166	1.24	* ecs.submit(null) throws NullPointerException
58	jsr166	1.7	*/
59	jsr166	1.24	public void testSubmitNullCallable() {
60	jsr166	1.31	CompletionService<Item> cs = new ExecutorCompletionService<>(cachedThreadPool);
61	jsr166	1.24	try {
62	dl	1.30	cs.submit((Callable<Item>) null);
63	jsr166	1.24	shouldThrow();
64			} catch (NullPointerException success) {}
65	dl	1.1	}
66
67			/**
68	jsr166	1.24	* ecs.submit(null, val) throws NullPointerException
69	jsr166	1.7	*/
70	jsr166	1.24	public void testSubmitNullRunnable() {
71	jsr166	1.31	CompletionService<Boolean> cs = new ExecutorCompletionService<>(cachedThreadPool);
72	jsr166	1.24	try {
73	jsr166	1.25	cs.submit((Runnable) null, Boolean.TRUE);
74	jsr166	1.24	shouldThrow();
75			} catch (NullPointerException success) {}
76	dl	1.1	}
77
78			/**
79			* A taken submitted task is completed
80	jsr166	1.7	*/
81	jsr166	1.29	public void testTake() throws Exception {
82	jsr166	1.31	CompletionService<String> cs = new ExecutorCompletionService<>(cachedThreadPool);
83	jsr166	1.25	cs.submit(new StringTask());
84	dl	1.30	Future<?> f = cs.take();
85	jsr166	1.24	assertTrue(f.isDone());
86			assertSame(TEST_STRING, f.get());
87	dl	1.1	}
88
89			/**
90			* Take returns the same future object returned by submit
91	jsr166	1.7	*/
92	jsr166	1.10	public void testTake2() throws InterruptedException {
93	jsr166	1.31	CompletionService<String> cs = new ExecutorCompletionService<>(cachedThreadPool);
94	dl	1.30	Future<?> f1 = cs.submit(new StringTask());
95			Future<?> f2 = cs.take();
96	jsr166	1.24	assertSame(f1, f2);
97			}
98
99			/**
100			* poll returns non-null when the returned task is completed
101			*/
102	jsr166	1.29	public void testPoll1() throws Exception {
103	jsr166	1.31	CompletionService<String> cs = new ExecutorCompletionService<>(cachedThreadPool);
104	jsr166	1.25	assertNull(cs.poll());
105			cs.submit(new StringTask());
106	jsr166	1.24
107			long startTime = System.nanoTime();
108	dl	1.30	Future<?> f;
109	jsr166	1.25	while ((f = cs.poll()) == null) {
110	jsr166	1.24	if (millisElapsedSince(startTime) > LONG_DELAY_MS)
111			fail("timed out");
112			Thread.yield();
113	dl	1.1	}
114	jsr166	1.24	assertTrue(f.isDone());
115			assertSame(TEST_STRING, f.get());
116	dl	1.1	}
117
118			/**
119	jsr166	1.24	* timed poll returns non-null when the returned task is completed
120	jsr166	1.7	*/
121	jsr166	1.29	public void testPoll2() throws Exception {
122	jsr166	1.31	CompletionService<String> cs = new ExecutorCompletionService<>(cachedThreadPool);
123	jsr166	1.25	assertNull(cs.poll());
124			cs.submit(new StringTask());
125	jsr166	1.24
126			long startTime = System.nanoTime();
127	dl	1.30	Future<?> f;
128	jsr166	1.29	while ((f = cs.poll(timeoutMillis(), MILLISECONDS)) == null) {
129			assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
130	jsr166	1.24	if (millisElapsedSince(startTime) > LONG_DELAY_MS)
131			fail("timed out");
132			Thread.yield();
133	dl	1.1	}
134	jsr166	1.24	assertTrue(f.isDone());
135			assertSame(TEST_STRING, f.get());
136			}
137
138			/**
139			* poll returns null before the returned task is completed
140			*/
141	jsr166	1.29	public void testPollReturnsNullBeforeCompletion() throws Exception {
142	jsr166	1.31	CompletionService<String> cs = new ExecutorCompletionService<>(cachedThreadPool);
143	jsr166	1.26	final CountDownLatch proceed = new CountDownLatch(1);
144	dl	1.30	cs.submit(new Callable<String>() { public String call() throws Exception {
145	jsr166	1.28	await(proceed);
146	jsr166	1.24	return TEST_STRING;
147			}});
148	jsr166	1.25	assertNull(cs.poll());
149			assertNull(cs.poll(0L, MILLISECONDS));
150			assertNull(cs.poll(Long.MIN_VALUE, MILLISECONDS));
151	jsr166	1.24	long startTime = System.nanoTime();
152	jsr166	1.25	assertNull(cs.poll(timeoutMillis(), MILLISECONDS));
153	jsr166	1.24	assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
154			proceed.countDown();
155	jsr166	1.25	assertSame(TEST_STRING, cs.take().get());
156	dl	1.1	}
157
158			/**
159	jsr166	1.24	* successful and failed tasks are both returned
160	jsr166	1.7	*/
161	jsr166	1.29	public void testTaskAssortment() throws Exception {
162	jsr166	1.31	CompletionService<String> cs = new ExecutorCompletionService<>(cachedThreadPool);
163	jsr166	1.25	ArithmeticException ex = new ArithmeticException();
164	jsr166	1.29	final int rounds = 2;
165			for (int i = rounds; i--> 0; ) {
166	jsr166	1.25	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	jsr166	1.29	for (int i = 3 * rounds; i--> 0; ) {
173	jsr166	1.25	try {
174	jsr166	1.29	assertSame(TEST_STRING, cs.take().get());
175			normalCompletions++;
176			} catch (ExecutionException expected) {
177			assertSame(ex, expected.getCause());
178	jsr166	1.25	exceptionalCompletions++;
179	jsr166	1.24	}
180	dl	1.1	}
181	jsr166	1.29	assertEquals(1 * rounds, normalCompletions);
182			assertEquals(2 * rounds, exceptionalCompletions);
183	jsr166	1.25	assertNull(cs.poll());
184	dl	1.1	}
185	jsr166	1.12
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	jsr166	1.24	@Override protected void done() { done.set(true); }
195	jsr166	1.12	}
196	jsr166	1.22	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	jsr166	1.24	CompletionService<String> cs = new ExecutorCompletionService<>(e);
204	jsr166	1.22	try (PoolCleaner cleaner = cleaner(e)) {
205	jsr166	1.24	assertNull(cs.poll());
206	jsr166	1.12	Callable<String> c = new StringTask();
207	dl	1.30	Future<?> f1 = cs.submit(c);
208	jsr166	1.12	assertTrue("submit must return MyCallableFuture",
209			f1 instanceof MyCallableFuture);
210	dl	1.30	Future<?> f2 = cs.take();
211	jsr166	1.12	assertSame("submit and take must return same objects", f1, f2);
212			assertTrue("completed task must have set done", done.get());
213			}
214			}
215	dl	1.4
216	jsr166	1.12	/**
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	jsr166	1.24	@Override protected void done() { done.set(true); }
225	jsr166	1.12	}
226	jsr166	1.22	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	jsr166	1.24	CompletionService<String> cs = new ExecutorCompletionService<>(e);
234	jsr166	1.22	try (PoolCleaner cleaner = cleaner(e)) {
235	jsr166	1.24	assertNull(cs.poll());
236	jsr166	1.12	Runnable r = new NoOpRunnable();
237	dl	1.30	Future<?> f1 = cs.submit(r, null);
238	jsr166	1.12	assertTrue("submit must return MyRunnableFuture",
239			f1 instanceof MyRunnableFuture);
240	dl	1.30	Future<?> f2 = cs.take();
241	jsr166	1.12	assertSame("submit and take must return same objects", f1, f2);
242			assertTrue("completed task must have set done", done.get());
243			}
244			}
245	dl	1.4
246	dl	1.1	}