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 junit.framework.*;
import java.util.*;
import java.util.concurrent.*;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import java.util.concurrent.atomic.*;
import java.security.*;

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

    /**
     * Creating a new ECS with null Executor throw NPE
     */
    public void testConstructorNPE() {
        try {
            ExecutorCompletionService ecs = new ExecutorCompletionService(null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }

    /**
     * Creating a new ECS with null queue throw NPE
     */
    public void testConstructorNPE2() {
        try {
            ExecutorService e = Executors.newCachedThreadPool();
            ExecutorCompletionService ecs = new ExecutorCompletionService(e, null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }

    /**
     * Submitting a null callable throws NPE
     */
    public void testSubmitNPE() {
        ExecutorService e = Executors.newCachedThreadPool();
        ExecutorCompletionService ecs = new ExecutorCompletionService(e);
        try {
            Callable c = null;
            ecs.submit(c);
            shouldThrow();
        } catch (NullPointerException success) {
        } finally {
            joinPool(e);
        }
    }

    /**
     * Submitting a null runnable throws NPE
     */
    public void testSubmitNPE2() {
        ExecutorService e = Executors.newCachedThreadPool();
        ExecutorCompletionService ecs = new ExecutorCompletionService(e);
        try {
            Runnable r = null;
            ecs.submit(r, Boolean.TRUE);
            shouldThrow();
        } catch (NullPointerException success) {
        } finally {
            joinPool(e);
        }
    }

    /**
     * A taken submitted task is completed
     */
    public void testTake() throws InterruptedException {
        ExecutorService e = Executors.newCachedThreadPool();
        ExecutorCompletionService ecs = new ExecutorCompletionService(e);
        try {
            Callable c = new StringTask();
            ecs.submit(c);
            Future f = ecs.take();
            assertTrue(f.isDone());
        } finally {
            joinPool(e);
        }
    }

    /**
     * Take returns the same future object returned by submit
     */
    public void testTake2() throws InterruptedException {
        ExecutorService e = Executors.newCachedThreadPool();
        ExecutorCompletionService ecs = new ExecutorCompletionService(e);
        try {
            Callable c = new StringTask();
            Future f1 = ecs.submit(c);
            Future f2 = ecs.take();
            assertSame(f1, f2);
        } finally {
            joinPool(e);
        }
    }

    /**
     * If poll returns non-null, the returned task is completed
     */
    public void testPoll1() throws Exception {
        ExecutorService e = Executors.newCachedThreadPool();
        ExecutorCompletionService ecs = new ExecutorCompletionService(e);
        try {
            assertNull(ecs.poll());
            Callable c = new StringTask();
            ecs.submit(c);

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

    /**
     * If timed poll returns non-null, the returned task is completed
     */
    public void testPoll2() throws InterruptedException {
        ExecutorService e = Executors.newCachedThreadPool();
        ExecutorCompletionService ecs = new ExecutorCompletionService(e);
        try {
            assertNull(ecs.poll());
            Callable c = new StringTask();
            ecs.submit(c);
            Future f = ecs.poll(SHORT_DELAY_MS, MILLISECONDS);
            if (f != null)
                assertTrue(f.isDone());
        } finally {
            joinPool(e);
        }
    }

    /**
     * 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); }
            protected void done() { done.set(true); }
        }
        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);
            }};
        ExecutorCompletionService<String> ecs =
            new ExecutorCompletionService<String>(e);
        try {
            assertNull(ecs.poll());
            Callable<String> c = new StringTask();
            Future f1 = ecs.submit(c);
            assertTrue("submit must return MyCallableFuture",
                       f1 instanceof MyCallableFuture);
            Future f2 = ecs.take();
            assertSame("submit and take must return same objects", f1, f2);
            assertTrue("completed task must have set done", done.get());
        } finally {
            joinPool(e);
        }
    }

    /**
     * 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); }
            protected void done() { done.set(true); }
        }
        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);
            }};
        ExecutorCompletionService<String> ecs =
            new ExecutorCompletionService<String>(e);
        try {
            assertNull(ecs.poll());
            Runnable r = new NoOpRunnable();
            Future f1 = ecs.submit(r, null);
            assertTrue("submit must return MyRunnableFuture",
                       f1 instanceof MyRunnableFuture);
            Future f2 = ecs.take();
            assertSame("submit and take must return same objects", f1, f2);
            assertTrue("completed task must have set done", done.get());
        } finally {
            joinPool(e);
        }
    }

}
Revision:	1.17
Committed:	Sun May 29 06:54:23 2011 UTC (12 years, 11 months ago) by jsr166
Branch:	MAIN
Changes since 1.16:	+0 -1 lines
Log Message:	tidy imports
#	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 junit.framework.*;
10	import java.util.*;
11	import java.util.concurrent.*;
12	import static java.util.concurrent.TimeUnit.MILLISECONDS;
13	import java.util.concurrent.atomic.*;
14	import java.security.*;
15
16	public class ExecutorCompletionServiceTest extends JSR166TestCase {
17	public static void main(String[] args) {
18	junit.textui.TestRunner.run(suite());
19	}
20	public static Test suite() {
21	return new TestSuite(ExecutorCompletionServiceTest.class);
22	}
23
24	/**
25	* Creating a new ECS with null Executor throw NPE
26	*/
27	public void testConstructorNPE() {
28	try {
29	ExecutorCompletionService ecs = new ExecutorCompletionService(null);
30	shouldThrow();
31	} catch (NullPointerException success) {}
32	}
33
34	/**
35	* Creating a new ECS with null queue throw NPE
36	*/
37	public void testConstructorNPE2() {
38	try {
39	ExecutorService e = Executors.newCachedThreadPool();
40	ExecutorCompletionService ecs = new ExecutorCompletionService(e, null);
41	shouldThrow();
42	} catch (NullPointerException success) {}
43	}
44
45	/**
46	* Submitting a null callable throws NPE
47	*/
48	public void testSubmitNPE() {
49	ExecutorService e = Executors.newCachedThreadPool();
50	ExecutorCompletionService ecs = new ExecutorCompletionService(e);
51	try {
52	Callable c = null;
53	ecs.submit(c);
54	shouldThrow();
55	} catch (NullPointerException success) {
56	} finally {
57	joinPool(e);
58	}
59	}
60
61	/**
62	* Submitting a null runnable throws NPE
63	*/
64	public void testSubmitNPE2() {
65	ExecutorService e = Executors.newCachedThreadPool();
66	ExecutorCompletionService ecs = new ExecutorCompletionService(e);
67	try {
68	Runnable r = null;
69	ecs.submit(r, Boolean.TRUE);
70	shouldThrow();
71	} catch (NullPointerException success) {
72	} finally {
73	joinPool(e);
74	}
75	}
76
77	/**
78	* A taken submitted task is completed
79	*/
80	public void testTake() throws InterruptedException {
81	ExecutorService e = Executors.newCachedThreadPool();
82	ExecutorCompletionService ecs = new ExecutorCompletionService(e);
83	try {
84	Callable c = new StringTask();
85	ecs.submit(c);
86	Future f = ecs.take();
87	assertTrue(f.isDone());
88	} finally {
89	joinPool(e);
90	}
91	}
92
93	/**
94	* Take returns the same future object returned by submit
95	*/
96	public void testTake2() throws InterruptedException {
97	ExecutorService e = Executors.newCachedThreadPool();
98	ExecutorCompletionService ecs = new ExecutorCompletionService(e);
99	try {
100	Callable c = new StringTask();
101	Future f1 = ecs.submit(c);
102	Future f2 = ecs.take();
103	assertSame(f1, f2);
104	} finally {
105	joinPool(e);
106	}
107	}
108
109	/**
110	* If poll returns non-null, the returned task is completed
111	*/
112	public void testPoll1() throws Exception {
113	ExecutorService e = Executors.newCachedThreadPool();
114	ExecutorCompletionService ecs = new ExecutorCompletionService(e);
115	try {
116	assertNull(ecs.poll());
117	Callable c = new StringTask();
118	ecs.submit(c);
119
120	long startTime = System.nanoTime();
121	Future f;
122	while ((f = ecs.poll()) == null) {
123	if (millisElapsedSince(startTime) > LONG_DELAY_MS)
124	fail("timed out");
125	Thread.yield();
126	}
127	assertTrue(f.isDone());
128	assertSame(TEST_STRING, f.get());
129	} finally {
130	joinPool(e);
131	}
132	}
133
134	/**
135	* If timed poll returns non-null, the returned task is completed
136	*/
137	public void testPoll2() throws InterruptedException {
138	ExecutorService e = Executors.newCachedThreadPool();
139	ExecutorCompletionService ecs = new ExecutorCompletionService(e);
140	try {
141	assertNull(ecs.poll());
142	Callable c = new StringTask();
143	ecs.submit(c);
144	Future f = ecs.poll(SHORT_DELAY_MS, MILLISECONDS);
145	if (f != null)
146	assertTrue(f.isDone());
147	} finally {
148	joinPool(e);
149	}
150	}
151
152	/**
153	* Submitting to underlying AES that overrides newTaskFor(Callable)
154	* returns and eventually runs Future returned by newTaskFor.
155	*/
156	public void testNewTaskForCallable() throws InterruptedException {
157	final AtomicBoolean done = new AtomicBoolean(false);
158	class MyCallableFuture<V> extends FutureTask<V> {
159	MyCallableFuture(Callable<V> c) { super(c); }
160	protected void done() { done.set(true); }
161	}
162	ExecutorService e = new ThreadPoolExecutor(
163	1, 1, 30L, TimeUnit.SECONDS,
164	new ArrayBlockingQueue<Runnable>(1)) {
165	protected <T> RunnableFuture<T> newTaskFor(Callable<T> c) {
166	return new MyCallableFuture<T>(c);
167	}};
168	ExecutorCompletionService<String> ecs =
169	new ExecutorCompletionService<String>(e);
170	try {
171	assertNull(ecs.poll());
172	Callable<String> c = new StringTask();
173	Future f1 = ecs.submit(c);
174	assertTrue("submit must return MyCallableFuture",
175	f1 instanceof MyCallableFuture);
176	Future f2 = ecs.take();
177	assertSame("submit and take must return same objects", f1, f2);
178	assertTrue("completed task must have set done", done.get());
179	} finally {
180	joinPool(e);
181	}
182	}
183
184	/**
185	* Submitting to underlying AES that overrides newTaskFor(Runnable,T)
186	* returns and eventually runs Future returned by newTaskFor.
187	*/
188	public void testNewTaskForRunnable() throws InterruptedException {
189	final AtomicBoolean done = new AtomicBoolean(false);
190	class MyRunnableFuture<V> extends FutureTask<V> {
191	MyRunnableFuture(Runnable t, V r) { super(t, r); }
192	protected void done() { done.set(true); }
193	}
194	ExecutorService e = new ThreadPoolExecutor(
195	1, 1, 30L, TimeUnit.SECONDS,
196	new ArrayBlockingQueue<Runnable>(1)) {
197	protected <T> RunnableFuture<T> newTaskFor(Runnable t, T r) {
198	return new MyRunnableFuture<T>(t, r);
199	}};
200	ExecutorCompletionService<String> ecs =
201	new ExecutorCompletionService<String>(e);
202	try {
203	assertNull(ecs.poll());
204	Runnable r = new NoOpRunnable();
205	Future f1 = ecs.submit(r, null);
206	assertTrue("submit must return MyRunnableFuture",
207	f1 instanceof MyRunnableFuture);
208	Future f2 = ecs.take();
209	assertSame("submit and take must return same objects", f1, f2);
210	assertTrue("completed task must have set done", done.get());
211	} finally {
212	joinPool(e);
213	}
214	}
215
216	}