1 /*
2 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
3 *
4 * This code is free software; you can redistribute it and/or modify it
5 * under the terms of the GNU General Public License version 2 only, as
6 * published by the Free Software Foundation.
7 *
8 * This code is distributed in the hope that it will be useful, but WITHOUT
9 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
11 * version 2 for more details (a copy is included in the LICENSE file that
12 * accompanied this code).
13 *
14 * You should have received a copy of the GNU General Public License version
15 * 2 along with this work; if not, write to the Free Software Foundation,
16 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
17 *
18 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
19 * or visit www.oracle.com if you need additional information or have any
20 * questions.
21 */
22
23 /*
24 * This file is available under and governed by the GNU General Public
25 * License version 2 only, as published by the Free Software Foundation.
26 * However, the following notice accompanied the original version of this
27 * file:
28 *
29 * Written by Doug Lea with assistance from members of JCP JSR-166
30 * Expert Group and released to the public domain, as explained at
31 * http://creativecommons.org/publicdomain/zero/1.0/
32 * Other contributors include Andrew Wright, Jeffrey Hayes,
33 * Pat Fisher, Mike Judd.
34 */
35
36 import static java.util.concurrent.TimeUnit.MILLISECONDS;
37 import static java.util.concurrent.TimeUnit.SECONDS;
38
39 import java.util.ArrayList;
40 import java.util.Collection;
41 import java.util.Collections;
42 import java.util.List;
43 import java.util.concurrent.ArrayBlockingQueue;
44 import java.util.concurrent.BlockingQueue;
45 import java.util.concurrent.Callable;
46 import java.util.concurrent.CancellationException;
47 import java.util.concurrent.CountDownLatch;
48 import java.util.concurrent.ExecutionException;
49 import java.util.concurrent.ExecutorService;
50 import java.util.concurrent.Future;
51 import java.util.concurrent.FutureTask;
52 import java.util.concurrent.LinkedBlockingQueue;
53 import java.util.concurrent.RejectedExecutionHandler;
54 import java.util.concurrent.RunnableFuture;
55 import java.util.concurrent.SynchronousQueue;
56 import java.util.concurrent.ThreadFactory;
57 import java.util.concurrent.ThreadLocalRandom;
58 import java.util.concurrent.ThreadPoolExecutor;
59 import java.util.concurrent.TimeoutException;
60 import java.util.concurrent.TimeUnit;
61 import java.util.concurrent.atomic.AtomicInteger;
62 import java.util.concurrent.locks.Condition;
63 import java.util.concurrent.locks.ReentrantLock;
64
65 import junit.framework.Test;
66 import junit.framework.TestSuite;
67
68 public class ThreadPoolExecutorSubclassTest extends JSR166TestCase {
69 public static void main(String[] args) {
70 main(suite(), args);
71 }
72 public static Test suite() {
73 return new TestSuite(ThreadPoolExecutorSubclassTest.class);
74 }
75
76 static class CustomTask<V> implements RunnableFuture<V> {
77 final Callable<V> callable;
78 final ReentrantLock lock = new ReentrantLock();
79 final Condition cond = lock.newCondition();
80 boolean done;
81 boolean cancelled;
82 V result;
83 Thread thread;
84 Exception exception;
85 CustomTask(Callable<V> c) {
86 if (c == null) throw new NullPointerException();
87 callable = c;
88 }
89 CustomTask(final Runnable r, final V res) {
90 if (r == null) throw new NullPointerException();
91 callable = new Callable<V>() {
92 public V call() throws Exception { r.run(); return res; }};
93 }
94 public boolean isDone() {
95 lock.lock(); try { return done; } finally { lock.unlock() ; }
96 }
97 public boolean isCancelled() {
98 lock.lock(); try { return cancelled; } finally { lock.unlock() ; }
99 }
100 public boolean cancel(boolean mayInterrupt) {
101 lock.lock();
102 try {
103 if (!done) {
104 cancelled = true;
105 done = true;
106 if (mayInterrupt && thread != null)
107 thread.interrupt();
108 return true;
109 }
110 return false;
111 }
112 finally { lock.unlock() ; }
113 }
114 public void run() {
115 lock.lock();
116 try {
117 if (done)
118 return;
119 thread = Thread.currentThread();
120 }
121 finally { lock.unlock() ; }
122 V v = null;
123 Exception e = null;
124 try {
125 v = callable.call();
126 }
127 catch (Exception ex) {
128 e = ex;
129 }
130 lock.lock();
131 try {
132 if (!done) {
133 result = v;
134 exception = e;
135 done = true;
136 thread = null;
137 cond.signalAll();
138 }
139 }
140 finally { lock.unlock(); }
141 }
142 public V get() throws InterruptedException, ExecutionException {
143 lock.lock();
144 try {
145 while (!done)
146 cond.await();
147 if (cancelled)
148 throw new CancellationException();
149 if (exception != null)
150 throw new ExecutionException(exception);
151 return result;
152 }
153 finally { lock.unlock(); }
154 }
155 public V get(long timeout, TimeUnit unit)
156 throws InterruptedException, ExecutionException, TimeoutException {
157 long nanos = unit.toNanos(timeout);
158 lock.lock();
159 try {
160 while (!done) {
161 if (nanos <= 0L)
162 throw new TimeoutException();
163 nanos = cond.awaitNanos(nanos);
164 }
165 if (cancelled)
166 throw new CancellationException();
167 if (exception != null)
168 throw new ExecutionException(exception);
169 return result;
170 }
171 finally { lock.unlock(); }
172 }
173 }
174
175 static class CustomTPE extends ThreadPoolExecutor {
176 protected <V> RunnableFuture<V> newTaskFor(Callable<V> c) {
177 return new CustomTask<V>(c);
178 }
179 protected <V> RunnableFuture<V> newTaskFor(Runnable r, V v) {
180 return new CustomTask<V>(r, v);
181 }
182
183 CustomTPE(int corePoolSize,
184 int maximumPoolSize,
185 long keepAliveTime,
186 TimeUnit unit,
187 BlockingQueue<Runnable> workQueue) {
188 super(corePoolSize, maximumPoolSize, keepAliveTime, unit,
189 workQueue);
190 }
191 CustomTPE(int corePoolSize,
192 int maximumPoolSize,
193 long keepAliveTime,
194 TimeUnit unit,
195 BlockingQueue<Runnable> workQueue,
196 ThreadFactory threadFactory) {
197 super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
198 threadFactory);
199 }
200
201 CustomTPE(int corePoolSize,
202 int maximumPoolSize,
203 long keepAliveTime,
204 TimeUnit unit,
205 BlockingQueue<Runnable> workQueue,
206 RejectedExecutionHandler handler) {
207 super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
208 handler);
209 }
210 CustomTPE(int corePoolSize,
211 int maximumPoolSize,
212 long keepAliveTime,
213 TimeUnit unit,
214 BlockingQueue<Runnable> workQueue,
215 ThreadFactory threadFactory,
216 RejectedExecutionHandler handler) {
217 super(corePoolSize, maximumPoolSize, keepAliveTime, unit,
218 workQueue, threadFactory, handler);
219 }
220
221 final CountDownLatch beforeCalled = new CountDownLatch(1);
222 final CountDownLatch afterCalled = new CountDownLatch(1);
223 final CountDownLatch terminatedCalled = new CountDownLatch(1);
224
225 public CustomTPE() {
226 super(1, 1, LONG_DELAY_MS, MILLISECONDS, new SynchronousQueue<Runnable>());
227 }
228 protected void beforeExecute(Thread t, Runnable r) {
229 beforeCalled.countDown();
230 }
231 protected void afterExecute(Runnable r, Throwable t) {
232 afterCalled.countDown();
233 }
234 protected void terminated() {
235 terminatedCalled.countDown();
236 }
237
238 public boolean beforeCalled() {
239 return beforeCalled.getCount() == 0;
240 }
241 public boolean afterCalled() {
242 return afterCalled.getCount() == 0;
243 }
244 public boolean terminatedCalled() {
245 return terminatedCalled.getCount() == 0;
246 }
247 }
248
249 static class FailingThreadFactory implements ThreadFactory {
250 int calls = 0;
251 public Thread newThread(Runnable r) {
252 if (++calls > 1) return null;
253 return new Thread(r);
254 }
255 }
256
257 /**
258 * execute successfully executes a runnable
259 */
260 public void testExecute() throws InterruptedException {
261 final ThreadPoolExecutor p =
262 new CustomTPE(1, 1,
263 2 * LONG_DELAY_MS, MILLISECONDS,
264 new ArrayBlockingQueue<Runnable>(10));
265 try (PoolCleaner cleaner = cleaner(p)) {
266 final CountDownLatch done = new CountDownLatch(1);
267 final Runnable task = new CheckedRunnable() {
268 public void realRun() { done.countDown(); }};
269 p.execute(task);
270 await(done);
271 }
272 }
273
274 /**
275 * getActiveCount increases but doesn't overestimate, when a
276 * thread becomes active
277 */
278 public void testGetActiveCount() throws InterruptedException {
279 final CountDownLatch done = new CountDownLatch(1);
280 final ThreadPoolExecutor p =
281 new CustomTPE(2, 2,
282 LONG_DELAY_MS, MILLISECONDS,
283 new ArrayBlockingQueue<Runnable>(10));
284 try (PoolCleaner cleaner = cleaner(p, done)) {
285 final CountDownLatch threadStarted = new CountDownLatch(1);
286 assertEquals(0, p.getActiveCount());
287 p.execute(new CheckedRunnable() {
288 public void realRun() throws InterruptedException {
289 threadStarted.countDown();
290 assertEquals(1, p.getActiveCount());
291 await(done);
292 }});
293 await(threadStarted);
294 assertEquals(1, p.getActiveCount());
295 }
296 }
297
298 /**
299 * prestartCoreThread starts a thread if under corePoolSize, else doesn't
300 */
301 public void testPrestartCoreThread() {
302 final ThreadPoolExecutor p =
303 new CustomTPE(2, 6,
304 LONG_DELAY_MS, MILLISECONDS,
305 new ArrayBlockingQueue<Runnable>(10));
306 try (PoolCleaner cleaner = cleaner(p)) {
307 assertEquals(0, p.getPoolSize());
308 assertTrue(p.prestartCoreThread());
309 assertEquals(1, p.getPoolSize());
310 assertTrue(p.prestartCoreThread());
311 assertEquals(2, p.getPoolSize());
312 assertFalse(p.prestartCoreThread());
313 assertEquals(2, p.getPoolSize());
314 p.setCorePoolSize(4);
315 assertTrue(p.prestartCoreThread());
316 assertEquals(3, p.getPoolSize());
317 assertTrue(p.prestartCoreThread());
318 assertEquals(4, p.getPoolSize());
319 assertFalse(p.prestartCoreThread());
320 assertEquals(4, p.getPoolSize());
321 }
322 }
323
324 /**
325 * prestartAllCoreThreads starts all corePoolSize threads
326 */
327 public void testPrestartAllCoreThreads() {
328 final ThreadPoolExecutor p =
329 new CustomTPE(2, 6,
330 LONG_DELAY_MS, MILLISECONDS,
331 new ArrayBlockingQueue<Runnable>(10));
332 try (PoolCleaner cleaner = cleaner(p)) {
333 assertEquals(0, p.getPoolSize());
334 p.prestartAllCoreThreads();
335 assertEquals(2, p.getPoolSize());
336 p.prestartAllCoreThreads();
337 assertEquals(2, p.getPoolSize());
338 p.setCorePoolSize(4);
339 p.prestartAllCoreThreads();
340 assertEquals(4, p.getPoolSize());
341 p.prestartAllCoreThreads();
342 assertEquals(4, p.getPoolSize());
343 }
344 }
345
346 /**
347 * getCompletedTaskCount increases, but doesn't overestimate,
348 * when tasks complete
349 */
350 public void testGetCompletedTaskCount() throws InterruptedException {
351 final ThreadPoolExecutor p =
352 new CustomTPE(2, 2,
353 LONG_DELAY_MS, MILLISECONDS,
354 new ArrayBlockingQueue<Runnable>(10));
355 try (PoolCleaner cleaner = cleaner(p)) {
356 final CountDownLatch threadStarted = new CountDownLatch(1);
357 final CountDownLatch threadProceed = new CountDownLatch(1);
358 final CountDownLatch threadDone = new CountDownLatch(1);
359 assertEquals(0, p.getCompletedTaskCount());
360 p.execute(new CheckedRunnable() {
361 public void realRun() throws InterruptedException {
362 threadStarted.countDown();
363 assertEquals(0, p.getCompletedTaskCount());
364 await(threadProceed);
365 threadDone.countDown();
366 }});
367 await(threadStarted);
368 assertEquals(0, p.getCompletedTaskCount());
369 threadProceed.countDown();
370 await(threadDone);
371 long startTime = System.nanoTime();
372 while (p.getCompletedTaskCount() != 1) {
373 if (millisElapsedSince(startTime) > LONG_DELAY_MS)
374 fail("timed out");
375 Thread.yield();
376 }
377 }
378 }
379
380 /**
381 * getCorePoolSize returns size given in constructor if not otherwise set
382 */
383 public void testGetCorePoolSize() {
384 final ThreadPoolExecutor p =
385 new CustomTPE(1, 1,
386 LONG_DELAY_MS, MILLISECONDS,
387 new ArrayBlockingQueue<Runnable>(10));
388 try (PoolCleaner cleaner = cleaner(p)) {
389 assertEquals(1, p.getCorePoolSize());
390 }
391 }
392
393 /**
394 * getKeepAliveTime returns value given in constructor if not otherwise set
395 */
396 public void testGetKeepAliveTime() {
397 final ThreadPoolExecutor p =
398 new CustomTPE(2, 2,
399 1000, MILLISECONDS,
400 new ArrayBlockingQueue<Runnable>(10));
401 try (PoolCleaner cleaner = cleaner(p)) {
402 assertEquals(1, p.getKeepAliveTime(SECONDS));
403 }
404 }
405
406 /**
407 * getThreadFactory returns factory in constructor if not set
408 */
409 public void testGetThreadFactory() {
410 final ThreadFactory threadFactory = new SimpleThreadFactory();
411 final ThreadPoolExecutor p =
412 new CustomTPE(1, 2,
413 LONG_DELAY_MS, MILLISECONDS,
414 new ArrayBlockingQueue<Runnable>(10),
415 threadFactory,
416 new NoOpREHandler());
417 try (PoolCleaner cleaner = cleaner(p)) {
418 assertSame(threadFactory, p.getThreadFactory());
419 }
420 }
421
422 /**
423 * setThreadFactory sets the thread factory returned by getThreadFactory
424 */
425 public void testSetThreadFactory() {
426 final ThreadPoolExecutor p =
427 new CustomTPE(1, 2,
428 LONG_DELAY_MS, MILLISECONDS,
429 new ArrayBlockingQueue<Runnable>(10));
430 try (PoolCleaner cleaner = cleaner(p)) {
431 ThreadFactory threadFactory = new SimpleThreadFactory();
432 p.setThreadFactory(threadFactory);
433 assertSame(threadFactory, p.getThreadFactory());
434 }
435 }
436
437 /**
438 * setThreadFactory(null) throws NPE
439 */
440 public void testSetThreadFactoryNull() {
441 final ThreadPoolExecutor p =
442 new CustomTPE(1, 2,
443 LONG_DELAY_MS, MILLISECONDS,
444 new ArrayBlockingQueue<Runnable>(10));
445 try (PoolCleaner cleaner = cleaner(p)) {
446 try {
447 p.setThreadFactory(null);
448 shouldThrow();
449 } catch (NullPointerException success) {}
450 }
451 }
452
453 /**
454 * getRejectedExecutionHandler returns handler in constructor if not set
455 */
456 public void testGetRejectedExecutionHandler() {
457 final RejectedExecutionHandler handler = new NoOpREHandler();
458 final ThreadPoolExecutor p =
459 new CustomTPE(1, 2,
460 LONG_DELAY_MS, MILLISECONDS,
461 new ArrayBlockingQueue<Runnable>(10),
462 handler);
463 try (PoolCleaner cleaner = cleaner(p)) {
464 assertSame(handler, p.getRejectedExecutionHandler());
465 }
466 }
467
468 /**
469 * setRejectedExecutionHandler sets the handler returned by
470 * getRejectedExecutionHandler
471 */
472 public void testSetRejectedExecutionHandler() {
473 final ThreadPoolExecutor p =
474 new CustomTPE(1, 2,
475 LONG_DELAY_MS, MILLISECONDS,
476 new ArrayBlockingQueue<Runnable>(10));
477 try (PoolCleaner cleaner = cleaner(p)) {
478 RejectedExecutionHandler handler = new NoOpREHandler();
479 p.setRejectedExecutionHandler(handler);
480 assertSame(handler, p.getRejectedExecutionHandler());
481 }
482 }
483
484 /**
485 * setRejectedExecutionHandler(null) throws NPE
486 */
487 public void testSetRejectedExecutionHandlerNull() {
488 final ThreadPoolExecutor p =
489 new CustomTPE(1, 2,
490 LONG_DELAY_MS, MILLISECONDS,
491 new ArrayBlockingQueue<Runnable>(10));
492 try (PoolCleaner cleaner = cleaner(p)) {
493 try {
494 p.setRejectedExecutionHandler(null);
495 shouldThrow();
496 } catch (NullPointerException success) {}
497 }
498 }
499
500 /**
501 * getLargestPoolSize increases, but doesn't overestimate, when
502 * multiple threads active
503 */
504 public void testGetLargestPoolSize() throws InterruptedException {
505 final int THREADS = 3;
506 final CountDownLatch done = new CountDownLatch(1);
507 final ThreadPoolExecutor p =
508 new CustomTPE(THREADS, THREADS,
509 LONG_DELAY_MS, MILLISECONDS,
510 new ArrayBlockingQueue<Runnable>(10));
511 try (PoolCleaner cleaner = cleaner(p, done)) {
512 assertEquals(0, p.getLargestPoolSize());
513 final CountDownLatch threadsStarted = new CountDownLatch(THREADS);
514 for (int i = 0; i < THREADS; i++)
515 p.execute(new CheckedRunnable() {
516 public void realRun() throws InterruptedException {
517 threadsStarted.countDown();
518 await(done);
519 assertEquals(THREADS, p.getLargestPoolSize());
520 }});
521 await(threadsStarted);
522 assertEquals(THREADS, p.getLargestPoolSize());
523 }
524 assertEquals(THREADS, p.getLargestPoolSize());
525 }
526
527 /**
528 * getMaximumPoolSize returns value given in constructor if not
529 * otherwise set
530 */
531 public void testGetMaximumPoolSize() {
532 final ThreadPoolExecutor p =
533 new CustomTPE(2, 3,
534 LONG_DELAY_MS, MILLISECONDS,
535 new ArrayBlockingQueue<Runnable>(10));
536 try (PoolCleaner cleaner = cleaner(p)) {
537 assertEquals(3, p.getMaximumPoolSize());
538 p.setMaximumPoolSize(5);
539 assertEquals(5, p.getMaximumPoolSize());
540 p.setMaximumPoolSize(4);
541 assertEquals(4, p.getMaximumPoolSize());
542 }
543 }
544
545 /**
546 * getPoolSize increases, but doesn't overestimate, when threads
547 * become active
548 */
549 public void testGetPoolSize() throws InterruptedException {
550 final CountDownLatch done = new CountDownLatch(1);
551 final ThreadPoolExecutor p =
552 new CustomTPE(1, 1,
553 LONG_DELAY_MS, MILLISECONDS,
554 new ArrayBlockingQueue<Runnable>(10));
555 try (PoolCleaner cleaner = cleaner(p, done)) {
556 assertEquals(0, p.getPoolSize());
557 final CountDownLatch threadStarted = new CountDownLatch(1);
558 p.execute(new CheckedRunnable() {
559 public void realRun() throws InterruptedException {
560 threadStarted.countDown();
561 assertEquals(1, p.getPoolSize());
562 await(done);
563 }});
564 await(threadStarted);
565 assertEquals(1, p.getPoolSize());
566 }
567 }
568
569 /**
570 * getTaskCount increases, but doesn't overestimate, when tasks submitted
571 */
572 public void testGetTaskCount() throws InterruptedException {
573 final int TASKS = 3;
574 final CountDownLatch done = new CountDownLatch(1);
575 final ThreadPoolExecutor p =
576 new CustomTPE(1, 1,
577 LONG_DELAY_MS, MILLISECONDS,
578 new ArrayBlockingQueue<Runnable>(10));
579 try (PoolCleaner cleaner = cleaner(p, done)) {
580 final CountDownLatch threadStarted = new CountDownLatch(1);
581 assertEquals(0, p.getTaskCount());
582 assertEquals(0, p.getCompletedTaskCount());
583 p.execute(new CheckedRunnable() {
584 public void realRun() throws InterruptedException {
585 threadStarted.countDown();
586 await(done);
587 }});
588 await(threadStarted);
589 assertEquals(1, p.getTaskCount());
590 assertEquals(0, p.getCompletedTaskCount());
591 for (int i = 0; i < TASKS; i++) {
592 assertEquals(1 + i, p.getTaskCount());
593 p.execute(new CheckedRunnable() {
594 public void realRun() throws InterruptedException {
595 threadStarted.countDown();
596 assertEquals(1 + TASKS, p.getTaskCount());
597 await(done);
598 }});
599 }
600 assertEquals(1 + TASKS, p.getTaskCount());
601 assertEquals(0, p.getCompletedTaskCount());
602 }
603 assertEquals(1 + TASKS, p.getTaskCount());
604 assertEquals(1 + TASKS, p.getCompletedTaskCount());
605 }
606
607 /**
608 * isShutdown is false before shutdown, true after
609 */
610 public void testIsShutdown() {
611 final ThreadPoolExecutor p =
612 new CustomTPE(1, 1,
613 LONG_DELAY_MS, MILLISECONDS,
614 new ArrayBlockingQueue<Runnable>(10));
615 try (PoolCleaner cleaner = cleaner(p)) {
616 assertFalse(p.isShutdown());
617 try { p.shutdown(); } catch (SecurityException ok) { return; }
618 assertTrue(p.isShutdown());
619 }
620 }
621
622 /**
623 * isTerminated is false before termination, true after
624 */
625 public void testIsTerminated() throws InterruptedException {
626 final ThreadPoolExecutor p =
627 new CustomTPE(1, 1,
628 LONG_DELAY_MS, MILLISECONDS,
629 new ArrayBlockingQueue<Runnable>(10));
630 try (PoolCleaner cleaner = cleaner(p)) {
631 final CountDownLatch threadStarted = new CountDownLatch(1);
632 final CountDownLatch done = new CountDownLatch(1);
633 assertFalse(p.isTerminating());
634 p.execute(new CheckedRunnable() {
635 public void realRun() throws InterruptedException {
636 assertFalse(p.isTerminating());
637 threadStarted.countDown();
638 await(done);
639 }});
640 await(threadStarted);
641 assertFalse(p.isTerminating());
642 done.countDown();
643 try { p.shutdown(); } catch (SecurityException ok) { return; }
644 assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS));
645 assertTrue(p.isTerminated());
646 assertFalse(p.isTerminating());
647 }
648 }
649
650 /**
651 * isTerminating is not true when running or when terminated
652 */
653 public void testIsTerminating() throws InterruptedException {
654 final ThreadPoolExecutor p =
655 new CustomTPE(1, 1,
656 LONG_DELAY_MS, MILLISECONDS,
657 new ArrayBlockingQueue<Runnable>(10));
658 try (PoolCleaner cleaner = cleaner(p)) {
659 final CountDownLatch threadStarted = new CountDownLatch(1);
660 final CountDownLatch done = new CountDownLatch(1);
661 assertFalse(p.isTerminating());
662 p.execute(new CheckedRunnable() {
663 public void realRun() throws InterruptedException {
664 assertFalse(p.isTerminating());
665 threadStarted.countDown();
666 await(done);
667 }});
668 await(threadStarted);
669 assertFalse(p.isTerminating());
670 done.countDown();
671 try { p.shutdown(); } catch (SecurityException ok) { return; }
672 assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS));
673 assertTrue(p.isTerminated());
674 assertFalse(p.isTerminating());
675 }
676 }
677
678 /**
679 * getQueue returns the work queue, which contains queued tasks
680 */
681 public void testGetQueue() throws InterruptedException {
682 final CountDownLatch done = new CountDownLatch(1);
683 final BlockingQueue<Runnable> q = new ArrayBlockingQueue<>(10);
684 final ThreadPoolExecutor p =
685 new CustomTPE(1, 1,
686 LONG_DELAY_MS, MILLISECONDS,
687 q);
688 try (PoolCleaner cleaner = cleaner(p, done)) {
689 final CountDownLatch threadStarted = new CountDownLatch(1);
690 FutureTask[] tasks = new FutureTask[5];
691 for (int i = 0; i < tasks.length; i++) {
692 Callable<Boolean> task = new CheckedCallable<Boolean>() {
693 public Boolean realCall() throws InterruptedException {
694 threadStarted.countDown();
695 assertSame(q, p.getQueue());
696 await(done);
697 return Boolean.TRUE;
698 }};
699 tasks[i] = new FutureTask(task);
700 p.execute(tasks[i]);
701 }
702 await(threadStarted);
703 assertSame(q, p.getQueue());
704 assertFalse(q.contains(tasks[0]));
705 assertTrue(q.contains(tasks[tasks.length - 1]));
706 assertEquals(tasks.length - 1, q.size());
707 }
708 }
709
710 /**
711 * remove(task) removes queued task, and fails to remove active task
712 */
713 public void testRemove() throws InterruptedException {
714 final CountDownLatch done = new CountDownLatch(1);
715 BlockingQueue<Runnable> q = new ArrayBlockingQueue<>(10);
716 final ThreadPoolExecutor p =
717 new CustomTPE(1, 1,
718 LONG_DELAY_MS, MILLISECONDS,
719 q);
720 try (PoolCleaner cleaner = cleaner(p, done)) {
721 Runnable[] tasks = new Runnable[6];
722 final CountDownLatch threadStarted = new CountDownLatch(1);
723 for (int i = 0; i < tasks.length; i++) {
724 tasks[i] = new CheckedRunnable() {
725 public void realRun() throws InterruptedException {
726 threadStarted.countDown();
727 await(done);
728 }};
729 p.execute(tasks[i]);
730 }
731 await(threadStarted);
732 assertFalse(p.remove(tasks[0]));
733 assertTrue(q.contains(tasks[4]));
734 assertTrue(q.contains(tasks[3]));
735 assertTrue(p.remove(tasks[4]));
736 assertFalse(p.remove(tasks[4]));
737 assertFalse(q.contains(tasks[4]));
738 assertTrue(q.contains(tasks[3]));
739 assertTrue(p.remove(tasks[3]));
740 assertFalse(q.contains(tasks[3]));
741 }
742 }
743
744 /**
745 * purge removes cancelled tasks from the queue
746 */
747 public void testPurge() throws InterruptedException {
748 final CountDownLatch threadStarted = new CountDownLatch(1);
749 final CountDownLatch done = new CountDownLatch(1);
750 final BlockingQueue<Runnable> q = new ArrayBlockingQueue<>(10);
751 final ThreadPoolExecutor p =
752 new CustomTPE(1, 1,
753 LONG_DELAY_MS, MILLISECONDS,
754 q);
755 try (PoolCleaner cleaner = cleaner(p, done)) {
756 FutureTask[] tasks = new FutureTask[5];
757 for (int i = 0; i < tasks.length; i++) {
758 Callable<Boolean> task = new CheckedCallable<Boolean>() {
759 public Boolean realCall() throws InterruptedException {
760 threadStarted.countDown();
761 await(done);
762 return Boolean.TRUE;
763 }};
764 tasks[i] = new FutureTask(task);
765 p.execute(tasks[i]);
766 }
767 await(threadStarted);
768 assertEquals(tasks.length, p.getTaskCount());
769 assertEquals(tasks.length - 1, q.size());
770 assertEquals(1L, p.getActiveCount());
771 assertEquals(0L, p.getCompletedTaskCount());
772 tasks[4].cancel(true);
773 tasks[3].cancel(false);
774 p.purge();
775 assertEquals(tasks.length - 3, q.size());
776 assertEquals(tasks.length - 2, p.getTaskCount());
777 p.purge(); // Nothing to do
778 assertEquals(tasks.length - 3, q.size());
779 assertEquals(tasks.length - 2, p.getTaskCount());
780 }
781 }
782
783 /**
784 * shutdownNow returns a list containing tasks that were not run,
785 * and those tasks are drained from the queue
786 */
787 public void testShutdownNow() throws InterruptedException {
788 final int poolSize = 2;
789 final int count = 5;
790 final AtomicInteger ran = new AtomicInteger(0);
791 final ThreadPoolExecutor p =
792 new CustomTPE(poolSize, poolSize,
793 LONG_DELAY_MS, MILLISECONDS,
794 new ArrayBlockingQueue<Runnable>(10));
795 final CountDownLatch threadsStarted = new CountDownLatch(poolSize);
796 Runnable waiter = new CheckedRunnable() { public void realRun() {
797 threadsStarted.countDown();
798 try {
799 MILLISECONDS.sleep(LONGER_DELAY_MS);
800 } catch (InterruptedException success) {}
801 ran.getAndIncrement();
802 }};
803 for (int i = 0; i < count; i++)
804 p.execute(waiter);
805 await(threadsStarted);
806 assertEquals(poolSize, p.getActiveCount());
807 assertEquals(0, p.getCompletedTaskCount());
808 final List<Runnable> queuedTasks;
809 try {
810 queuedTasks = p.shutdownNow();
811 } catch (SecurityException ok) {
812 return; // Allowed in case test doesn't have privs
813 }
814 assertTrue(p.isShutdown());
815 assertTrue(p.getQueue().isEmpty());
816 assertEquals(count - poolSize, queuedTasks.size());
817 assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS));
818 assertTrue(p.isTerminated());
819 assertEquals(poolSize, ran.get());
820 assertEquals(poolSize, p.getCompletedTaskCount());
821 }
822
823 // Exception Tests
824
825 /**
826 * Constructor throws if corePoolSize argument is less than zero
827 */
828 public void testConstructor1() {
829 try {
830 new CustomTPE(-1, 1, 1L, SECONDS,
831 new ArrayBlockingQueue<Runnable>(10));
832 shouldThrow();
833 } catch (IllegalArgumentException success) {}
834 }
835
836 /**
837 * Constructor throws if maximumPoolSize is less than zero
838 */
839 public void testConstructor2() {
840 try {
841 new CustomTPE(1, -1, 1L, SECONDS,
842 new ArrayBlockingQueue<Runnable>(10));
843 shouldThrow();
844 } catch (IllegalArgumentException success) {}
845 }
846
847 /**
848 * Constructor throws if maximumPoolSize is equal to zero
849 */
850 public void testConstructor3() {
851 try {
852 new CustomTPE(1, 0, 1L, SECONDS,
853 new ArrayBlockingQueue<Runnable>(10));
854 shouldThrow();
855 } catch (IllegalArgumentException success) {}
856 }
857
858 /**
859 * Constructor throws if keepAliveTime is less than zero
860 */
861 public void testConstructor4() {
862 try {
863 new CustomTPE(1, 2, -1L, SECONDS,
864 new ArrayBlockingQueue<Runnable>(10));
865 shouldThrow();
866 } catch (IllegalArgumentException success) {}
867 }
868
869 /**
870 * Constructor throws if corePoolSize is greater than the maximumPoolSize
871 */
872 public void testConstructor5() {
873 try {
874 new CustomTPE(2, 1, 1L, SECONDS,
875 new ArrayBlockingQueue<Runnable>(10));
876 shouldThrow();
877 } catch (IllegalArgumentException success) {}
878 }
879
880 /**
881 * Constructor throws if workQueue is set to null
882 */
883 public void testConstructorNullPointerException() {
884 try {
885 new CustomTPE(1, 2, 1L, SECONDS, null);
886 shouldThrow();
887 } catch (NullPointerException success) {}
888 }
889
890 /**
891 * Constructor throws if corePoolSize argument is less than zero
892 */
893 public void testConstructor6() {
894 try {
895 new CustomTPE(-1, 1, 1L, SECONDS,
896 new ArrayBlockingQueue<Runnable>(10),
897 new SimpleThreadFactory());
898 shouldThrow();
899 } catch (IllegalArgumentException success) {}
900 }
901
902 /**
903 * Constructor throws if maximumPoolSize is less than zero
904 */
905 public void testConstructor7() {
906 try {
907 new CustomTPE(1,-1, 1L, SECONDS,
908 new ArrayBlockingQueue<Runnable>(10),
909 new SimpleThreadFactory());
910 shouldThrow();
911 } catch (IllegalArgumentException success) {}
912 }
913
914 /**
915 * Constructor throws if maximumPoolSize is equal to zero
916 */
917 public void testConstructor8() {
918 try {
919 new CustomTPE(1, 0, 1L, SECONDS,
920 new ArrayBlockingQueue<Runnable>(10),
921 new SimpleThreadFactory());
922 shouldThrow();
923 } catch (IllegalArgumentException success) {}
924 }
925
926 /**
927 * Constructor throws if keepAliveTime is less than zero
928 */
929 public void testConstructor9() {
930 try {
931 new CustomTPE(1, 2, -1L, SECONDS,
932 new ArrayBlockingQueue<Runnable>(10),
933 new SimpleThreadFactory());
934 shouldThrow();
935 } catch (IllegalArgumentException success) {}
936 }
937
938 /**
939 * Constructor throws if corePoolSize is greater than the maximumPoolSize
940 */
941 public void testConstructor10() {
942 try {
943 new CustomTPE(2, 1, 1L, SECONDS,
944 new ArrayBlockingQueue<Runnable>(10),
945 new SimpleThreadFactory());
946 shouldThrow();
947 } catch (IllegalArgumentException success) {}
948 }
949
950 /**
951 * Constructor throws if workQueue is set to null
952 */
953 public void testConstructorNullPointerException2() {
954 try {
955 new CustomTPE(1, 2, 1L, SECONDS, null, new SimpleThreadFactory());
956 shouldThrow();
957 } catch (NullPointerException success) {}
958 }
959
960 /**
961 * Constructor throws if threadFactory is set to null
962 */
963 public void testConstructorNullPointerException3() {
964 try {
965 new CustomTPE(1, 2, 1L, SECONDS,
966 new ArrayBlockingQueue<Runnable>(10),
967 (ThreadFactory) null);
968 shouldThrow();
969 } catch (NullPointerException success) {}
970 }
971
972 /**
973 * Constructor throws if corePoolSize argument is less than zero
974 */
975 public void testConstructor11() {
976 try {
977 new CustomTPE(-1, 1, 1L, SECONDS,
978 new ArrayBlockingQueue<Runnable>(10),
979 new NoOpREHandler());
980 shouldThrow();
981 } catch (IllegalArgumentException success) {}
982 }
983
984 /**
985 * Constructor throws if maximumPoolSize is less than zero
986 */
987 public void testConstructor12() {
988 try {
989 new CustomTPE(1, -1, 1L, SECONDS,
990 new ArrayBlockingQueue<Runnable>(10),
991 new NoOpREHandler());
992 shouldThrow();
993 } catch (IllegalArgumentException success) {}
994 }
995
996 /**
997 * Constructor throws if maximumPoolSize is equal to zero
998 */
999 public void testConstructor13() {
1000 try {
1001 new CustomTPE(1, 0, 1L, SECONDS,
1002 new ArrayBlockingQueue<Runnable>(10),
1003 new NoOpREHandler());
1004 shouldThrow();
1005 } catch (IllegalArgumentException success) {}
1006 }
1007
1008 /**
1009 * Constructor throws if keepAliveTime is less than zero
1010 */
1011 public void testConstructor14() {
1012 try {
1013 new CustomTPE(1, 2, -1L, SECONDS,
1014 new ArrayBlockingQueue<Runnable>(10),
1015 new NoOpREHandler());
1016 shouldThrow();
1017 } catch (IllegalArgumentException success) {}
1018 }
1019
1020 /**
1021 * Constructor throws if corePoolSize is greater than the maximumPoolSize
1022 */
1023 public void testConstructor15() {
1024 try {
1025 new CustomTPE(2, 1, 1L, SECONDS,
1026 new ArrayBlockingQueue<Runnable>(10),
1027 new NoOpREHandler());
1028 shouldThrow();
1029 } catch (IllegalArgumentException success) {}
1030 }
1031
1032 /**
1033 * Constructor throws if workQueue is set to null
1034 */
1035 public void testConstructorNullPointerException4() {
1036 try {
1037 new CustomTPE(1, 2, 1L, SECONDS,
1038 null,
1039 new NoOpREHandler());
1040 shouldThrow();
1041 } catch (NullPointerException success) {}
1042 }
1043
1044 /**
1045 * Constructor throws if handler is set to null
1046 */
1047 public void testConstructorNullPointerException5() {
1048 try {
1049 new CustomTPE(1, 2, 1L, SECONDS,
1050 new ArrayBlockingQueue<Runnable>(10),
1051 (RejectedExecutionHandler) null);
1052 shouldThrow();
1053 } catch (NullPointerException success) {}
1054 }
1055
1056 /**
1057 * Constructor throws if corePoolSize argument is less than zero
1058 */
1059 public void testConstructor16() {
1060 try {
1061 new CustomTPE(-1, 1, 1L, SECONDS,
1062 new ArrayBlockingQueue<Runnable>(10),
1063 new SimpleThreadFactory(),
1064 new NoOpREHandler());
1065 shouldThrow();
1066 } catch (IllegalArgumentException success) {}
1067 }
1068
1069 /**
1070 * Constructor throws if maximumPoolSize is less than zero
1071 */
1072 public void testConstructor17() {
1073 try {
1074 new CustomTPE(1, -1, 1L, SECONDS,
1075 new ArrayBlockingQueue<Runnable>(10),
1076 new SimpleThreadFactory(),
1077 new NoOpREHandler());
1078 shouldThrow();
1079 } catch (IllegalArgumentException success) {}
1080 }
1081
1082 /**
1083 * Constructor throws if maximumPoolSize is equal to zero
1084 */
1085 public void testConstructor18() {
1086 try {
1087 new CustomTPE(1, 0, 1L, SECONDS,
1088 new ArrayBlockingQueue<Runnable>(10),
1089 new SimpleThreadFactory(),
1090 new NoOpREHandler());
1091 shouldThrow();
1092 } catch (IllegalArgumentException success) {}
1093 }
1094
1095 /**
1096 * Constructor throws if keepAliveTime is less than zero
1097 */
1098 public void testConstructor19() {
1099 try {
1100 new CustomTPE(1, 2, -1L, SECONDS,
1101 new ArrayBlockingQueue<Runnable>(10),
1102 new SimpleThreadFactory(),
1103 new NoOpREHandler());
1104 shouldThrow();
1105 } catch (IllegalArgumentException success) {}
1106 }
1107
1108 /**
1109 * Constructor throws if corePoolSize is greater than the maximumPoolSize
1110 */
1111 public void testConstructor20() {
1112 try {
1113 new CustomTPE(2, 1, 1L, SECONDS,
1114 new ArrayBlockingQueue<Runnable>(10),
1115 new SimpleThreadFactory(),
1116 new NoOpREHandler());
1117 shouldThrow();
1118 } catch (IllegalArgumentException success) {}
1119 }
1120
1121 /**
1122 * Constructor throws if workQueue is null
1123 */
1124 public void testConstructorNullPointerException6() {
1125 try {
1126 new CustomTPE(1, 2, 1L, SECONDS,
1127 null,
1128 new SimpleThreadFactory(),
1129 new NoOpREHandler());
1130 shouldThrow();
1131 } catch (NullPointerException success) {}
1132 }
1133
1134 /**
1135 * Constructor throws if handler is null
1136 */
1137 public void testConstructorNullPointerException7() {
1138 try {
1139 new CustomTPE(1, 2, 1L, SECONDS,
1140 new ArrayBlockingQueue<Runnable>(10),
1141 new SimpleThreadFactory(),
1142 (RejectedExecutionHandler) null);
1143 shouldThrow();
1144 } catch (NullPointerException success) {}
1145 }
1146
1147 /**
1148 * Constructor throws if ThreadFactory is null
1149 */
1150 public void testConstructorNullPointerException8() {
1151 try {
1152 new CustomTPE(1, 2, 1L, SECONDS,
1153 new ArrayBlockingQueue<Runnable>(10),
1154 (ThreadFactory) null,
1155 new NoOpREHandler());
1156 shouldThrow();
1157 } catch (NullPointerException success) {}
1158 }
1159
1160 /**
1161 * Submitted tasks are rejected when saturated or shutdown
1162 */
1163 public void testSubmittedTasksRejectedWhenSaturatedOrShutdown() throws InterruptedException {
1164 final ThreadPoolExecutor p =
1165 new CustomTPE(1, 1,
1166 LONG_DELAY_MS, MILLISECONDS,
1167 new ArrayBlockingQueue<Runnable>(1));
1168 final int saturatedSize = saturatedSize(p);
1169 final ThreadLocalRandom rnd = ThreadLocalRandom.current();
1170 final CountDownLatch threadsStarted = new CountDownLatch(p.getMaximumPoolSize());
1171 final CountDownLatch done = new CountDownLatch(1);
1172 final Runnable r = () -> {
1173 threadsStarted.countDown();
1174 for (;;) {
1175 try {
1176 done.await();
1177 return;
1178 } catch (InterruptedException shutdownNowDeliberatelyIgnored) {}
1179 }};
1180 final Callable<Boolean> c = () -> {
1181 threadsStarted.countDown();
1182 for (;;) {
1183 try {
1184 done.await();
1185 return Boolean.TRUE;
1186 } catch (InterruptedException shutdownNowDeliberatelyIgnored) {}
1187 }};
1188 final boolean shutdownNow = rnd.nextBoolean();
1189
1190 try (PoolCleaner cleaner = cleaner(p, done)) {
1191 // saturate
1192 for (int i = saturatedSize; i--> 0; ) {
1193 switch (rnd.nextInt(4)) {
1194 case 0: p.execute(r); break;
1195 case 1: assertFalse(p.submit(r).isDone()); break;
1196 case 2: assertFalse(p.submit(r, Boolean.TRUE).isDone()); break;
1197 case 3: assertFalse(p.submit(c).isDone()); break;
1198 }
1199 }
1200
1201 await(threadsStarted);
1202 assertTaskSubmissionsAreRejected(p);
1203
1204 if (shutdownNow)
1205 p.shutdownNow();
1206 else
1207 p.shutdown();
1208 // Pool is shutdown, but not yet terminated
1209 assertTaskSubmissionsAreRejected(p);
1210 assertFalse(p.isTerminated());
1211
1212 done.countDown(); // release blocking tasks
1213 assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS));
1214
1215 assertTaskSubmissionsAreRejected(p);
1216 }
1217 assertEquals(saturatedSize(p)
1218 - (shutdownNow ? p.getQueue().remainingCapacity() : 0),
1219 p.getCompletedTaskCount());
1220 }
1221
1222 /**
1223 * executor using DiscardOldestPolicy drops oldest task if saturated.
1224 */
1225 public void testSaturatedExecute_DiscardOldestPolicy() {
1226 final CountDownLatch done = new CountDownLatch(1);
1227 LatchAwaiter r1 = awaiter(done);
1228 LatchAwaiter r2 = awaiter(done);
1229 LatchAwaiter r3 = awaiter(done);
1230 final ThreadPoolExecutor p =
1231 new CustomTPE(1, 1,
1232 LONG_DELAY_MS, MILLISECONDS,
1233 new ArrayBlockingQueue<Runnable>(1),
1234 new ThreadPoolExecutor.DiscardOldestPolicy());
1235 try (PoolCleaner cleaner = cleaner(p, done)) {
1236 assertEquals(LatchAwaiter.NEW, r1.state);
1237 assertEquals(LatchAwaiter.NEW, r2.state);
1238 assertEquals(LatchAwaiter.NEW, r3.state);
1239 p.execute(r1);
1240 p.execute(r2);
1241 assertTrue(p.getQueue().contains(r2));
1242 p.execute(r3);
1243 assertFalse(p.getQueue().contains(r2));
1244 assertTrue(p.getQueue().contains(r3));
1245 }
1246 assertEquals(LatchAwaiter.DONE, r1.state);
1247 assertEquals(LatchAwaiter.NEW, r2.state);
1248 assertEquals(LatchAwaiter.DONE, r3.state);
1249 }
1250
1251 /**
1252 * execute using DiscardOldestPolicy drops task on shutdown
1253 */
1254 public void testDiscardOldestOnShutdown() {
1255 final ThreadPoolExecutor p =
1256 new CustomTPE(1, 1,
1257 LONG_DELAY_MS, MILLISECONDS,
1258 new ArrayBlockingQueue<Runnable>(1),
1259 new ThreadPoolExecutor.DiscardOldestPolicy());
1260
1261 try { p.shutdown(); } catch (SecurityException ok) { return; }
1262 try (PoolCleaner cleaner = cleaner(p)) {
1263 TrackedNoOpRunnable r = new TrackedNoOpRunnable();
1264 p.execute(r);
1265 assertFalse(r.done);
1266 }
1267 }
1268
1269 /**
1270 * Submitting null tasks throws NullPointerException
1271 */
1272 public void testNullTaskSubmission() {
1273 final ThreadPoolExecutor p =
1274 new CustomTPE(1, 2,
1275 1L, SECONDS,
1276 new ArrayBlockingQueue<Runnable>(10));
1277 try (PoolCleaner cleaner = cleaner(p)) {
1278 assertNullTaskSubmissionThrowsNullPointerException(p);
1279 }
1280 }
1281
1282 /**
1283 * setCorePoolSize of negative value throws IllegalArgumentException
1284 */
1285 public void testCorePoolSizeIllegalArgumentException() {
1286 final ThreadPoolExecutor p =
1287 new CustomTPE(1, 2,
1288 LONG_DELAY_MS, MILLISECONDS,
1289 new ArrayBlockingQueue<Runnable>(10));
1290 try (PoolCleaner cleaner = cleaner(p)) {
1291 try {
1292 p.setCorePoolSize(-1);
1293 shouldThrow();
1294 } catch (IllegalArgumentException success) {}
1295 }
1296 }
1297
1298 /**
1299 * setMaximumPoolSize(int) throws IllegalArgumentException
1300 * if given a value less the core pool size
1301 */
1302 public void testMaximumPoolSizeIllegalArgumentException() {
1303 final ThreadPoolExecutor p =
1304 new CustomTPE(2, 3,
1305 LONG_DELAY_MS, MILLISECONDS,
1306 new ArrayBlockingQueue<Runnable>(10));
1307 try (PoolCleaner cleaner = cleaner(p)) {
1308 try {
1309 p.setMaximumPoolSize(1);
1310 shouldThrow();
1311 } catch (IllegalArgumentException success) {}
1312 }
1313 }
1314
1315 /**
1316 * setMaximumPoolSize throws IllegalArgumentException
1317 * if given a negative value
1318 */
1319 public void testMaximumPoolSizeIllegalArgumentException2() {
1320 final ThreadPoolExecutor p =
1321 new CustomTPE(2, 3,
1322 LONG_DELAY_MS, MILLISECONDS,
1323 new ArrayBlockingQueue<Runnable>(10));
1324 try (PoolCleaner cleaner = cleaner(p)) {
1325 try {
1326 p.setMaximumPoolSize(-1);
1327 shouldThrow();
1328 } catch (IllegalArgumentException success) {}
1329 }
1330 }
1331
1332 /**
1333 * setKeepAliveTime throws IllegalArgumentException
1334 * when given a negative value
1335 */
1336 public void testKeepAliveTimeIllegalArgumentException() {
1337 final ThreadPoolExecutor p =
1338 new CustomTPE(2, 3,
1339 LONG_DELAY_MS, MILLISECONDS,
1340 new ArrayBlockingQueue<Runnable>(10));
1341 try (PoolCleaner cleaner = cleaner(p)) {
1342 try {
1343 p.setKeepAliveTime(-1, MILLISECONDS);
1344 shouldThrow();
1345 } catch (IllegalArgumentException success) {}
1346 }
1347 }
1348
1349 /**
1350 * terminated() is called on termination
1351 */
1352 public void testTerminated() {
1353 CustomTPE p = new CustomTPE();
1354 try (PoolCleaner cleaner = cleaner(p)) {
1355 try { p.shutdown(); } catch (SecurityException ok) { return; }
1356 assertTrue(p.terminatedCalled());
1357 assertTrue(p.isShutdown());
1358 }
1359 }
1360
1361 /**
1362 * beforeExecute and afterExecute are called when executing task
1363 */
1364 public void testBeforeAfter() throws InterruptedException {
1365 CustomTPE p = new CustomTPE();
1366 try (PoolCleaner cleaner = cleaner(p)) {
1367 final CountDownLatch done = new CountDownLatch(1);
1368 p.execute(new CheckedRunnable() {
1369 public void realRun() {
1370 done.countDown();
1371 }});
1372 await(p.afterCalled);
1373 assertEquals(0, done.getCount());
1374 assertTrue(p.afterCalled());
1375 assertTrue(p.beforeCalled());
1376 }
1377 }
1378
1379 /**
1380 * completed submit of callable returns result
1381 */
1382 public void testSubmitCallable() throws Exception {
1383 final ExecutorService e =
1384 new CustomTPE(2, 2,
1385 LONG_DELAY_MS, MILLISECONDS,
1386 new ArrayBlockingQueue<Runnable>(10));
1387 try (PoolCleaner cleaner = cleaner(e)) {
1388 Future<String> future = e.submit(new StringTask());
1389 String result = future.get();
1390 assertSame(TEST_STRING, result);
1391 }
1392 }
1393
1394 /**
1395 * completed submit of runnable returns successfully
1396 */
1397 public void testSubmitRunnable() throws Exception {
1398 final ExecutorService e =
1399 new CustomTPE(2, 2,
1400 LONG_DELAY_MS, MILLISECONDS,
1401 new ArrayBlockingQueue<Runnable>(10));
1402 try (PoolCleaner cleaner = cleaner(e)) {
1403 Future<?> future = e.submit(new NoOpRunnable());
1404 future.get();
1405 assertTrue(future.isDone());
1406 }
1407 }
1408
1409 /**
1410 * completed submit of (runnable, result) returns result
1411 */
1412 public void testSubmitRunnable2() throws Exception {
1413 final ExecutorService e =
1414 new CustomTPE(2, 2,
1415 LONG_DELAY_MS, MILLISECONDS,
1416 new ArrayBlockingQueue<Runnable>(10));
1417 try (PoolCleaner cleaner = cleaner(e)) {
1418 Future<String> future = e.submit(new NoOpRunnable(), TEST_STRING);
1419 String result = future.get();
1420 assertSame(TEST_STRING, result);
1421 }
1422 }
1423
1424 /**
1425 * invokeAny(null) throws NullPointerException
1426 */
1427 public void testInvokeAny1() throws Exception {
1428 final ExecutorService e =
1429 new CustomTPE(2, 2,
1430 LONG_DELAY_MS, MILLISECONDS,
1431 new ArrayBlockingQueue<Runnable>(10));
1432 try (PoolCleaner cleaner = cleaner(e)) {
1433 try {
1434 e.invokeAny(null);
1435 shouldThrow();
1436 } catch (NullPointerException success) {}
1437 }
1438 }
1439
1440 /**
1441 * invokeAny(empty collection) throws IllegalArgumentException
1442 */
1443 public void testInvokeAny2() throws Exception {
1444 final ExecutorService e =
1445 new CustomTPE(2, 2,
1446 LONG_DELAY_MS, MILLISECONDS,
1447 new ArrayBlockingQueue<Runnable>(10));
1448 try (PoolCleaner cleaner = cleaner(e)) {
1449 try {
1450 e.invokeAny(new ArrayList<Callable<String>>());
1451 shouldThrow();
1452 } catch (IllegalArgumentException success) {}
1453 }
1454 }
1455
1456 /**
1457 * invokeAny(c) throws NPE if c has null elements
1458 */
1459 public void testInvokeAny3() throws Exception {
1460 CountDownLatch latch = new CountDownLatch(1);
1461 final ExecutorService e =
1462 new CustomTPE(2, 2,
1463 LONG_DELAY_MS, MILLISECONDS,
1464 new ArrayBlockingQueue<Runnable>(10));
1465 try (PoolCleaner cleaner = cleaner(e)) {
1466 List<Callable<String>> l = new ArrayList<>();
1467 l.add(latchAwaitingStringTask(latch));
1468 l.add(null);
1469 try {
1470 e.invokeAny(l);
1471 shouldThrow();
1472 } catch (NullPointerException success) {}
1473 latch.countDown();
1474 }
1475 }
1476
1477 /**
1478 * invokeAny(c) throws ExecutionException if no task completes
1479 */
1480 public void testInvokeAny4() throws Exception {
1481 final ExecutorService e =
1482 new CustomTPE(2, 2,
1483 LONG_DELAY_MS, MILLISECONDS,
1484 new ArrayBlockingQueue<Runnable>(10));
1485 try (PoolCleaner cleaner = cleaner(e)) {
1486 List<Callable<String>> l = new ArrayList<>();
1487 l.add(new NPETask());
1488 try {
1489 e.invokeAny(l);
1490 shouldThrow();
1491 } catch (ExecutionException success) {
1492 assertTrue(success.getCause() instanceof NullPointerException);
1493 }
1494 }
1495 }
1496
1497 /**
1498 * invokeAny(c) returns result of some task
1499 */
1500 public void testInvokeAny5() throws Exception {
1501 final ExecutorService e =
1502 new CustomTPE(2, 2,
1503 LONG_DELAY_MS, MILLISECONDS,
1504 new ArrayBlockingQueue<Runnable>(10));
1505 try (PoolCleaner cleaner = cleaner(e)) {
1506 List<Callable<String>> l = new ArrayList<>();
1507 l.add(new StringTask());
1508 l.add(new StringTask());
1509 String result = e.invokeAny(l);
1510 assertSame(TEST_STRING, result);
1511 }
1512 }
1513
1514 /**
1515 * invokeAll(null) throws NPE
1516 */
1517 public void testInvokeAll1() throws Exception {
1518 final ExecutorService e =
1519 new CustomTPE(2, 2,
1520 LONG_DELAY_MS, MILLISECONDS,
1521 new ArrayBlockingQueue<Runnable>(10));
1522 try (PoolCleaner cleaner = cleaner(e)) {
1523 try {
1524 e.invokeAll(null);
1525 shouldThrow();
1526 } catch (NullPointerException success) {}
1527 }
1528 }
1529
1530 /**
1531 * invokeAll(empty collection) returns empty list
1532 */
1533 public void testInvokeAll2() throws Exception {
1534 final ExecutorService e =
1535 new CustomTPE(2, 2,
1536 LONG_DELAY_MS, MILLISECONDS,
1537 new ArrayBlockingQueue<Runnable>(10));
1538 final Collection<Callable<String>> emptyCollection
1539 = Collections.emptyList();
1540 try (PoolCleaner cleaner = cleaner(e)) {
1541 List<Future<String>> r = e.invokeAll(emptyCollection);
1542 assertTrue(r.isEmpty());
1543 }
1544 }
1545
1546 /**
1547 * invokeAll(c) throws NPE if c has null elements
1548 */
1549 public void testInvokeAll3() throws Exception {
1550 final ExecutorService e =
1551 new CustomTPE(2, 2,
1552 LONG_DELAY_MS, MILLISECONDS,
1553 new ArrayBlockingQueue<Runnable>(10));
1554 try (PoolCleaner cleaner = cleaner(e)) {
1555 List<Callable<String>> l = new ArrayList<>();
1556 l.add(new StringTask());
1557 l.add(null);
1558 try {
1559 e.invokeAll(l);
1560 shouldThrow();
1561 } catch (NullPointerException success) {}
1562 }
1563 }
1564
1565 /**
1566 * get of element of invokeAll(c) throws exception on failed task
1567 */
1568 public void testInvokeAll4() throws Exception {
1569 final ExecutorService e =
1570 new CustomTPE(2, 2,
1571 LONG_DELAY_MS, MILLISECONDS,
1572 new ArrayBlockingQueue<Runnable>(10));
1573 try (PoolCleaner cleaner = cleaner(e)) {
1574 List<Callable<String>> l = new ArrayList<>();
1575 l.add(new NPETask());
1576 List<Future<String>> futures = e.invokeAll(l);
1577 assertEquals(1, futures.size());
1578 try {
1579 futures.get(0).get();
1580 shouldThrow();
1581 } catch (ExecutionException success) {
1582 assertTrue(success.getCause() instanceof NullPointerException);
1583 }
1584 }
1585 }
1586
1587 /**
1588 * invokeAll(c) returns results of all completed tasks
1589 */
1590 public void testInvokeAll5() throws Exception {
1591 final ExecutorService e =
1592 new CustomTPE(2, 2,
1593 LONG_DELAY_MS, MILLISECONDS,
1594 new ArrayBlockingQueue<Runnable>(10));
1595 try (PoolCleaner cleaner = cleaner(e)) {
1596 List<Callable<String>> l = new ArrayList<>();
1597 l.add(new StringTask());
1598 l.add(new StringTask());
1599 List<Future<String>> futures = e.invokeAll(l);
1600 assertEquals(2, futures.size());
1601 for (Future<String> future : futures)
1602 assertSame(TEST_STRING, future.get());
1603 }
1604 }
1605
1606 /**
1607 * timed invokeAny(null) throws NPE
1608 */
1609 public void testTimedInvokeAny1() throws Exception {
1610 final ExecutorService e =
1611 new CustomTPE(2, 2,
1612 LONG_DELAY_MS, MILLISECONDS,
1613 new ArrayBlockingQueue<Runnable>(10));
1614 try (PoolCleaner cleaner = cleaner(e)) {
1615 try {
1616 e.invokeAny(null, randomTimeout(), randomTimeUnit());
1617 shouldThrow();
1618 } catch (NullPointerException success) {}
1619 }
1620 }
1621
1622 /**
1623 * timed invokeAny(,,null) throws NPE
1624 */
1625 public void testTimedInvokeAnyNullTimeUnit() throws Exception {
1626 final ExecutorService e =
1627 new CustomTPE(2, 2,
1628 LONG_DELAY_MS, MILLISECONDS,
1629 new ArrayBlockingQueue<Runnable>(10));
1630 try (PoolCleaner cleaner = cleaner(e)) {
1631 List<Callable<String>> l = new ArrayList<>();
1632 l.add(new StringTask());
1633 try {
1634 e.invokeAny(l, randomTimeout(), null);
1635 shouldThrow();
1636 } catch (NullPointerException success) {}
1637 }
1638 }
1639
1640 /**
1641 * timed invokeAny(empty collection) throws IllegalArgumentException
1642 */
1643 public void testTimedInvokeAny2() throws Exception {
1644 final ExecutorService e =
1645 new CustomTPE(2, 2,
1646 LONG_DELAY_MS, MILLISECONDS,
1647 new ArrayBlockingQueue<Runnable>(10));
1648 final Collection<Callable<String>> emptyCollection
1649 = Collections.emptyList();
1650 try (PoolCleaner cleaner = cleaner(e)) {
1651 try {
1652 e.invokeAny(emptyCollection, randomTimeout(), randomTimeUnit());
1653 shouldThrow();
1654 } catch (IllegalArgumentException success) {}
1655 }
1656 }
1657
1658 /**
1659 * timed invokeAny(c) throws NPE if c has null elements
1660 */
1661 public void testTimedInvokeAny3() throws Exception {
1662 CountDownLatch latch = new CountDownLatch(1);
1663 final ExecutorService e =
1664 new CustomTPE(2, 2,
1665 LONG_DELAY_MS, MILLISECONDS,
1666 new ArrayBlockingQueue<Runnable>(10));
1667 try (PoolCleaner cleaner = cleaner(e)) {
1668 List<Callable<String>> l = new ArrayList<>();
1669 l.add(latchAwaitingStringTask(latch));
1670 l.add(null);
1671 try {
1672 e.invokeAny(l, randomTimeout(), randomTimeUnit());
1673 shouldThrow();
1674 } catch (NullPointerException success) {}
1675 latch.countDown();
1676 }
1677 }
1678
1679 /**
1680 * timed invokeAny(c) throws ExecutionException if no task completes
1681 */
1682 public void testTimedInvokeAny4() throws Exception {
1683 final ExecutorService e =
1684 new CustomTPE(2, 2,
1685 LONG_DELAY_MS, MILLISECONDS,
1686 new ArrayBlockingQueue<Runnable>(10));
1687 try (PoolCleaner cleaner = cleaner(e)) {
1688 long startTime = System.nanoTime();
1689 List<Callable<String>> l = new ArrayList<>();
1690 l.add(new NPETask());
1691 try {
1692 e.invokeAny(l, LONG_DELAY_MS, MILLISECONDS);
1693 shouldThrow();
1694 } catch (ExecutionException success) {
1695 assertTrue(success.getCause() instanceof NullPointerException);
1696 }
1697 assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS);
1698 }
1699 }
1700
1701 /**
1702 * timed invokeAny(c) returns result of some task
1703 */
1704 public void testTimedInvokeAny5() throws Exception {
1705 final ExecutorService e =
1706 new CustomTPE(2, 2,
1707 LONG_DELAY_MS, MILLISECONDS,
1708 new ArrayBlockingQueue<Runnable>(10));
1709 try (PoolCleaner cleaner = cleaner(e)) {
1710 long startTime = System.nanoTime();
1711 List<Callable<String>> l = new ArrayList<>();
1712 l.add(new StringTask());
1713 l.add(new StringTask());
1714 String result = e.invokeAny(l, LONG_DELAY_MS, MILLISECONDS);
1715 assertSame(TEST_STRING, result);
1716 assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS);
1717 }
1718 }
1719
1720 /**
1721 * timed invokeAll(null) throws NPE
1722 */
1723 public void testTimedInvokeAll1() throws Exception {
1724 final ExecutorService e =
1725 new CustomTPE(2, 2,
1726 LONG_DELAY_MS, MILLISECONDS,
1727 new ArrayBlockingQueue<Runnable>(10));
1728 try (PoolCleaner cleaner = cleaner(e)) {
1729 try {
1730 e.invokeAll(null, randomTimeout(), randomTimeUnit());
1731 shouldThrow();
1732 } catch (NullPointerException success) {}
1733 }
1734 }
1735
1736 /**
1737 * timed invokeAll(,,null) throws NPE
1738 */
1739 public void testTimedInvokeAllNullTimeUnit() throws Exception {
1740 final ExecutorService e =
1741 new CustomTPE(2, 2,
1742 LONG_DELAY_MS, MILLISECONDS,
1743 new ArrayBlockingQueue<Runnable>(10));
1744 try (PoolCleaner cleaner = cleaner(e)) {
1745 List<Callable<String>> l = new ArrayList<>();
1746 l.add(new StringTask());
1747 try {
1748 e.invokeAll(l, randomTimeout(), null);
1749 shouldThrow();
1750 } catch (NullPointerException success) {}
1751 }
1752 }
1753
1754 /**
1755 * timed invokeAll(empty collection) returns empty list
1756 */
1757 public void testTimedInvokeAll2() throws Exception {
1758 final ExecutorService e =
1759 new CustomTPE(2, 2,
1760 LONG_DELAY_MS, MILLISECONDS,
1761 new ArrayBlockingQueue<Runnable>(10));
1762 final Collection<Callable<String>> emptyCollection
1763 = Collections.emptyList();
1764 try (PoolCleaner cleaner = cleaner(e)) {
1765 List<Future<String>> r =
1766 e.invokeAll(emptyCollection, randomTimeout(), randomTimeUnit());
1767 assertTrue(r.isEmpty());
1768 }
1769 }
1770
1771 /**
1772 * timed invokeAll(c) throws NPE if c has null elements
1773 */
1774 public void testTimedInvokeAll3() throws Exception {
1775 final ExecutorService e =
1776 new CustomTPE(2, 2,
1777 LONG_DELAY_MS, MILLISECONDS,
1778 new ArrayBlockingQueue<Runnable>(10));
1779 try (PoolCleaner cleaner = cleaner(e)) {
1780 List<Callable<String>> l = new ArrayList<>();
1781 l.add(new StringTask());
1782 l.add(null);
1783 try {
1784 e.invokeAll(l, randomTimeout(), randomTimeUnit());
1785 shouldThrow();
1786 } catch (NullPointerException success) {}
1787 }
1788 }
1789
1790 /**
1791 * get of element of invokeAll(c) throws exception on failed task
1792 */
1793 public void testTimedInvokeAll4() throws Exception {
1794 final ExecutorService e =
1795 new CustomTPE(2, 2,
1796 LONG_DELAY_MS, MILLISECONDS,
1797 new ArrayBlockingQueue<Runnable>(10));
1798 try (PoolCleaner cleaner = cleaner(e)) {
1799 List<Callable<String>> l = new ArrayList<>();
1800 l.add(new NPETask());
1801 List<Future<String>> futures =
1802 e.invokeAll(l, LONG_DELAY_MS, MILLISECONDS);
1803 assertEquals(1, futures.size());
1804 try {
1805 futures.get(0).get();
1806 shouldThrow();
1807 } catch (ExecutionException success) {
1808 assertTrue(success.getCause() instanceof NullPointerException);
1809 }
1810 }
1811 }
1812
1813 /**
1814 * timed invokeAll(c) returns results of all completed tasks
1815 */
1816 public void testTimedInvokeAll5() throws Exception {
1817 final ExecutorService e =
1818 new CustomTPE(2, 2,
1819 LONG_DELAY_MS, MILLISECONDS,
1820 new ArrayBlockingQueue<Runnable>(10));
1821 try (PoolCleaner cleaner = cleaner(e)) {
1822 List<Callable<String>> l = new ArrayList<>();
1823 l.add(new StringTask());
1824 l.add(new StringTask());
1825 List<Future<String>> futures =
1826 e.invokeAll(l, LONG_DELAY_MS, MILLISECONDS);
1827 assertEquals(2, futures.size());
1828 for (Future<String> future : futures)
1829 assertSame(TEST_STRING, future.get());
1830 }
1831 }
1832
1833 /**
1834 * timed invokeAll(c) cancels tasks not completed by timeout
1835 */
1836 public void testTimedInvokeAll6() throws Exception {
1837 for (long timeout = timeoutMillis();;) {
1838 final CountDownLatch done = new CountDownLatch(1);
1839 final Callable<String> waiter = new CheckedCallable<String>() {
1840 public String realCall() {
1841 try { done.await(LONG_DELAY_MS, MILLISECONDS); }
1842 catch (InterruptedException ok) {}
1843 return "1"; }};
1844 final ExecutorService p =
1845 new CustomTPE(2, 2,
1846 LONG_DELAY_MS, MILLISECONDS,
1847 new ArrayBlockingQueue<Runnable>(10));
1848 try (PoolCleaner cleaner = cleaner(p, done)) {
1849 List<Callable<String>> tasks = new ArrayList<>();
1850 tasks.add(new StringTask("0"));
1851 tasks.add(waiter);
1852 tasks.add(new StringTask("2"));
1853 long startTime = System.nanoTime();
1854 List<Future<String>> futures =
1855 p.invokeAll(tasks, timeout, MILLISECONDS);
1856 assertEquals(tasks.size(), futures.size());
1857 assertTrue(millisElapsedSince(startTime) >= timeout);
1858 for (Future future : futures)
1859 assertTrue(future.isDone());
1860 assertTrue(futures.get(1).isCancelled());
1861 try {
1862 assertEquals("0", futures.get(0).get());
1863 assertEquals("2", futures.get(2).get());
1864 break;
1865 } catch (CancellationException retryWithLongerTimeout) {
1866 timeout *= 2;
1867 if (timeout >= LONG_DELAY_MS / 2)
1868 fail("expected exactly one task to be cancelled");
1869 }
1870 }
1871 }
1872 }
1873
1874 /**
1875 * Execution continues if there is at least one thread even if
1876 * thread factory fails to create more
1877 */
1878 public void testFailingThreadFactory() throws InterruptedException {
1879 final ExecutorService e =
1880 new CustomTPE(100, 100,
1881 LONG_DELAY_MS, MILLISECONDS,
1882 new LinkedBlockingQueue<Runnable>(),
1883 new FailingThreadFactory());
1884 try (PoolCleaner cleaner = cleaner(e)) {
1885 final int TASKS = 100;
1886 final CountDownLatch done = new CountDownLatch(TASKS);
1887 for (int k = 0; k < TASKS; ++k)
1888 e.execute(new CheckedRunnable() {
1889 public void realRun() {
1890 done.countDown();
1891 }});
1892 await(done);
1893 }
1894 }
1895
1896 /**
1897 * allowsCoreThreadTimeOut is by default false.
1898 */
1899 public void testAllowsCoreThreadTimeOut() {
1900 final ThreadPoolExecutor p =
1901 new CustomTPE(2, 2,
1902 1000, MILLISECONDS,
1903 new ArrayBlockingQueue<Runnable>(10));
1904 try (PoolCleaner cleaner = cleaner(p)) {
1905 assertFalse(p.allowsCoreThreadTimeOut());
1906 }
1907 }
1908
1909 /**
1910 * allowCoreThreadTimeOut(true) causes idle threads to time out
1911 */
1912 public void testAllowCoreThreadTimeOut_true() throws Exception {
1913 long keepAliveTime = timeoutMillis();
1914 final ThreadPoolExecutor p =
1915 new CustomTPE(2, 10,
1916 keepAliveTime, MILLISECONDS,
1917 new ArrayBlockingQueue<Runnable>(10));
1918 try (PoolCleaner cleaner = cleaner(p)) {
1919 final CountDownLatch threadStarted = new CountDownLatch(1);
1920 p.allowCoreThreadTimeOut(true);
1921 p.execute(new CheckedRunnable() {
1922 public void realRun() {
1923 threadStarted.countDown();
1924 assertEquals(1, p.getPoolSize());
1925 }});
1926 await(threadStarted);
1927 delay(keepAliveTime);
1928 long startTime = System.nanoTime();
1929 while (p.getPoolSize() > 0
1930 && millisElapsedSince(startTime) < LONG_DELAY_MS)
1931 Thread.yield();
1932 assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS);
1933 assertEquals(0, p.getPoolSize());
1934 }
1935 }
1936
1937 /**
1938 * allowCoreThreadTimeOut(false) causes idle threads not to time out
1939 */
1940 public void testAllowCoreThreadTimeOut_false() throws Exception {
1941 long keepAliveTime = timeoutMillis();
1942 final ThreadPoolExecutor p =
1943 new CustomTPE(2, 10,
1944 keepAliveTime, MILLISECONDS,
1945 new ArrayBlockingQueue<Runnable>(10));
1946 try (PoolCleaner cleaner = cleaner(p)) {
1947 final CountDownLatch threadStarted = new CountDownLatch(1);
1948 p.allowCoreThreadTimeOut(false);
1949 p.execute(new CheckedRunnable() {
1950 public void realRun() throws InterruptedException {
1951 threadStarted.countDown();
1952 assertTrue(p.getPoolSize() >= 1);
1953 }});
1954 delay(2 * keepAliveTime);
1955 assertTrue(p.getPoolSize() >= 1);
1956 }
1957 }
1958
1959 /**
1960 * get(cancelled task) throws CancellationException
1961 * (in part, a test of CustomTPE itself)
1962 */
1963 public void testGet_cancelled() throws Exception {
1964 final CountDownLatch done = new CountDownLatch(1);
1965 final ExecutorService e =
1966 new CustomTPE(1, 1,
1967 LONG_DELAY_MS, MILLISECONDS,
1968 new LinkedBlockingQueue<Runnable>());
1969 try (PoolCleaner cleaner = cleaner(e, done)) {
1970 final CountDownLatch blockerStarted = new CountDownLatch(1);
1971 final List<Future<?>> futures = new ArrayList<>();
1972 for (int i = 0; i < 2; i++) {
1973 Runnable r = new CheckedRunnable() { public void realRun()
1974 throws Throwable {
1975 blockerStarted.countDown();
1976 assertTrue(done.await(2 * LONG_DELAY_MS, MILLISECONDS));
1977 }};
1978 futures.add(e.submit(r));
1979 }
1980 await(blockerStarted);
1981 for (Future<?> future : futures) future.cancel(false);
1982 for (Future<?> future : futures) {
1983 try {
1984 future.get();
1985 shouldThrow();
1986 } catch (CancellationException success) {}
1987 try {
1988 future.get(LONG_DELAY_MS, MILLISECONDS);
1989 shouldThrow();
1990 } catch (CancellationException success) {}
1991 assertTrue(future.isCancelled());
1992 assertTrue(future.isDone());
1993 }
1994 }
1995 }
1996
1997 public void testFinalizeMethodCallsSuperFinalize() {
1998 new CustomTPE(1, 1,
1999 LONG_DELAY_MS, MILLISECONDS,
2000 new LinkedBlockingQueue<Runnable>()) {
2001
2002 /**
2003 * A finalize method without "throws Throwable", that
2004 * calls super.finalize().
2005 */
2006 protected void finalize() {
2007 super.finalize();
2008 }
2009 }.shutdown();
2010 }
2011
2012 }