| /* | |
| * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. | |
| * | |
| * This code is free software; you can redistribute it and/or modify it | |
| * under the terms of the GNU General Public License version 2 only, as | |
| * published by the Free Software Foundation. | |
| * | |
| * This code is distributed in the hope that it will be useful, but WITHOUT | |
| * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or | |
| * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License | |
| * version 2 for more details (a copy is included in the LICENSE file that | |
| * accompanied this code). | |
| * | |
| * You should have received a copy of the GNU General Public License version | |
| * 2 along with this work; if not, write to the Free Software Foundation, | |
| * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. | |
| * | |
| * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA | |
| * or visit www.oracle.com if you need additional information or have any | |
| * questions. | |
| */ | |
| /* | |
| * This file is available under and governed by the GNU General Public | |
| * License version 2 only, as published by the Free Software Foundation. | |
| * However, the following notice accompanied the original version of this | |
| * file: | |
| * | |
| * Written by Doug Lea with assistance from members of JCP JSR-166 | |
| * Expert Group and released to the public domain, as explained at | |
| * http://creativecommons.org/publicdomain/zero/1.0/ | |
| */ | |
| import java.util.ArrayList; | |
| import java.util.Collection; | |
| import java.util.Collections; | |
| import java.util.HashSet; | |
| import java.util.List; | |
| import java.util.concurrent.BlockingQueue; | |
| import java.util.concurrent.Callable; | |
| import java.util.concurrent.CancellationException; | |
| import java.util.concurrent.CountDownLatch; | |
| import java.util.concurrent.ExecutionException; | |
| import java.util.concurrent.ExecutorService; | |
| import java.util.concurrent.ForkJoinPool; | |
| import java.util.concurrent.ForkJoinTask; | |
| import java.util.concurrent.ForkJoinWorkerThread; | |
| import java.util.concurrent.Future; | |
| import java.util.concurrent.RejectedExecutionException; | |
| import java.util.concurrent.ScheduledExecutorService; | |
| import java.util.concurrent.ScheduledFuture; | |
| import java.util.concurrent.ThreadFactory; | |
| import java.util.concurrent.ThreadLocalRandom; | |
| import java.util.concurrent.atomic.AtomicBoolean; | |
| import java.util.concurrent.atomic.AtomicInteger; | |
| import java.util.concurrent.atomic.AtomicLong; | |
| import java.util.concurrent.atomic.AtomicReference; | |
| import java.util.concurrent.locks.LockSupport; | |
| import java.util.stream.Stream; | |
| import static java.util.concurrent.TimeUnit.MILLISECONDS; | |
| import static java.util.concurrent.TimeUnit.NANOSECONDS; | |
| import static java.util.concurrent.TimeUnit.SECONDS; | |
| import junit.framework.Test; | |
| import junit.framework.TestSuite; | |
| /** | |
| * Tests for ForkJoinPool and ForkJoinWorkerThread additions in JDK 20. | |
| */ | |
| public class ForkJoinPool20Test extends JSR166TestCase { | |
| public static void main(String[] args) { | |
| main(suite(), args); | |
| } | |
| public static Test suite() { | |
| return new TestSuite(ForkJoinPool20Test.class); | |
| } | |
| /** | |
| * Test that tasks submitted with externalSubmit execute. | |
| */ | |
| public void testExternalSubmit1() throws Exception { | |
| try (var pool = new ForkJoinPool()) { | |
| // submit from external client | |
| var task1 = ForkJoinTask.adapt(() -> "foo"); | |
| pool.externalSubmit(task1); | |
| assertEquals(task1.get(), "foo"); | |
| // submit from worker thread | |
| Future<Future<String>> task2 = pool.submit(() -> { | |
| return pool.externalSubmit(ForkJoinTask.adapt(() -> "foo")); | |
| }); | |
| assertEquals(task2.get().get(), "foo"); | |
| } | |
| } | |
| /** | |
| * Test that tasks submitted with externalSubmit are pushed to a submission queue. | |
| */ | |
| public void testExternalSubmit2() throws Exception { | |
| try (var pool = new ForkJoinPool(1)) { | |
| pool.submit(() -> { | |
| assertTrue(pool.getQueuedTaskCount() == 0); | |
| assertTrue(pool.getQueuedSubmissionCount() == 0); | |
| for (int count = 1; count <= 3; count++) { | |
| var task = ForkJoinTask.adapt(() -> { }); | |
| pool.externalSubmit(task); | |
| assertTrue(pool.getQueuedTaskCount() == 0); | |
| assertTrue(pool.getQueuedSubmissionCount() == count); | |
| } | |
| }).get(); | |
| } | |
| } | |
| /** | |
| * Test externalSubmit return value. | |
| */ | |
| public void testExternalSubmitReturnsTask() { | |
| try (var pool = new ForkJoinPool()) { | |
| var task = ForkJoinTask.adapt(() -> "foo"); | |
| assertTrue(pool.externalSubmit(task) == task); | |
| } | |
| } | |
| /** | |
| * Test externalSubmit(null) throws NullPointerException. | |
| */ | |
| public void testExternalSubmitWithNull() { | |
| try (var pool = new ForkJoinPool()) { | |
| assertThrows(NullPointerException.class, () -> pool.externalSubmit(null)); | |
| } | |
| } | |
| /** | |
| * Test externalSubmit throws RejectedExecutionException when pool is shutdown. | |
| */ | |
| public void testExternalSubmitWhenShutdown() { | |
| try (var pool = new ForkJoinPool()) { | |
| pool.shutdown(); | |
| var task = ForkJoinTask.adapt(() -> { }); | |
| assertThrows(RejectedExecutionException.class, () -> pool.externalSubmit(task)); | |
| } | |
| } | |
| /** | |
| * Test that tasks submitted with submit(ForkJoinTask) are pushed to a | |
| * submission queue. | |
| */ | |
| public void testSubmit() throws Exception { | |
| try (var pool = new ForkJoinPool(1)) { | |
| ForkJoinWorkerThread worker = submitBusyTask(pool); | |
| try { | |
| assertTrue(worker.getQueuedTaskCount() == 0); | |
| assertTrue(pool.getQueuedTaskCount() == 0); | |
| assertTrue(pool.getQueuedSubmissionCount() == 0); | |
| for (int count = 1; count <= 3; count++) { | |
| var task = ForkJoinTask.adapt(() -> { }); | |
| pool.submit(task); | |
| // task should be in submission queue | |
| assertTrue(worker.getQueuedTaskCount() == 0); | |
| assertTrue(pool.getQueuedTaskCount() == 0); | |
| assertTrue(pool.getQueuedSubmissionCount() == count); | |
| } | |
| } finally { | |
| LockSupport.unpark(worker); | |
| } | |
| } | |
| } | |
| /** | |
| * Test ForkJoinWorkerThread::getQueuedTaskCount returns the number of tasks in the | |
| * current thread's queue. This test runs with parallelism of 1 to ensure that tasks | |
| * aren't stolen. | |
| */ | |
| public void testGetQueuedTaskCount1() throws Exception { | |
| try (var pool = new ForkJoinPool(1)) { | |
| pool.submit(() -> { | |
| var worker = (ForkJoinWorkerThread) Thread.currentThread(); | |
| assertTrue(worker.getQueuedTaskCount() == 0); | |
| for (int count = 1; count <= 3; count++) { | |
| pool.submit(() -> { }); | |
| // task should be in this thread's task queue | |
| assertTrue(worker.getQueuedTaskCount() == count); | |
| assertTrue(pool.getQueuedTaskCount() == count); | |
| assertTrue(pool.getQueuedSubmissionCount() == 0); | |
| } | |
| }).get(); | |
| } | |
| } | |
| /** | |
| * Test ForkJoinWorkerThread::getQueuedTaskCount returns the number of tasks in the | |
| * thread's queue. This test runs with parallelism of 2 and one worker active running | |
| * a task. This gives the test two task queues to sample. | |
| */ | |
| public void testGetQueuedTaskCount2() throws Exception { | |
| try (var pool = new ForkJoinPool(2)) { | |
| // keep one worker thread active | |
| ForkJoinWorkerThread worker1 = submitBusyTask(pool); | |
| try { | |
| pool.submit(() -> { | |
| var worker2 = (ForkJoinWorkerThread) Thread.currentThread(); | |
| for (int count = 1; count <= 3; count++) { | |
| pool.submit(() -> { }); | |
| // task should be in this thread's task queue | |
| assertTrue(worker1.getQueuedTaskCount() == 0); | |
| assertTrue(worker2.getQueuedTaskCount() == count); | |
| assertTrue(pool.getQueuedTaskCount() == count); | |
| assertTrue(pool.getQueuedSubmissionCount() == 0); | |
| } | |
| }).get(); | |
| } finally { | |
| LockSupport.unpark(worker1); // release worker1 | |
| } | |
| } | |
| } | |
| /** | |
| * Submits a task to the pool, returning the worker thread that runs the | |
| * task. The task runs until the thread is unparked. | |
| */ | |
| static ForkJoinWorkerThread submitBusyTask(ForkJoinPool pool) throws Exception { | |
| var ref = new AtomicReference<ForkJoinWorkerThread>(); | |
| pool.submit(() -> { | |
| ref.set((ForkJoinWorkerThread) Thread.currentThread()); | |
| LockSupport.park(); | |
| }); | |
| ForkJoinWorkerThread worker; | |
| while ((worker = ref.get()) == null) { | |
| Thread.sleep(20); | |
| } | |
| return worker; | |
| } | |
| // additions for ScheduledExecutorService | |
| /** | |
| * delayed schedule of callable successfully executes after delay | |
| */ | |
| public void testSchedule1() throws Exception { | |
| final ForkJoinPool p = new ForkJoinPool(2); | |
| try (PoolCleaner cleaner = cleaner(p)) { | |
| final long startTime = System.nanoTime(); | |
| final CountDownLatch done = new CountDownLatch(1); | |
| Callable<Boolean> task = new CheckedCallable<>() { | |
| public Boolean realCall() { | |
| done.countDown(); | |
| assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); | |
| return Boolean.TRUE; | |
| }}; | |
| Future<Boolean> f = p.schedule(task, timeoutMillis(), MILLISECONDS); | |
| assertSame(Boolean.TRUE, f.get()); | |
| assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); | |
| assertEquals(0L, done.getCount()); | |
| } | |
| } | |
| /** | |
| * delayed schedule of callable successfully executes after delay | |
| * even if shutdown. | |
| */ | |
| public void testSchedule1b() throws Exception { | |
| final ForkJoinPool p = new ForkJoinPool(2); | |
| try (PoolCleaner cleaner = cleaner(p)) { | |
| final long startTime = System.nanoTime(); | |
| final CountDownLatch done = new CountDownLatch(1); | |
| Callable<Boolean> task = new CheckedCallable<>() { | |
| public Boolean realCall() { | |
| done.countDown(); | |
| assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); | |
| return Boolean.TRUE; | |
| }}; | |
| Future<Boolean> f = p.schedule(task, timeoutMillis(), MILLISECONDS); | |
| p.shutdown(); | |
| assertSame(Boolean.TRUE, f.get()); | |
| assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); | |
| assertEquals(0L, done.getCount()); | |
| } | |
| } | |
| /** | |
| * delayed schedule of runnable successfully executes after delay | |
| */ | |
| public void testSchedule3() throws Exception { | |
| final ForkJoinPool p = new ForkJoinPool(2); | |
| try (PoolCleaner cleaner = cleaner(p)) { | |
| final long startTime = System.nanoTime(); | |
| final CountDownLatch done = new CountDownLatch(1); | |
| Runnable task = new CheckedRunnable() { | |
| public void realRun() { | |
| done.countDown(); | |
| assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); | |
| }}; | |
| Future<?> f = p.schedule(task, timeoutMillis(), MILLISECONDS); | |
| await(done); | |
| assertNull(f.get(LONG_DELAY_MS, MILLISECONDS)); | |
| assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); | |
| } | |
| } | |
| /** | |
| * scheduleAtFixedRate executes runnable after given initial delay | |
| */ | |
| public void testSchedule4() throws Exception { | |
| final ForkJoinPool p = new ForkJoinPool(2); | |
| try (PoolCleaner cleaner = cleaner(p)) { | |
| final long startTime = System.nanoTime(); | |
| final CountDownLatch done = new CountDownLatch(1); | |
| Runnable task = new CheckedRunnable() { | |
| public void realRun() { | |
| done.countDown(); | |
| assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); | |
| }}; | |
| ScheduledFuture<?> f = | |
| p.scheduleAtFixedRate(task, timeoutMillis(), | |
| LONG_DELAY_MS, MILLISECONDS); | |
| await(done); | |
| assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); | |
| f.cancel(true); | |
| } | |
| } | |
| /** | |
| * scheduleAtFixedRate with 0 initial delay re-rexecutes | |
| */ | |
| public void testSchedule4a() throws Exception { | |
| final ForkJoinPool p = new ForkJoinPool(2); | |
| try (PoolCleaner cleaner = cleaner(p)) { | |
| final long startTime = System.nanoTime(); | |
| final CountDownLatch done = new CountDownLatch(2); | |
| Runnable task = new Runnable() { | |
| public void run() { | |
| done.countDown(); | |
| }}; | |
| ScheduledFuture<?> f = | |
| p.scheduleAtFixedRate(task, 0L, timeoutMillis(), | |
| MILLISECONDS); | |
| await(done); | |
| f.cancel(true); | |
| assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); | |
| } | |
| } | |
| /** | |
| * scheduleWithFixedDelay executes runnable after given initial delay | |
| */ | |
| public void testSchedule5() throws Exception { | |
| final ForkJoinPool p = new ForkJoinPool(2); | |
| try (PoolCleaner cleaner = cleaner(p)) { | |
| final long startTime = System.nanoTime(); | |
| final CountDownLatch done = new CountDownLatch(1); | |
| Runnable task = new CheckedRunnable() { | |
| public void realRun() { | |
| done.countDown(); | |
| assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); | |
| }}; | |
| ScheduledFuture<?> f = | |
| p.scheduleWithFixedDelay(task, timeoutMillis(), | |
| LONG_DELAY_MS, MILLISECONDS); | |
| await(done); | |
| assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); | |
| f.cancel(true); | |
| } | |
| } | |
| /** | |
| * scheduleWithFixedDelay with 0 initial delay re-rexecutes | |
| */ | |
| public void testSchedule5a() throws Exception { | |
| final ForkJoinPool p = new ForkJoinPool(2); | |
| try (PoolCleaner cleaner = cleaner(p)) { | |
| final long startTime = System.nanoTime(); | |
| final CountDownLatch done = new CountDownLatch(2); | |
| Runnable task = new Runnable() { | |
| public void run() { | |
| done.countDown(); | |
| }}; | |
| ScheduledFuture<?> f = | |
| p.scheduleWithFixedDelay(task, 0L, timeoutMillis(), | |
| MILLISECONDS); | |
| await(done); | |
| assertTrue(millisElapsedSince(startTime) >= timeoutMillis()); | |
| f.cancel(true); | |
| } | |
| } | |
| static class RunnableCounter implements Runnable { | |
| AtomicInteger count = new AtomicInteger(0); | |
| public void run() { count.getAndIncrement(); } | |
| } | |
| /** | |
| * scheduleAtFixedRate executes series of tasks at given rate. | |
| * Eventually, it must hold that: | |
| * cycles - 1 <= elapsedMillis/delay < cycles | |
| * Additionally, periodic tasks are not run after shutdown. | |
| */ | |
| public void testFixedRateSequence() throws InterruptedException { | |
| final ForkJoinPool p = new ForkJoinPool(4); | |
| try (PoolCleaner cleaner = cleaner(p)) { | |
| for (int delay = 1; delay <= LONG_DELAY_MS; delay *= 3) { | |
| final long startTime = System.nanoTime(); | |
| final int cycles = 8; | |
| final CountDownLatch done = new CountDownLatch(cycles); | |
| final Runnable task = new CheckedRunnable() { | |
| public void realRun() { done.countDown(); }}; | |
| final ScheduledFuture<?> periodicTask = | |
| p.scheduleAtFixedRate(task, 0, delay, MILLISECONDS); | |
| final int totalDelayMillis = (cycles - 1) * delay; | |
| await(done, totalDelayMillis + LONG_DELAY_MS); | |
| final long elapsedMillis = millisElapsedSince(startTime); | |
| assertTrue(elapsedMillis >= totalDelayMillis); | |
| if (elapsedMillis <= cycles * delay) | |
| return; | |
| periodicTask.cancel(true); // retry with longer delay | |
| } | |
| fail("unexpected execution rate"); | |
| } | |
| } | |
| /** | |
| * scheduleWithFixedDelay executes series of tasks with given | |
| * period. Eventually, it must hold that each task starts at | |
| * least delay and at most 2 * delay after the termination of the | |
| * previous task. Additionally, periodic tasks are not run after | |
| * shutdown. | |
| */ | |
| public void testFixedDelaySequence() throws InterruptedException { | |
| final ForkJoinPool p = new ForkJoinPool(1); | |
| try (PoolCleaner cleaner = cleaner(p)) { | |
| for (int delay = 1; delay <= LONG_DELAY_MS; delay *= 3) { | |
| final long startTime = System.nanoTime(); | |
| final AtomicLong previous = new AtomicLong(startTime); | |
| final AtomicBoolean tryLongerDelay = new AtomicBoolean(false); | |
| final int cycles = 8; | |
| final CountDownLatch done = new CountDownLatch(cycles); | |
| final int d = delay; | |
| final Runnable task = new CheckedRunnable() { | |
| public void realRun() { | |
| long now = System.nanoTime(); | |
| long elapsedMillis | |
| = NANOSECONDS.toMillis(now - previous.get()); | |
| if (elapsedMillis >= (done.getCount() == cycles ? d : 2 * d)) | |
| tryLongerDelay.set(true); | |
| previous.set(now); | |
| done.countDown(); | |
| }}; | |
| final ScheduledFuture<?> periodicTask = | |
| p.scheduleWithFixedDelay(task, 0, delay, MILLISECONDS); | |
| final int totalDelayMillis = (cycles - 1) * delay; | |
| await(done, totalDelayMillis + cycles * LONG_DELAY_MS); | |
| final long elapsedMillis = millisElapsedSince(startTime); | |
| assertTrue(elapsedMillis >= totalDelayMillis); | |
| if (!tryLongerDelay.get()) | |
| return; | |
| periodicTask.cancel(true); // retry with longer delay | |
| } | |
| fail("unexpected execution rate"); | |
| } | |
| } | |
| /** | |
| * Submitting null tasks throws NullPointerException | |
| */ | |
| public void testNullTaskSubmission() { | |
| final ForkJoinPool p = new ForkJoinPool(1); | |
| try (PoolCleaner cleaner = cleaner(p)) { | |
| assertNullTaskSubmissionThrowsNullPointerException(p); | |
| } | |
| } | |
| /** | |
| * Submitted tasks are rejected when shutdown | |
| */ | |
| public void testSubmittedTasksRejectedWhenShutdown() throws InterruptedException { | |
| final ForkJoinPool p = new ForkJoinPool(4); | |
| final ThreadLocalRandom rnd = ThreadLocalRandom.current(); | |
| final CountDownLatch threadsStarted = new CountDownLatch(p.getParallelism()); | |
| final CountDownLatch done = new CountDownLatch(1); | |
| final Runnable r = () -> { | |
| threadsStarted.countDown(); | |
| for (;;) { | |
| try { | |
| done.await(); | |
| return; | |
| } catch (InterruptedException shutdownNowDeliberatelyIgnored) {} | |
| }}; | |
| final Callable<Boolean> c = () -> { | |
| threadsStarted.countDown(); | |
| for (;;) { | |
| try { | |
| done.await(); | |
| return Boolean.TRUE; | |
| } catch (InterruptedException shutdownNowDeliberatelyIgnored) {} | |
| }}; | |
| try (PoolCleaner cleaner = cleaner(p, done)) { | |
| for (int i = p.getParallelism(); i-- > 0; ) { | |
| switch (rnd.nextInt(4)) { | |
| case 0: p.execute(r); break; | |
| case 1: assertFalse(p.submit(r).isDone()); break; | |
| case 2: assertFalse(p.submit(r, Boolean.TRUE).isDone()); break; | |
| case 3: assertFalse(p.submit(c).isDone()); break; | |
| } | |
| } | |
| await(threadsStarted); | |
| p.shutdown(); | |
| done.countDown(); // release blocking tasks | |
| assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS)); | |
| } | |
| } | |
| /** | |
| * A fixed delay task with overflowing period should not prevent a | |
| * one-shot task from executing. | |
| * https://bugs.openjdk.org/browse/JDK-8051859 | |
| */ | |
| @SuppressWarnings("FutureReturnValueIgnored") | |
| public void testScheduleWithFixedDelay_overflow() throws Exception { | |
| final CountDownLatch delayedDone = new CountDownLatch(1); | |
| final CountDownLatch immediateDone = new CountDownLatch(1); | |
| final ForkJoinPool p = new ForkJoinPool(2); | |
| p.cancelDelayedTasksOnShutdown(); | |
| try (PoolCleaner cleaner = cleaner(p)) { | |
| final Runnable delayed = () -> { | |
| delayedDone.countDown(); | |
| p.submit(() -> immediateDone.countDown()); | |
| }; | |
| p.scheduleWithFixedDelay(delayed, 0L, Long.MAX_VALUE, SECONDS); | |
| await(delayedDone); | |
| await(immediateDone); | |
| } | |
| } | |
| /** | |
| * shutdownNow cancels tasks that were not run | |
| */ | |
| public void testShutdownNow_delayedTasks() throws InterruptedException { | |
| final ForkJoinPool p = new ForkJoinPool(2); | |
| List<ScheduledFuture<?>> tasks = new ArrayList<>(); | |
| final int DELAY = 100; | |
| for (int i = 0; i < 3; i++) { | |
| Runnable r = new NoOpRunnable(); | |
| tasks.add(p.schedule(r, DELAY, SECONDS)); | |
| tasks.add(p.scheduleAtFixedRate(r, DELAY, DELAY, SECONDS)); | |
| tasks.add(p.scheduleWithFixedDelay(r, DELAY, DELAY, SECONDS)); | |
| } | |
| p.shutdownNow(); | |
| assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS)); | |
| for (ScheduledFuture<?> task : tasks) { | |
| assertTrue(task.isDone()); | |
| } | |
| assertTrue(p.isTerminated()); | |
| } | |
| /** | |
| * submitWithTimeout (eventually) cancels task after timeout | |
| */ | |
| public void testSubmitWithTimeoutCancels() throws InterruptedException { | |
| final ForkJoinPool p = ForkJoinPool.commonPool(); | |
| Callable<Boolean> c = new Callable<Boolean>() { | |
| public Boolean call() throws Exception { | |
| Thread.sleep(LONGER_DELAY_MS); return Boolean.TRUE; }}; | |
| ForkJoinTask<?> task = p.submitWithTimeout(c, 1, NANOSECONDS, null); | |
| while(!task.isCancelled()) | |
| Thread.sleep(timeoutMillis()); | |
| } | |
| static final class SubmitWithTimeoutException extends RuntimeException {} | |
| /** | |
| * submitWithTimeout using complete completes after timeout | |
| */ | |
| public void testSubmitWithCompleterTimeoutCompletes() throws InterruptedException { | |
| final ForkJoinPool p = ForkJoinPool.commonPool(); | |
| Callable<Item> c = new Callable<Item>() { | |
| public Item call() throws Exception { | |
| Thread.sleep(LONGER_DELAY_MS); return one; }}; | |
| ForkJoinTask<Item> task = p.submitWithTimeout( | |
| c, 1, NANOSECONDS, | |
| (ForkJoinTask<Item> t) -> | |
| t.complete(two)); | |
| assertEquals(task.join(), two); | |
| } | |
| /** | |
| * submitWithTimeout using completeExceptionally throws after timeout | |
| */ | |
| public void testSubmitWithTimeoutThrows() throws InterruptedException { | |
| final ForkJoinPool p = ForkJoinPool.commonPool(); | |
| Callable<Boolean> c = new Callable<Boolean>() { | |
| public Boolean call() throws Exception { | |
| Thread.sleep(LONGER_DELAY_MS); return Boolean.TRUE; }}; | |
| ForkJoinTask<Boolean> task = p.submitWithTimeout( | |
| c, 1, NANOSECONDS, | |
| (ForkJoinTask<Boolean> t) -> | |
| t.completeExceptionally(new SubmitWithTimeoutException())); | |
| try { | |
| task.join(); | |
| shouldThrow(); | |
| } | |
| catch (Exception ex) { | |
| assertTrue(ex instanceof SubmitWithTimeoutException); | |
| } | |
| } | |
| /** | |
| * submitWithTimeout doesn't cancel if completed before timeout | |
| */ | |
| public void testSubmitWithTimeout_NoTimeout() throws InterruptedException { | |
| final ForkJoinPool p = ForkJoinPool.commonPool(); | |
| Callable<Boolean> c = new Callable<Boolean>() { | |
| public Boolean call() throws Exception { | |
| return Boolean.TRUE; }}; | |
| ForkJoinTask<?> task = p.submitWithTimeout(c, LONGER_DELAY_MS, MILLISECONDS, null); | |
| Thread.sleep(timeoutMillis()); | |
| assertFalse(task.isCancelled()); | |
| assertEquals(task.join(), Boolean.TRUE); | |
| } | |
| /** | |
| * A delayed task completes (possibly abnormally) if shutdown after | |
| * calling cancelDelayedTasksOnShutdown() | |
| */ | |
| public void testCancelDelayedTasksOnShutdown() throws Exception { | |
| final ForkJoinPool p = new ForkJoinPool(2); | |
| p.cancelDelayedTasksOnShutdown(); | |
| try (PoolCleaner cleaner = cleaner(p)) { | |
| Callable<Boolean> task = new CheckedCallable<>() { | |
| public Boolean realCall() { | |
| return Boolean.TRUE; | |
| }}; | |
| Future<Boolean> f = p.schedule(task, LONGER_DELAY_MS, MILLISECONDS); | |
| p.shutdown(); | |
| assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS)); | |
| assertTrue(f.isDone()); | |
| } | |
| } | |
| /** | |
| * schedule throws RejectedExecutionException if shutdown before | |
| * first delayed task is submitted | |
| */ | |
| public void testInitialScheduleAfterShutdown() throws InterruptedException { | |
| Runnable r = new NoOpRunnable(); | |
| boolean rje = false; | |
| try (final ForkJoinPool p = new ForkJoinPool(1)) { | |
| p.shutdown(); | |
| assertTrue(p.awaitTermination(LONG_DELAY_MS, MILLISECONDS)); | |
| try { | |
| p.schedule(r, 1, MILLISECONDS); | |
| } catch (RejectedExecutionException ok) { | |
| rje = true; | |
| } | |
| } | |
| assertTrue(rje); | |
| } | |
| } |