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