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.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) {
        main(suite(), args);
    }
    public static Test suite() {
        return new TestSuite(ExecutorCompletionServiceTest.class);
    }

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

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

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

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

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

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

    /**
     * If poll returns non-null, the returned task is completed
     */
    public void testPoll1() throws Exception {
        final ExecutorService e = Executors.newCachedThreadPool();
        final ExecutorCompletionService ecs = new ExecutorCompletionService(e);
        try (PoolCleaner cleaner = cleaner(e)) {
            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());
        }
    }

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

    /**
     * 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); }
        }
        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> ecs =
            new ExecutorCompletionService<>(e);
        try (PoolCleaner cleaner = cleaner(e)) {
            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());
        }
    }

    /**
     * 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); }
        }
        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);
                }};
        final CompletionService<String> ecs =
            new ExecutorCompletionService<>(e);
        try (PoolCleaner cleaner = cleaner(e)) {
            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());
        }
    }

}
Revision:	1.23
Committed:	Tue May 3 23:06:12 2016 UTC (8 years ago) by jsr166
Branch:	MAIN
Changes since 1.22:	+5 -4 lines
Log Message:	use <> as in javadoc code sample
#	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	jsr166	1.23	import java.util.concurrent.CompletionService;
14	jsr166	1.18	import java.util.concurrent.ExecutorCompletionService;
15	jsr166	1.19	import java.util.concurrent.Executors;
16	jsr166	1.18	import java.util.concurrent.ExecutorService;
17			import java.util.concurrent.Future;
18			import java.util.concurrent.FutureTask;
19			import java.util.concurrent.RunnableFuture;
20			import java.util.concurrent.ThreadPoolExecutor;
21			import java.util.concurrent.TimeUnit;
22			import java.util.concurrent.atomic.AtomicBoolean;
23	jsr166	1.19
24			import junit.framework.Test;
25			import junit.framework.TestSuite;
26	dl	1.1
27	jsr166	1.8	public class ExecutorCompletionServiceTest extends JSR166TestCase {
28	dl	1.1	public static void main(String[] args) {
29	jsr166	1.21	main(suite(), args);
30	dl	1.1	}
31			public static Test suite() {
32			return new TestSuite(ExecutorCompletionServiceTest.class);
33			}
34
35			/**
36			* Creating a new ECS with null Executor throw NPE
37	jsr166	1.7	*/
38	dl	1.1	public void testConstructorNPE() {
39			try {
40	jsr166	1.20	new ExecutorCompletionService(null);
41	dl	1.1	shouldThrow();
42	jsr166	1.10	} catch (NullPointerException success) {}
43	dl	1.1	}
44
45			/**
46			* Creating a new ECS with null queue throw NPE
47	jsr166	1.7	*/
48	dl	1.1	public void testConstructorNPE2() {
49			try {
50			ExecutorService e = Executors.newCachedThreadPool();
51	jsr166	1.20	new ExecutorCompletionService(e, null);
52	dl	1.1	shouldThrow();
53	jsr166	1.10	} catch (NullPointerException success) {}
54	dl	1.1	}
55
56			/**
57			* Submitting a null callable throws NPE
58	jsr166	1.7	*/
59	dl	1.1	public void testSubmitNPE() {
60	jsr166	1.22	final ExecutorService e = Executors.newCachedThreadPool();
61			final ExecutorCompletionService ecs = new ExecutorCompletionService(e);
62			try (PoolCleaner cleaner = cleaner(e)) {
63	dl	1.1	Callable c = null;
64	jsr166	1.22	try {
65			ecs.submit(c);
66			shouldThrow();
67			} catch (NullPointerException success) {}
68	dl	1.1	}
69			}
70
71			/**
72			* Submitting a null runnable throws NPE
73	jsr166	1.7	*/
74	dl	1.1	public void testSubmitNPE2() {
75	jsr166	1.22	final ExecutorService e = Executors.newCachedThreadPool();
76			final ExecutorCompletionService ecs = new ExecutorCompletionService(e);
77			try (PoolCleaner cleaner = cleaner(e)) {
78	dl	1.1	Runnable r = null;
79	jsr166	1.22	try {
80			ecs.submit(r, Boolean.TRUE);
81			shouldThrow();
82			} catch (NullPointerException success) {}
83	dl	1.1	}
84			}
85
86			/**
87			* A taken submitted task is completed
88	jsr166	1.7	*/
89	jsr166	1.10	public void testTake() throws InterruptedException {
90	jsr166	1.22	final ExecutorService e = Executors.newCachedThreadPool();
91			final ExecutorCompletionService ecs = new ExecutorCompletionService(e);
92			try (PoolCleaner cleaner = cleaner(e)) {
93	dl	1.1	Callable c = new StringTask();
94			ecs.submit(c);
95			Future f = ecs.take();
96	dl	1.3	assertTrue(f.isDone());
97	dl	1.1	}
98			}
99
100			/**
101			* Take returns the same future object returned by submit
102	jsr166	1.7	*/
103	jsr166	1.10	public void testTake2() throws InterruptedException {
104	jsr166	1.22	final ExecutorService e = Executors.newCachedThreadPool();
105			final ExecutorCompletionService ecs = new ExecutorCompletionService(e);
106			try (PoolCleaner cleaner = cleaner(e)) {
107	dl	1.1	Callable c = new StringTask();
108			Future f1 = ecs.submit(c);
109			Future f2 = ecs.take();
110			assertSame(f1, f2);
111			}
112			}
113
114			/**
115			* If poll returns non-null, the returned task is completed
116	jsr166	1.7	*/
117	jsr166	1.16	public void testPoll1() throws Exception {
118	jsr166	1.22	final ExecutorService e = Executors.newCachedThreadPool();
119			final ExecutorCompletionService ecs = new ExecutorCompletionService(e);
120			try (PoolCleaner cleaner = cleaner(e)) {
121	dl	1.1	assertNull(ecs.poll());
122			Callable c = new StringTask();
123			ecs.submit(c);
124	jsr166	1.16
125			long startTime = System.nanoTime();
126			Future f;
127			while ((f = ecs.poll()) == null) {
128			if (millisElapsedSince(startTime) > LONG_DELAY_MS)
129			fail("timed out");
130			Thread.yield();
131	dl	1.1	}
132	jsr166	1.16	assertTrue(f.isDone());
133			assertSame(TEST_STRING, f.get());
134	dl	1.1	}
135			}
136
137			/**
138			* If timed poll returns non-null, the returned task is completed
139	jsr166	1.7	*/
140	jsr166	1.10	public void testPoll2() throws InterruptedException {
141	jsr166	1.22	final ExecutorService e = Executors.newCachedThreadPool();
142			final ExecutorCompletionService ecs = new ExecutorCompletionService(e);
143			try (PoolCleaner cleaner = cleaner(e)) {
144	dl	1.1	assertNull(ecs.poll());
145			Callable c = new StringTask();
146			ecs.submit(c);
147	jsr166	1.9	Future f = ecs.poll(SHORT_DELAY_MS, MILLISECONDS);
148	jsr166	1.7	if (f != null)
149	dl	1.3	assertTrue(f.isDone());
150	dl	1.1	}
151			}
152	jsr166	1.12
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	jsr166	1.22	final ExecutorService e =
164			new ThreadPoolExecutor(1, 1,
165			30L, TimeUnit.SECONDS,
166			new ArrayBlockingQueue<Runnable>(1)) {
167			protected <T> RunnableFuture<T> newTaskFor(Callable<T> c) {
168			return new MyCallableFuture<T>(c);
169			}};
170	jsr166	1.23	CompletionService<String> ecs =
171			new ExecutorCompletionService<>(e);
172	jsr166	1.22	try (PoolCleaner cleaner = cleaner(e)) {
173	jsr166	1.12	assertNull(ecs.poll());
174			Callable<String> c = new StringTask();
175			Future f1 = ecs.submit(c);
176			assertTrue("submit must return MyCallableFuture",
177			f1 instanceof MyCallableFuture);
178			Future f2 = ecs.take();
179			assertSame("submit and take must return same objects", f1, f2);
180			assertTrue("completed task must have set done", done.get());
181			}
182			}
183	dl	1.4
184	jsr166	1.12	/**
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	jsr166	1.22	final ExecutorService e =
195			new ThreadPoolExecutor(1, 1,
196			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	jsr166	1.23	final CompletionService<String> ecs =
202			new ExecutorCompletionService<>(e);
203	jsr166	1.22	try (PoolCleaner cleaner = cleaner(e)) {
204	jsr166	1.12	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			}
213			}
214	dl	1.4
215	dl	1.1	}