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