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.Executor;
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(null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }

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

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

    /**
     * ecs.submit(null, val) throws NullPointerException
     */
    public void testSubmitNullRunnable() {
        CompletionService 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 InterruptedException, ExecutionException {
        CompletionService 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 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 InterruptedException, ExecutionException {
        CompletionService 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 InterruptedException, ExecutionException {
        CompletionService cs = new ExecutorCompletionService(cachedThreadPool);
        assertNull(cs.poll());
        cs.submit(new StringTask());

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