1 /* 2 * Written by Doug Lea with assistance from members of JCP JSR-166 3 * Expert Group and released to the public domain, as explained at 4 * http://creativecommons.org/publicdomain/zero/1.0/ 5 */ 6 7 package jsr166; 8 9 import static java.util.concurrent.TimeUnit.MILLISECONDS; 10 import static java.util.concurrent.TimeUnit.NANOSECONDS; 11 import static java.util.concurrent.TimeUnit.SECONDS; 12 13 import java.util.ArrayList; 14 import java.util.HashSet; 15 import java.util.List; 16 import java.util.concurrent.BlockingQueue; 17 import java.util.concurrent.Callable; 18 import java.util.concurrent.CancellationException; 19 import java.util.concurrent.CountDownLatch; 20 import java.util.concurrent.Delayed; 21 import java.util.concurrent.ExecutionException; 22 import java.util.concurrent.Executors; 23 import java.util.concurrent.ExecutorService; 24 import java.util.concurrent.Future; 25 import java.util.concurrent.RejectedExecutionException; 26 import java.util.concurrent.RejectedExecutionHandler; 27 import java.util.concurrent.RunnableScheduledFuture; 28 import java.util.concurrent.ScheduledFuture; 29 import java.util.concurrent.ScheduledThreadPoolExecutor; 30 import java.util.concurrent.ThreadFactory; 31 import java.util.concurrent.ThreadPoolExecutor; 32 import java.util.concurrent.TimeoutException; 33 import java.util.concurrent.TimeUnit; 34 import java.util.concurrent.atomic.AtomicBoolean; 35 import java.util.concurrent.atomic.AtomicInteger; 36 import java.util.concurrent.atomic.AtomicLong; 37 38 import junit.framework.Test; 39 import junit.framework.TestSuite; 40 41 public class ScheduledExecutorSubclassTest extends JSR166TestCase { 42 // android-note: Removed because the CTS runner does a bad job of 43 // retrying tests that have suite() declarations. 44 // 45 // public static void main(String[] args) { 46 // main(suite(), args); 47 // } 48 // public static Test suite() { 49 // return new TestSuite(ScheduledExecutorSubclassTest.class); 50 // } 51 52 static class CustomTask<V> implements RunnableScheduledFuture<V> { 53 RunnableScheduledFuture<V> task; 54 volatile boolean ran; CustomTask(RunnableScheduledFuture<V> t)55 CustomTask(RunnableScheduledFuture<V> t) { task = t; } isPeriodic()56 public boolean isPeriodic() { return task.isPeriodic(); } run()57 public void run() { 58 ran = true; 59 task.run(); 60 } getDelay(TimeUnit unit)61 public long getDelay(TimeUnit unit) { return task.getDelay(unit); } compareTo(Delayed t)62 public int compareTo(Delayed t) { 63 return task.compareTo(((CustomTask)t).task); 64 } cancel(boolean mayInterruptIfRunning)65 public boolean cancel(boolean mayInterruptIfRunning) { 66 return task.cancel(mayInterruptIfRunning); 67 } isCancelled()68 public boolean isCancelled() { return task.isCancelled(); } isDone()69 public boolean isDone() { return task.isDone(); } get()70 public V get() throws InterruptedException, ExecutionException { 71 V v = task.get(); 72 assertTrue(ran); 73 return v; 74 } get(long time, TimeUnit unit)75 public V get(long time, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { 76 V v = task.get(time, unit); 77 assertTrue(ran); 78 return v; 79 } 80 } 81 82 public class CustomExecutor extends ScheduledThreadPoolExecutor { 83 decorateTask(Runnable r, RunnableScheduledFuture<V> task)84 protected <V> RunnableScheduledFuture<V> decorateTask(Runnable r, RunnableScheduledFuture<V> task) { 85 return new CustomTask<V>(task); 86 } 87 decorateTask(Callable<V> c, RunnableScheduledFuture<V> task)88 protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> c, RunnableScheduledFuture<V> task) { 89 return new CustomTask<V>(task); 90 } CustomExecutor(int corePoolSize)91 CustomExecutor(int corePoolSize) { super(corePoolSize); } CustomExecutor(int corePoolSize, RejectedExecutionHandler handler)92 CustomExecutor(int corePoolSize, RejectedExecutionHandler handler) { 93 super(corePoolSize, handler); 94 } 95 CustomExecutor(int corePoolSize, ThreadFactory threadFactory)96 CustomExecutor(int corePoolSize, ThreadFactory threadFactory) { 97 super(corePoolSize, threadFactory); 98 } CustomExecutor(int corePoolSize, ThreadFactory threadFactory, RejectedExecutionHandler handler)99 CustomExecutor(int corePoolSize, ThreadFactory threadFactory, 100 RejectedExecutionHandler handler) { 101 super(corePoolSize, threadFactory, handler); 102 } 103 104 } 105 106 /** 107 * execute successfully executes a runnable 108 */ testExecute()109 public void testExecute() throws InterruptedException { 110 final CustomExecutor p = new CustomExecutor(1); 111 try (PoolCleaner cleaner = cleaner(p)) { 112 final CountDownLatch done = new CountDownLatch(1); 113 final Runnable task = new CheckedRunnable() { 114 public void realRun() { done.countDown(); }}; 115 p.execute(task); 116 await(done); 117 } 118 } 119 120 /** 121 * delayed schedule of callable successfully executes after delay 122 */ testSchedule1()123 public void testSchedule1() throws Exception { 124 final CountDownLatch done = new CountDownLatch(1); 125 final CustomExecutor p = new CustomExecutor(1); 126 try (PoolCleaner cleaner = cleaner(p, done)) { 127 final long startTime = System.nanoTime(); 128 Callable task = new CheckedCallable<Boolean>() { 129 public Boolean realCall() { 130 done.countDown(); 131 assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); 132 return Boolean.TRUE; 133 }}; 134 Future f = p.schedule(task, timeoutMillis(), MILLISECONDS); 135 assertSame(Boolean.TRUE, f.get()); 136 assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); 137 } 138 } 139 140 /** 141 * delayed schedule of runnable successfully executes after delay 142 */ testSchedule3()143 public void testSchedule3() throws Exception { 144 final CustomExecutor p = new CustomExecutor(1); 145 try (PoolCleaner cleaner = cleaner(p)) { 146 final long startTime = System.nanoTime(); 147 final CountDownLatch done = new CountDownLatch(1); 148 Runnable task = new CheckedRunnable() { 149 public void realRun() { 150 done.countDown(); 151 assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); 152 }}; 153 Future f = p.schedule(task, timeoutMillis(), MILLISECONDS); 154 await(done); 155 assertNull(f.get(LONG_DELAY_MS, MILLISECONDS)); 156 assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); 157 } 158 } 159 160 /** 161 * scheduleAtFixedRate executes runnable after given initial delay 162 */ testSchedule4()163 public void testSchedule4() throws InterruptedException { 164 final CustomExecutor p = new CustomExecutor(1); 165 try (PoolCleaner cleaner = cleaner(p)) { 166 final long startTime = System.nanoTime(); 167 final CountDownLatch done = new CountDownLatch(1); 168 Runnable task = new CheckedRunnable() { 169 public void realRun() { 170 done.countDown(); 171 assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); 172 }}; 173 ScheduledFuture f = 174 p.scheduleAtFixedRate(task, timeoutMillis(), 175 LONG_DELAY_MS, MILLISECONDS); 176 await(done); 177 assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); 178 f.cancel(true); 179 } 180 } 181 182 /** 183 * scheduleWithFixedDelay executes runnable after given initial delay 184 */ testSchedule5()185 public void testSchedule5() throws InterruptedException { 186 final CustomExecutor p = new CustomExecutor(1); 187 try (PoolCleaner cleaner = cleaner(p)) { 188 final long startTime = System.nanoTime(); 189 final CountDownLatch done = new CountDownLatch(1); 190 Runnable task = new CheckedRunnable() { 191 public void realRun() { 192 done.countDown(); 193 assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); 194 }}; 195 ScheduledFuture f = 196 p.scheduleWithFixedDelay(task, timeoutMillis(), 197 LONG_DELAY_MS, MILLISECONDS); 198 await(done); 199 assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); 200 f.cancel(true); 201 } 202 } 203 204 static class RunnableCounter implements Runnable { 205 AtomicInteger count = new AtomicInteger(0); run()206 public void run() { count.getAndIncrement(); } 207 } 208 209 /** 210 * scheduleAtFixedRate executes series of tasks at given rate. 211 * Eventually, it must hold that: 212 * cycles - 1 <= elapsedMillis/delay < cycles 213 */ testFixedRateSequence()214 public void testFixedRateSequence() throws InterruptedException { 215 final CustomExecutor p = new CustomExecutor(1); 216 try (PoolCleaner cleaner = cleaner(p)) { 217 for (int delay = 1; delay <= LONG_DELAY_MS; delay *= 3) { 218 final long startTime = System.nanoTime(); 219 final int cycles = 8; 220 final CountDownLatch done = new CountDownLatch(cycles); 221 final Runnable task = new CheckedRunnable() { 222 public void realRun() { done.countDown(); }}; 223 final ScheduledFuture periodicTask = 224 p.scheduleAtFixedRate(task, 0, delay, MILLISECONDS); 225 final int totalDelayMillis = (cycles - 1) * delay; 226 await(done, totalDelayMillis + LONG_DELAY_MS); 227 periodicTask.cancel(true); 228 final long elapsedMillis = millisElapsedSince(startTime); 229 assertTrue(elapsedMillis >= totalDelayMillis); 230 if (elapsedMillis <= cycles * delay) 231 return; 232 // else retry with longer delay 233 } 234 fail("unexpected execution rate"); 235 } 236 } 237 238 /** 239 * scheduleWithFixedDelay executes series of tasks with given period. 240 * Eventually, it must hold that each task starts at least delay and at 241 * most 2 * delay after the termination of the previous task. 242 */ testFixedDelaySequence()243 public void testFixedDelaySequence() throws InterruptedException { 244 final CustomExecutor p = new CustomExecutor(1); 245 try (PoolCleaner cleaner = cleaner(p)) { 246 for (int delay = 1; delay <= LONG_DELAY_MS; delay *= 3) { 247 final long startTime = System.nanoTime(); 248 final AtomicLong previous = new AtomicLong(startTime); 249 final AtomicBoolean tryLongerDelay = new AtomicBoolean(false); 250 final int cycles = 8; 251 final CountDownLatch done = new CountDownLatch(cycles); 252 final int d = delay; 253 final Runnable task = new CheckedRunnable() { 254 public void realRun() { 255 long now = System.nanoTime(); 256 long elapsedMillis 257 = NANOSECONDS.toMillis(now - previous.get()); 258 if (done.getCount() == cycles) { // first execution 259 if (elapsedMillis >= d) 260 tryLongerDelay.set(true); 261 } else { 262 assertTrue(elapsedMillis >= d); 263 if (elapsedMillis >= 2 * d) 264 tryLongerDelay.set(true); 265 } 266 previous.set(now); 267 done.countDown(); 268 }}; 269 final ScheduledFuture periodicTask = 270 p.scheduleWithFixedDelay(task, 0, delay, MILLISECONDS); 271 final int totalDelayMillis = (cycles - 1) * delay; 272 await(done, totalDelayMillis + cycles * LONG_DELAY_MS); 273 periodicTask.cancel(true); 274 final long elapsedMillis = millisElapsedSince(startTime); 275 assertTrue(elapsedMillis >= totalDelayMillis); 276 if (!tryLongerDelay.get()) 277 return; 278 // else retry with longer delay 279 } 280 fail("unexpected execution rate"); 281 } 282 } 283 284 /** 285 * execute(null) throws NPE 286 */ testExecuteNull()287 public void testExecuteNull() throws InterruptedException { 288 final CustomExecutor p = new CustomExecutor(1); 289 try (PoolCleaner cleaner = cleaner(p)) { 290 try { 291 p.execute(null); 292 shouldThrow(); 293 } catch (NullPointerException success) {} 294 } 295 } 296 297 /** 298 * schedule(null) throws NPE 299 */ testScheduleNull()300 public void testScheduleNull() throws InterruptedException { 301 final CustomExecutor p = new CustomExecutor(1); 302 try (PoolCleaner cleaner = cleaner(p)) { 303 try { 304 TrackedCallable callable = null; 305 Future f = p.schedule(callable, SHORT_DELAY_MS, MILLISECONDS); 306 shouldThrow(); 307 } catch (NullPointerException success) {} 308 } 309 } 310 311 /** 312 * execute throws RejectedExecutionException if shutdown 313 */ testSchedule1_RejectedExecutionException()314 public void testSchedule1_RejectedExecutionException() { 315 final CustomExecutor p = new CustomExecutor(1); 316 try (PoolCleaner cleaner = cleaner(p)) { 317 try { 318 p.shutdown(); 319 p.schedule(new NoOpRunnable(), 320 MEDIUM_DELAY_MS, MILLISECONDS); 321 shouldThrow(); 322 } catch (RejectedExecutionException success) { 323 } catch (SecurityException ok) {} 324 } 325 } 326 327 /** 328 * schedule throws RejectedExecutionException if shutdown 329 */ testSchedule2_RejectedExecutionException()330 public void testSchedule2_RejectedExecutionException() { 331 final CustomExecutor p = new CustomExecutor(1); 332 try (PoolCleaner cleaner = cleaner(p)) { 333 try { 334 p.shutdown(); 335 p.schedule(new NoOpCallable(), 336 MEDIUM_DELAY_MS, MILLISECONDS); 337 shouldThrow(); 338 } catch (RejectedExecutionException success) { 339 } catch (SecurityException ok) {} 340 } 341 } 342 343 /** 344 * schedule callable throws RejectedExecutionException if shutdown 345 */ testSchedule3_RejectedExecutionException()346 public void testSchedule3_RejectedExecutionException() { 347 final CustomExecutor p = new CustomExecutor(1); 348 try (PoolCleaner cleaner = cleaner(p)) { 349 try { 350 p.shutdown(); 351 p.schedule(new NoOpCallable(), 352 MEDIUM_DELAY_MS, MILLISECONDS); 353 shouldThrow(); 354 } catch (RejectedExecutionException success) { 355 } catch (SecurityException ok) {} 356 } 357 } 358 359 /** 360 * scheduleAtFixedRate throws RejectedExecutionException if shutdown 361 */ testScheduleAtFixedRate1_RejectedExecutionException()362 public void testScheduleAtFixedRate1_RejectedExecutionException() { 363 final CustomExecutor p = new CustomExecutor(1); 364 try (PoolCleaner cleaner = cleaner(p)) { 365 try { 366 p.shutdown(); 367 p.scheduleAtFixedRate(new NoOpRunnable(), 368 MEDIUM_DELAY_MS, MEDIUM_DELAY_MS, MILLISECONDS); 369 shouldThrow(); 370 } catch (RejectedExecutionException success) { 371 } catch (SecurityException ok) {} 372 } 373 } 374 375 /** 376 * scheduleWithFixedDelay throws RejectedExecutionException if shutdown 377 */ testScheduleWithFixedDelay1_RejectedExecutionException()378 public void testScheduleWithFixedDelay1_RejectedExecutionException() { 379 final CustomExecutor p = new CustomExecutor(1); 380 try (PoolCleaner cleaner = cleaner(p)) { 381 try { 382 p.shutdown(); 383 p.scheduleWithFixedDelay(new NoOpRunnable(), 384 MEDIUM_DELAY_MS, MEDIUM_DELAY_MS, MILLISECONDS); 385 shouldThrow(); 386 } catch (RejectedExecutionException success) { 387 } catch (SecurityException ok) {} 388 } 389 } 390 391 /** 392 * getActiveCount increases but doesn't overestimate, when a 393 * thread becomes active 394 */ testGetActiveCount()395 public void testGetActiveCount() throws InterruptedException { 396 final CountDownLatch done = new CountDownLatch(1); 397 final ThreadPoolExecutor p = new CustomExecutor(2); 398 try (PoolCleaner cleaner = cleaner(p, done)) { 399 final CountDownLatch threadStarted = new CountDownLatch(1); 400 assertEquals(0, p.getActiveCount()); 401 p.execute(new CheckedRunnable() { 402 public void realRun() throws InterruptedException { 403 threadStarted.countDown(); 404 assertEquals(1, p.getActiveCount()); 405 await(done); 406 }}); 407 await(threadStarted); 408 assertEquals(1, p.getActiveCount()); 409 } 410 } 411 412 /** 413 * getCompletedTaskCount increases, but doesn't overestimate, 414 * when tasks complete 415 */ testGetCompletedTaskCount()416 public void testGetCompletedTaskCount() throws InterruptedException { 417 final ThreadPoolExecutor p = new CustomExecutor(2); 418 try (PoolCleaner cleaner = cleaner(p)) { 419 final CountDownLatch threadStarted = new CountDownLatch(1); 420 final CountDownLatch threadProceed = new CountDownLatch(1); 421 final CountDownLatch threadDone = new CountDownLatch(1); 422 assertEquals(0, p.getCompletedTaskCount()); 423 p.execute(new CheckedRunnable() { 424 public void realRun() throws InterruptedException { 425 threadStarted.countDown(); 426 assertEquals(0, p.getCompletedTaskCount()); 427 threadProceed.await(); 428 threadDone.countDown(); 429 }}); 430 await(threadStarted); 431 assertEquals(0, p.getCompletedTaskCount()); 432 threadProceed.countDown(); 433 threadDone.await(); 434 long startTime = System.nanoTime(); 435 while (p.getCompletedTaskCount() != 1) { 436 if (millisElapsedSince(startTime) > LONG_DELAY_MS) 437 fail("timed out"); 438 Thread.yield(); 439 } 440 } 441 } 442 443 /** 444 * getCorePoolSize returns size given in constructor if not otherwise set 445 */ testGetCorePoolSize()446 public void testGetCorePoolSize() { 447 final CustomExecutor p = new CustomExecutor(1); 448 try (PoolCleaner cleaner = cleaner(p)) { 449 assertEquals(1, p.getCorePoolSize()); 450 } 451 } 452 453 /** 454 * getLargestPoolSize increases, but doesn't overestimate, when 455 * multiple threads active 456 */ testGetLargestPoolSize()457 public void testGetLargestPoolSize() throws InterruptedException { 458 final int THREADS = 3; 459 final CountDownLatch done = new CountDownLatch(1); 460 final ThreadPoolExecutor p = new CustomExecutor(THREADS); 461 try (PoolCleaner cleaner = cleaner(p, done)) { 462 final CountDownLatch threadsStarted = new CountDownLatch(THREADS); 463 assertEquals(0, p.getLargestPoolSize()); 464 for (int i = 0; i < THREADS; i++) 465 p.execute(new CheckedRunnable() { 466 public void realRun() throws InterruptedException { 467 threadsStarted.countDown(); 468 await(done); 469 assertEquals(THREADS, p.getLargestPoolSize()); 470 }}); 471 await(threadsStarted); 472 assertEquals(THREADS, p.getLargestPoolSize()); 473 } 474 assertEquals(THREADS, p.getLargestPoolSize()); 475 } 476 477 /** 478 * getPoolSize increases, but doesn't overestimate, when threads 479 * become active 480 */ testGetPoolSize()481 public void testGetPoolSize() throws InterruptedException { 482 final CountDownLatch done = new CountDownLatch(1); 483 final ThreadPoolExecutor p = new CustomExecutor(1); 484 try (PoolCleaner cleaner = cleaner(p, done)) { 485 final CountDownLatch threadStarted = new CountDownLatch(1); 486 assertEquals(0, p.getPoolSize()); 487 p.execute(new CheckedRunnable() { 488 public void realRun() throws InterruptedException { 489 threadStarted.countDown(); 490 assertEquals(1, p.getPoolSize()); 491 await(done); 492 }}); 493 await(threadStarted); 494 assertEquals(1, p.getPoolSize()); 495 } 496 } 497 498 /** 499 * getTaskCount increases, but doesn't overestimate, when tasks 500 * submitted 501 */ testGetTaskCount()502 public void testGetTaskCount() throws InterruptedException { 503 final int TASKS = 3; 504 final CountDownLatch done = new CountDownLatch(1); 505 final ThreadPoolExecutor p = new CustomExecutor(1); 506 try (PoolCleaner cleaner = cleaner(p, done)) { 507 final CountDownLatch threadStarted = new CountDownLatch(1); 508 assertEquals(0, p.getTaskCount()); 509 assertEquals(0, p.getCompletedTaskCount()); 510 p.execute(new CheckedRunnable() { 511 public void realRun() throws InterruptedException { 512 threadStarted.countDown(); 513 await(done); 514 }}); 515 await(threadStarted); 516 assertEquals(1, p.getTaskCount()); 517 assertEquals(0, p.getCompletedTaskCount()); 518 for (int i = 0; i < TASKS; i++) { 519 assertEquals(1 + i, p.getTaskCount()); 520 p.execute(new CheckedRunnable() { 521 public void realRun() throws InterruptedException { 522 threadStarted.countDown(); 523 assertEquals(1 + TASKS, p.getTaskCount()); 524 await(done); 525 }}); 526 } 527 assertEquals(1 + TASKS, p.getTaskCount()); 528 assertEquals(0, p.getCompletedTaskCount()); 529 } 530 assertEquals(1 + TASKS, p.getTaskCount()); 531 assertEquals(1 + TASKS, p.getCompletedTaskCount()); 532 } 533 534 /** 535 * getThreadFactory returns factory in constructor if not set 536 */ testGetThreadFactory()537 public void testGetThreadFactory() { 538 final ThreadFactory threadFactory = new SimpleThreadFactory(); 539 final CustomExecutor p = new CustomExecutor(1, threadFactory); 540 try (PoolCleaner cleaner = cleaner(p)) { 541 assertSame(threadFactory, p.getThreadFactory()); 542 } 543 } 544 545 /** 546 * setThreadFactory sets the thread factory returned by getThreadFactory 547 */ testSetThreadFactory()548 public void testSetThreadFactory() { 549 final ThreadFactory threadFactory = new SimpleThreadFactory(); 550 final CustomExecutor p = new CustomExecutor(1); 551 try (PoolCleaner cleaner = cleaner(p)) { 552 p.setThreadFactory(threadFactory); 553 assertSame(threadFactory, p.getThreadFactory()); 554 } 555 } 556 557 /** 558 * setThreadFactory(null) throws NPE 559 */ testSetThreadFactoryNull()560 public void testSetThreadFactoryNull() { 561 final CustomExecutor p = new CustomExecutor(1); 562 try (PoolCleaner cleaner = cleaner(p)) { 563 try { 564 p.setThreadFactory(null); 565 shouldThrow(); 566 } catch (NullPointerException success) {} 567 } 568 } 569 570 /** 571 * isShutdown is false before shutdown, true after 572 */ testIsShutdown()573 public void testIsShutdown() { 574 final CustomExecutor p = new CustomExecutor(1); 575 try (PoolCleaner cleaner = cleaner(p)) { 576 assertFalse(p.isShutdown()); 577 try { p.shutdown(); } catch (SecurityException ok) { return; } 578 assertTrue(p.isShutdown()); 579 } 580 } 581 582 /** 583 * isTerminated is false before termination, true after 584 */ testIsTerminated()585 public void testIsTerminated() throws InterruptedException { 586 final CountDownLatch done = new CountDownLatch(1); 587 final ThreadPoolExecutor p = new CustomExecutor(1); 588 try (PoolCleaner cleaner = cleaner(p)) { 589 final CountDownLatch threadStarted = new CountDownLatch(1); 590 p.execute(new CheckedRunnable() { 591 public void realRun() throws InterruptedException { 592 assertFalse(p.isTerminated()); 593 threadStarted.countDown(); 594 await(done); 595 }}); 596 await(threadStarted); 597 assertFalse(p.isTerminated()); 598 assertFalse(p.isTerminating()); 599 done.countDown(); 600 try { p.shutdown(); } catch (SecurityException ok) { return; } 601 assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS)); 602 assertTrue(p.isTerminated()); 603 } 604 } 605 606 /** 607 * isTerminating is not true when running or when terminated 608 */ testIsTerminating()609 public void testIsTerminating() throws InterruptedException { 610 final CountDownLatch done = new CountDownLatch(1); 611 final ThreadPoolExecutor p = new CustomExecutor(1); 612 try (PoolCleaner cleaner = cleaner(p)) { 613 final CountDownLatch threadStarted = new CountDownLatch(1); 614 assertFalse(p.isTerminating()); 615 p.execute(new CheckedRunnable() { 616 public void realRun() throws InterruptedException { 617 assertFalse(p.isTerminating()); 618 threadStarted.countDown(); 619 await(done); 620 }}); 621 await(threadStarted); 622 assertFalse(p.isTerminating()); 623 done.countDown(); 624 try { p.shutdown(); } catch (SecurityException ok) { return; } 625 assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS)); 626 assertTrue(p.isTerminated()); 627 assertFalse(p.isTerminating()); 628 } 629 } 630 631 /** 632 * getQueue returns the work queue, which contains queued tasks 633 */ testGetQueue()634 public void testGetQueue() throws InterruptedException { 635 final CountDownLatch done = new CountDownLatch(1); 636 final ScheduledThreadPoolExecutor p = new CustomExecutor(1); 637 try (PoolCleaner cleaner = cleaner(p, done)) { 638 final CountDownLatch threadStarted = new CountDownLatch(1); 639 ScheduledFuture[] tasks = new ScheduledFuture[5]; 640 for (int i = 0; i < tasks.length; i++) { 641 Runnable r = new CheckedRunnable() { 642 public void realRun() throws InterruptedException { 643 threadStarted.countDown(); 644 await(done); 645 }}; 646 tasks[i] = p.schedule(r, 1, MILLISECONDS); 647 } 648 await(threadStarted); 649 BlockingQueue<Runnable> q = p.getQueue(); 650 assertTrue(q.contains(tasks[tasks.length - 1])); 651 assertFalse(q.contains(tasks[0])); 652 } 653 } 654 655 /** 656 * remove(task) removes queued task, and fails to remove active task 657 */ testRemove()658 public void testRemove() throws InterruptedException { 659 final CountDownLatch done = new CountDownLatch(1); 660 final ScheduledThreadPoolExecutor p = new CustomExecutor(1); 661 try (PoolCleaner cleaner = cleaner(p, done)) { 662 ScheduledFuture[] tasks = new ScheduledFuture[5]; 663 final CountDownLatch threadStarted = new CountDownLatch(1); 664 for (int i = 0; i < tasks.length; i++) { 665 Runnable r = new CheckedRunnable() { 666 public void realRun() throws InterruptedException { 667 threadStarted.countDown(); 668 await(done); 669 }}; 670 tasks[i] = p.schedule(r, 1, MILLISECONDS); 671 } 672 await(threadStarted); 673 BlockingQueue<Runnable> q = p.getQueue(); 674 assertFalse(p.remove((Runnable)tasks[0])); 675 assertTrue(q.contains((Runnable)tasks[4])); 676 assertTrue(q.contains((Runnable)tasks[3])); 677 assertTrue(p.remove((Runnable)tasks[4])); 678 assertFalse(p.remove((Runnable)tasks[4])); 679 assertFalse(q.contains((Runnable)tasks[4])); 680 assertTrue(q.contains((Runnable)tasks[3])); 681 assertTrue(p.remove((Runnable)tasks[3])); 682 assertFalse(q.contains((Runnable)tasks[3])); 683 } 684 } 685 686 /** 687 * purge removes cancelled tasks from the queue 688 */ testPurge()689 public void testPurge() throws InterruptedException { 690 final ScheduledFuture[] tasks = new ScheduledFuture[5]; 691 final Runnable releaser = new Runnable() { public void run() { 692 for (ScheduledFuture task : tasks) 693 if (task != null) task.cancel(true); }}; 694 final CustomExecutor p = new CustomExecutor(1); 695 try (PoolCleaner cleaner = cleaner(p, releaser)) { 696 for (int i = 0; i < tasks.length; i++) 697 tasks[i] = p.schedule(new SmallPossiblyInterruptedRunnable(), 698 LONG_DELAY_MS, MILLISECONDS); 699 int max = tasks.length; 700 if (tasks[4].cancel(true)) --max; 701 if (tasks[3].cancel(true)) --max; 702 // There must eventually be an interference-free point at 703 // which purge will not fail. (At worst, when queue is empty.) 704 long startTime = System.nanoTime(); 705 do { 706 p.purge(); 707 long count = p.getTaskCount(); 708 if (count == max) 709 return; 710 } while (millisElapsedSince(startTime) < LONG_DELAY_MS); 711 fail("Purge failed to remove cancelled tasks"); 712 } 713 } 714 715 /** 716 * shutdownNow returns a list containing tasks that were not run, 717 * and those tasks are drained from the queue 718 */ testShutdownNow()719 public void testShutdownNow() throws InterruptedException { 720 final int poolSize = 2; 721 final int count = 5; 722 final AtomicInteger ran = new AtomicInteger(0); 723 final CustomExecutor p = new CustomExecutor(poolSize); 724 final CountDownLatch threadsStarted = new CountDownLatch(poolSize); 725 Runnable waiter = new CheckedRunnable() { public void realRun() { 726 threadsStarted.countDown(); 727 try { 728 MILLISECONDS.sleep(2 * LONG_DELAY_MS); 729 } catch (InterruptedException success) {} 730 ran.getAndIncrement(); 731 }}; 732 for (int i = 0; i < count; i++) 733 p.execute(waiter); 734 await(threadsStarted); 735 assertEquals(poolSize, p.getActiveCount()); 736 assertEquals(0, p.getCompletedTaskCount()); 737 final List<Runnable> queuedTasks; 738 try { 739 queuedTasks = p.shutdownNow(); 740 } catch (SecurityException ok) { 741 return; // Allowed in case test doesn't have privs 742 } 743 assertTrue(p.isShutdown()); 744 assertTrue(p.getQueue().isEmpty()); 745 assertEquals(count - poolSize, queuedTasks.size()); 746 assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS)); 747 assertTrue(p.isTerminated()); 748 assertEquals(poolSize, ran.get()); 749 assertEquals(poolSize, p.getCompletedTaskCount()); 750 } 751 752 /** 753 * shutdownNow returns a list containing tasks that were not run, 754 * and those tasks are drained from the queue 755 */ testShutdownNow_delayedTasks()756 public void testShutdownNow_delayedTasks() throws InterruptedException { 757 final CustomExecutor p = new CustomExecutor(1); 758 List<ScheduledFuture> tasks = new ArrayList<>(); 759 for (int i = 0; i < 3; i++) { 760 Runnable r = new NoOpRunnable(); 761 tasks.add(p.schedule(r, 9, SECONDS)); 762 tasks.add(p.scheduleAtFixedRate(r, 9, 9, SECONDS)); 763 tasks.add(p.scheduleWithFixedDelay(r, 9, 9, SECONDS)); 764 } 765 if (testImplementationDetails) 766 assertEquals(new HashSet(tasks), new HashSet(p.getQueue())); 767 final List<Runnable> queuedTasks; 768 try { 769 queuedTasks = p.shutdownNow(); 770 } catch (SecurityException ok) { 771 return; // Allowed in case test doesn't have privs 772 } 773 assertTrue(p.isShutdown()); 774 assertTrue(p.getQueue().isEmpty()); 775 if (testImplementationDetails) 776 assertEquals(new HashSet(tasks), new HashSet(queuedTasks)); 777 assertEquals(tasks.size(), queuedTasks.size()); 778 for (ScheduledFuture task : tasks) { 779 assertFalse(((CustomTask)task).ran); 780 assertFalse(task.isDone()); 781 assertFalse(task.isCancelled()); 782 } 783 assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS)); 784 assertTrue(p.isTerminated()); 785 } 786 787 /** 788 * By default, periodic tasks are cancelled at shutdown. 789 * By default, delayed tasks keep running after shutdown. 790 * Check that changing the default values work: 791 * - setExecuteExistingDelayedTasksAfterShutdownPolicy 792 * - setContinueExistingPeriodicTasksAfterShutdownPolicy 793 */ testShutdown_cancellation()794 public void testShutdown_cancellation() throws Exception { 795 Boolean[] allBooleans = { null, Boolean.FALSE, Boolean.TRUE }; 796 for (Boolean policy : allBooleans) 797 { 798 final int poolSize = 2; 799 final CustomExecutor p = new CustomExecutor(poolSize); 800 final boolean effectiveDelayedPolicy = (policy != Boolean.FALSE); 801 final boolean effectivePeriodicPolicy = (policy == Boolean.TRUE); 802 final boolean effectiveRemovePolicy = (policy == Boolean.TRUE); 803 if (policy != null) { 804 p.setExecuteExistingDelayedTasksAfterShutdownPolicy(policy); 805 p.setContinueExistingPeriodicTasksAfterShutdownPolicy(policy); 806 p.setRemoveOnCancelPolicy(policy); 807 } 808 assertEquals(effectiveDelayedPolicy, 809 p.getExecuteExistingDelayedTasksAfterShutdownPolicy()); 810 assertEquals(effectivePeriodicPolicy, 811 p.getContinueExistingPeriodicTasksAfterShutdownPolicy()); 812 assertEquals(effectiveRemovePolicy, 813 p.getRemoveOnCancelPolicy()); 814 // Strategy: Wedge the pool with poolSize "blocker" threads 815 final AtomicInteger ran = new AtomicInteger(0); 816 final CountDownLatch poolBlocked = new CountDownLatch(poolSize); 817 final CountDownLatch unblock = new CountDownLatch(1); 818 final CountDownLatch periodicLatch1 = new CountDownLatch(2); 819 final CountDownLatch periodicLatch2 = new CountDownLatch(2); 820 Runnable task = new CheckedRunnable() { public void realRun() 821 throws InterruptedException { 822 poolBlocked.countDown(); 823 assertTrue(unblock.await(LONG_DELAY_MS, MILLISECONDS)); 824 ran.getAndIncrement(); 825 }}; 826 List<Future<?>> blockers = new ArrayList<>(); 827 List<Future<?>> periodics = new ArrayList<>(); 828 List<Future<?>> delayeds = new ArrayList<>(); 829 for (int i = 0; i < poolSize; i++) 830 blockers.add(p.submit(task)); 831 assertTrue(poolBlocked.await(LONG_DELAY_MS, MILLISECONDS)); 832 833 periodics.add(p.scheduleAtFixedRate(countDowner(periodicLatch1), 834 1, 1, MILLISECONDS)); 835 periodics.add(p.scheduleWithFixedDelay(countDowner(periodicLatch2), 836 1, 1, MILLISECONDS)); 837 delayeds.add(p.schedule(task, 1, MILLISECONDS)); 838 839 assertTrue(p.getQueue().containsAll(periodics)); 840 assertTrue(p.getQueue().containsAll(delayeds)); 841 try { p.shutdown(); } catch (SecurityException ok) { return; } 842 assertTrue(p.isShutdown()); 843 assertFalse(p.isTerminated()); 844 for (Future<?> periodic : periodics) { 845 assertTrue(effectivePeriodicPolicy ^ periodic.isCancelled()); 846 assertTrue(effectivePeriodicPolicy ^ periodic.isDone()); 847 } 848 for (Future<?> delayed : delayeds) { 849 assertTrue(effectiveDelayedPolicy ^ delayed.isCancelled()); 850 assertTrue(effectiveDelayedPolicy ^ delayed.isDone()); 851 } 852 if (testImplementationDetails) { 853 assertEquals(effectivePeriodicPolicy, 854 p.getQueue().containsAll(periodics)); 855 assertEquals(effectiveDelayedPolicy, 856 p.getQueue().containsAll(delayeds)); 857 } 858 // Release all pool threads 859 unblock.countDown(); 860 861 for (Future<?> delayed : delayeds) { 862 if (effectiveDelayedPolicy) { 863 assertNull(delayed.get()); 864 } 865 } 866 if (effectivePeriodicPolicy) { 867 assertTrue(periodicLatch1.await(LONG_DELAY_MS, MILLISECONDS)); 868 assertTrue(periodicLatch2.await(LONG_DELAY_MS, MILLISECONDS)); 869 for (Future<?> periodic : periodics) { 870 assertTrue(periodic.cancel(false)); 871 assertTrue(periodic.isCancelled()); 872 assertTrue(periodic.isDone()); 873 } 874 } 875 assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS)); 876 assertTrue(p.isTerminated()); 877 assertEquals(2 + (effectiveDelayedPolicy ? 1 : 0), ran.get()); 878 }} 879 880 /** 881 * completed submit of callable returns result 882 */ testSubmitCallable()883 public void testSubmitCallable() throws Exception { 884 final ExecutorService e = new CustomExecutor(2); 885 try (PoolCleaner cleaner = cleaner(e)) { 886 Future<String> future = e.submit(new StringTask()); 887 String result = future.get(); 888 assertSame(TEST_STRING, result); 889 } 890 } 891 892 /** 893 * completed submit of runnable returns successfully 894 */ testSubmitRunnable()895 public void testSubmitRunnable() throws Exception { 896 final ExecutorService e = new CustomExecutor(2); 897 try (PoolCleaner cleaner = cleaner(e)) { 898 Future<?> future = e.submit(new NoOpRunnable()); 899 future.get(); 900 assertTrue(future.isDone()); 901 } 902 } 903 904 /** 905 * completed submit of (runnable, result) returns result 906 */ testSubmitRunnable2()907 public void testSubmitRunnable2() throws Exception { 908 final ExecutorService e = new CustomExecutor(2); 909 try (PoolCleaner cleaner = cleaner(e)) { 910 Future<String> future = e.submit(new NoOpRunnable(), TEST_STRING); 911 String result = future.get(); 912 assertSame(TEST_STRING, result); 913 } 914 } 915 916 /** 917 * invokeAny(null) throws NPE 918 */ testInvokeAny1()919 public void testInvokeAny1() throws Exception { 920 final ExecutorService e = new CustomExecutor(2); 921 try (PoolCleaner cleaner = cleaner(e)) { 922 try { 923 e.invokeAny(null); 924 shouldThrow(); 925 } catch (NullPointerException success) {} 926 } 927 } 928 929 /** 930 * invokeAny(empty collection) throws IAE 931 */ testInvokeAny2()932 public void testInvokeAny2() throws Exception { 933 final ExecutorService e = new CustomExecutor(2); 934 try (PoolCleaner cleaner = cleaner(e)) { 935 try { 936 e.invokeAny(new ArrayList<Callable<String>>()); 937 shouldThrow(); 938 } catch (IllegalArgumentException success) {} 939 } 940 } 941 942 /** 943 * invokeAny(c) throws NPE if c has null elements 944 */ testInvokeAny3()945 public void testInvokeAny3() throws Exception { 946 final CountDownLatch latch = new CountDownLatch(1); 947 final ExecutorService e = new CustomExecutor(2); 948 try (PoolCleaner cleaner = cleaner(e)) { 949 List<Callable<String>> l = new ArrayList<Callable<String>>(); 950 l.add(latchAwaitingStringTask(latch)); 951 l.add(null); 952 try { 953 e.invokeAny(l); 954 shouldThrow(); 955 } catch (NullPointerException success) {} 956 latch.countDown(); 957 } 958 } 959 960 /** 961 * invokeAny(c) throws ExecutionException if no task completes 962 */ testInvokeAny4()963 public void testInvokeAny4() throws Exception { 964 final ExecutorService e = new CustomExecutor(2); 965 try (PoolCleaner cleaner = cleaner(e)) { 966 List<Callable<String>> l = new ArrayList<Callable<String>>(); 967 l.add(new NPETask()); 968 try { 969 e.invokeAny(l); 970 shouldThrow(); 971 } catch (ExecutionException success) { 972 assertTrue(success.getCause() instanceof NullPointerException); 973 } 974 } 975 } 976 977 /** 978 * invokeAny(c) returns result of some task 979 */ testInvokeAny5()980 public void testInvokeAny5() throws Exception { 981 final ExecutorService e = new CustomExecutor(2); 982 try (PoolCleaner cleaner = cleaner(e)) { 983 List<Callable<String>> l = new ArrayList<Callable<String>>(); 984 l.add(new StringTask()); 985 l.add(new StringTask()); 986 String result = e.invokeAny(l); 987 assertSame(TEST_STRING, result); 988 } 989 } 990 991 /** 992 * invokeAll(null) throws NPE 993 */ testInvokeAll1()994 public void testInvokeAll1() throws Exception { 995 final ExecutorService e = new CustomExecutor(2); 996 try (PoolCleaner cleaner = cleaner(e)) { 997 try { 998 e.invokeAll(null); 999 shouldThrow(); 1000 } catch (NullPointerException success) {} 1001 } 1002 } 1003 1004 /** 1005 * invokeAll(empty collection) returns empty collection 1006 */ testInvokeAll2()1007 public void testInvokeAll2() throws Exception { 1008 final ExecutorService e = new CustomExecutor(2); 1009 try (PoolCleaner cleaner = cleaner(e)) { 1010 List<Future<String>> r = e.invokeAll(new ArrayList<Callable<String>>()); 1011 assertTrue(r.isEmpty()); 1012 } 1013 } 1014 1015 /** 1016 * invokeAll(c) throws NPE if c has null elements 1017 */ testInvokeAll3()1018 public void testInvokeAll3() throws Exception { 1019 final ExecutorService e = new CustomExecutor(2); 1020 try (PoolCleaner cleaner = cleaner(e)) { 1021 List<Callable<String>> l = new ArrayList<Callable<String>>(); 1022 l.add(new StringTask()); 1023 l.add(null); 1024 try { 1025 e.invokeAll(l); 1026 shouldThrow(); 1027 } catch (NullPointerException success) {} 1028 } 1029 } 1030 1031 /** 1032 * get of invokeAll(c) throws exception on failed task 1033 */ testInvokeAll4()1034 public void testInvokeAll4() throws Exception { 1035 final ExecutorService e = new CustomExecutor(2); 1036 try (PoolCleaner cleaner = cleaner(e)) { 1037 List<Callable<String>> l = new ArrayList<Callable<String>>(); 1038 l.add(new NPETask()); 1039 List<Future<String>> futures = e.invokeAll(l); 1040 assertEquals(1, futures.size()); 1041 try { 1042 futures.get(0).get(); 1043 shouldThrow(); 1044 } catch (ExecutionException success) { 1045 assertTrue(success.getCause() instanceof NullPointerException); 1046 } 1047 } 1048 } 1049 1050 /** 1051 * invokeAll(c) returns results of all completed tasks 1052 */ testInvokeAll5()1053 public void testInvokeAll5() throws Exception { 1054 final ExecutorService e = new CustomExecutor(2); 1055 try (PoolCleaner cleaner = cleaner(e)) { 1056 List<Callable<String>> l = new ArrayList<Callable<String>>(); 1057 l.add(new StringTask()); 1058 l.add(new StringTask()); 1059 List<Future<String>> futures = e.invokeAll(l); 1060 assertEquals(2, futures.size()); 1061 for (Future<String> future : futures) 1062 assertSame(TEST_STRING, future.get()); 1063 } 1064 } 1065 1066 /** 1067 * timed invokeAny(null) throws NPE 1068 */ testTimedInvokeAny1()1069 public void testTimedInvokeAny1() throws Exception { 1070 final ExecutorService e = new CustomExecutor(2); 1071 try (PoolCleaner cleaner = cleaner(e)) { 1072 try { 1073 e.invokeAny(null, MEDIUM_DELAY_MS, MILLISECONDS); 1074 shouldThrow(); 1075 } catch (NullPointerException success) {} 1076 } 1077 } 1078 1079 /** 1080 * timed invokeAny(,,null) throws NPE 1081 */ testTimedInvokeAnyNullTimeUnit()1082 public void testTimedInvokeAnyNullTimeUnit() throws Exception { 1083 final ExecutorService e = new CustomExecutor(2); 1084 try (PoolCleaner cleaner = cleaner(e)) { 1085 List<Callable<String>> l = new ArrayList<Callable<String>>(); 1086 l.add(new StringTask()); 1087 try { 1088 e.invokeAny(l, MEDIUM_DELAY_MS, null); 1089 shouldThrow(); 1090 } catch (NullPointerException success) {} 1091 } 1092 } 1093 1094 /** 1095 * timed invokeAny(empty collection) throws IAE 1096 */ testTimedInvokeAny2()1097 public void testTimedInvokeAny2() throws Exception { 1098 final ExecutorService e = new CustomExecutor(2); 1099 try (PoolCleaner cleaner = cleaner(e)) { 1100 try { 1101 e.invokeAny(new ArrayList<Callable<String>>(), MEDIUM_DELAY_MS, MILLISECONDS); 1102 shouldThrow(); 1103 } catch (IllegalArgumentException success) {} 1104 } 1105 } 1106 1107 /** 1108 * timed invokeAny(c) throws NPE if c has null elements 1109 */ testTimedInvokeAny3()1110 public void testTimedInvokeAny3() throws Exception { 1111 CountDownLatch latch = new CountDownLatch(1); 1112 final ExecutorService e = new CustomExecutor(2); 1113 try (PoolCleaner cleaner = cleaner(e)) { 1114 List<Callable<String>> l = new ArrayList<Callable<String>>(); 1115 l.add(latchAwaitingStringTask(latch)); 1116 l.add(null); 1117 try { 1118 e.invokeAny(l, MEDIUM_DELAY_MS, MILLISECONDS); 1119 shouldThrow(); 1120 } catch (NullPointerException success) {} 1121 latch.countDown(); 1122 } 1123 } 1124 1125 /** 1126 * timed invokeAny(c) throws ExecutionException if no task completes 1127 */ testTimedInvokeAny4()1128 public void testTimedInvokeAny4() throws Exception { 1129 final ExecutorService e = new CustomExecutor(2); 1130 try (PoolCleaner cleaner = cleaner(e)) { 1131 long startTime = System.nanoTime(); 1132 List<Callable<String>> l = new ArrayList<Callable<String>>(); 1133 l.add(new NPETask()); 1134 try { 1135 e.invokeAny(l, LONG_DELAY_MS, MILLISECONDS); 1136 shouldThrow(); 1137 } catch (ExecutionException success) { 1138 assertTrue(success.getCause() instanceof NullPointerException); 1139 } 1140 assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS); 1141 } 1142 } 1143 1144 /** 1145 * timed invokeAny(c) returns result of some task 1146 */ 1147 public void testTimedInvokeAny5() throws Exception { 1148 final ExecutorService e = new CustomExecutor(2); 1149 try (PoolCleaner cleaner = cleaner(e)) { 1150 long startTime = System.nanoTime(); 1151 List<Callable<String>> l = new ArrayList<Callable<String>>(); 1152 l.add(new StringTask()); 1153 l.add(new StringTask()); 1154 String result = e.invokeAny(l, LONG_DELAY_MS, MILLISECONDS); 1155 assertSame(TEST_STRING, result); 1156 assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS); 1157 } 1158 } 1159 1160 /** 1161 * timed invokeAll(null) throws NPE 1162 */ 1163 public void testTimedInvokeAll1() throws Exception { 1164 final ExecutorService e = new CustomExecutor(2); 1165 try (PoolCleaner cleaner = cleaner(e)) { 1166 try { 1167 e.invokeAll(null, MEDIUM_DELAY_MS, MILLISECONDS); 1168 shouldThrow(); 1169 } catch (NullPointerException success) {} 1170 } 1171 } 1172 1173 /** 1174 * timed invokeAll(,,null) throws NPE 1175 */ 1176 public void testTimedInvokeAllNullTimeUnit() throws Exception { 1177 final ExecutorService e = new CustomExecutor(2); 1178 try (PoolCleaner cleaner = cleaner(e)) { 1179 List<Callable<String>> l = new ArrayList<Callable<String>>(); 1180 l.add(new StringTask()); 1181 try { 1182 e.invokeAll(l, MEDIUM_DELAY_MS, null); 1183 shouldThrow(); 1184 } catch (NullPointerException success) {} 1185 } 1186 } 1187 1188 /** 1189 * timed invokeAll(empty collection) returns empty collection 1190 */ 1191 public void testTimedInvokeAll2() throws Exception { 1192 final ExecutorService e = new CustomExecutor(2); 1193 try (PoolCleaner cleaner = cleaner(e)) { 1194 List<Future<String>> r = e.invokeAll(new ArrayList<Callable<String>>(), MEDIUM_DELAY_MS, MILLISECONDS); 1195 assertTrue(r.isEmpty()); 1196 } 1197 } 1198 1199 /** 1200 * timed invokeAll(c) throws NPE if c has null elements 1201 */ 1202 public void testTimedInvokeAll3() throws Exception { 1203 final ExecutorService e = new CustomExecutor(2); 1204 try (PoolCleaner cleaner = cleaner(e)) { 1205 List<Callable<String>> l = new ArrayList<Callable<String>>(); 1206 l.add(new StringTask()); 1207 l.add(null); 1208 try { 1209 e.invokeAll(l, MEDIUM_DELAY_MS, MILLISECONDS); 1210 shouldThrow(); 1211 } catch (NullPointerException success) {} 1212 } 1213 } 1214 1215 /** 1216 * get of element of invokeAll(c) throws exception on failed task 1217 */ 1218 public void testTimedInvokeAll4() throws Exception { 1219 final ExecutorService e = new CustomExecutor(2); 1220 try (PoolCleaner cleaner = cleaner(e)) { 1221 List<Callable<String>> l = new ArrayList<Callable<String>>(); 1222 l.add(new NPETask()); 1223 List<Future<String>> futures = 1224 e.invokeAll(l, MEDIUM_DELAY_MS, MILLISECONDS); 1225 assertEquals(1, futures.size()); 1226 try { 1227 futures.get(0).get(); 1228 shouldThrow(); 1229 } catch (ExecutionException success) { 1230 assertTrue(success.getCause() instanceof NullPointerException); 1231 } 1232 } 1233 } 1234 1235 /** 1236 * timed invokeAll(c) returns results of all completed tasks 1237 */ 1238 public void testTimedInvokeAll5() throws Exception { 1239 final ExecutorService e = new CustomExecutor(2); 1240 try (PoolCleaner cleaner = cleaner(e)) { 1241 List<Callable<String>> l = new ArrayList<Callable<String>>(); 1242 l.add(new StringTask()); 1243 l.add(new StringTask()); 1244 List<Future<String>> futures = 1245 e.invokeAll(l, LONG_DELAY_MS, MILLISECONDS); 1246 assertEquals(2, futures.size()); 1247 for (Future<String> future : futures) 1248 assertSame(TEST_STRING, future.get()); 1249 } 1250 } 1251 1252 /** 1253 * timed invokeAll(c) cancels tasks not completed by timeout 1254 */ 1255 public void testTimedInvokeAll6() throws Exception { 1256 for (long timeout = timeoutMillis();;) { 1257 final CountDownLatch done = new CountDownLatch(1); 1258 final Callable<String> waiter = new CheckedCallable<String>() { 1259 public String realCall() { 1260 try { done.await(LONG_DELAY_MS, MILLISECONDS); } 1261 catch (InterruptedException ok) {} 1262 return "1"; }}; 1263 final ExecutorService p = new CustomExecutor(2); 1264 try (PoolCleaner cleaner = cleaner(p, done)) { 1265 List<Callable<String>> tasks = new ArrayList<>(); 1266 tasks.add(new StringTask("0")); 1267 tasks.add(waiter); 1268 tasks.add(new StringTask("2")); 1269 long startTime = System.nanoTime(); 1270 List<Future<String>> futures = 1271 p.invokeAll(tasks, timeout, MILLISECONDS); 1272 assertEquals(tasks.size(), futures.size()); 1273 assertTrue(millisElapsedSince(startTime) >= timeout); 1274 for (Future future : futures) 1275 assertTrue(future.isDone()); 1276 assertTrue(futures.get(1).isCancelled()); 1277 try { 1278 assertEquals("0", futures.get(0).get()); 1279 assertEquals("2", futures.get(2).get()); 1280 break; 1281 } catch (CancellationException retryWithLongerTimeout) { 1282 timeout *= 2; 1283 if (timeout >= LONG_DELAY_MS / 2) 1284 fail("expected exactly one task to be cancelled"); 1285 } 1286 } 1287 } 1288 } 1289 1290 } 1291