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