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