Java 类java.util.concurrent.ForkJoinTask 实例源码

/**
 * join of a forked task throws exception when task cancelled
 */
public void testCancelledForkJoinCC() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF f = new LCCF(null, 8);
            assertTrue(f.cancel(true));
            assertSame(f, f.fork());
            try {
                f.join();
                shouldThrow();
            } catch (CancellationException success) {
                checkCancelled(f);
            }
        }};
    checkInvoke(a);
}

/**
 * invokeAll(tasks) with > 2 argument throws exception if any task does
 */
public void testAbnormalInvokeAll3() {
    RecursiveAction a = new CheckedRecursiveAction() {
        protected void realCompute() {
            AsyncFib f = new AsyncFib(8);
            FailingAsyncFib g = new FailingAsyncFib(9);
            AsyncFib h = new AsyncFib(7);
            ForkJoinTask[] tasks = { f, g, h };
            shuffle(tasks);
            try {
                invokeAll(tasks);
                shouldThrow();
            } catch (FJException success) {
                checkCompletedAbnormally(g, success);
            }
        }};
    testInvokeOnPool(mainPool(), a);
}

/**
 * invokeAll(collection) throws exception if any task does
 */
public void testAbnormalInvokeAllCollection() {
    RecursiveAction a = new CheckedRecursiveAction() {
        protected void realCompute() {
            FailingAsyncFib f = new FailingAsyncFib(8);
            AsyncFib g = new AsyncFib(9);
            AsyncFib h = new AsyncFib(7);
            ForkJoinTask[] tasks = { f, g, h };
            shuffle(tasks);
            try {
                invokeAll(Arrays.asList(tasks));
                shouldThrow();
            } catch (FJException success) {
                checkCompletedAbnormally(f, success);
            }
        }};
    testInvokeOnPool(mainPool(), a);
}

void checkNotDone(ForkJoinTask a) {
    assertFalse(a.isDone());
    assertFalse(a.isCompletedNormally());
    assertFalse(a.isCompletedAbnormally());
    assertFalse(a.isCancelled());
    assertNull(a.getException());
    assertNull(a.getRawResult());
    if (a instanceof BinaryAsyncAction)
        assertEquals(INITIAL_STATE,
                     ((BinaryAsyncAction)a).getForkJoinTaskTag());

    try {
        a.get(randomExpiredTimeout(), randomTimeUnit());
        shouldThrow();
    } catch (TimeoutException success) {
    } catch (Throwable fail) { threadUnexpectedException(fail); }
}

public void testAbnormalInvokeAll2(ForkJoinPool pool) {
    RecursiveAction a = new CheckedRecursiveAction() {
        protected void realCompute() {
            AsyncFib f = new AsyncFib(8);
            FailingAsyncFib g = new FailingAsyncFib(9);
            ForkJoinTask[] tasks = { f, g };
            shuffle(tasks);
            try {
                invokeAll(tasks[0], tasks[1]);
                shouldThrow();
            } catch (FJException success) {
                checkCompletedAbnormally(g, success);
            }
        }};
    testInvokeOnPool(pool, a);
}

public void testAbnormalInvokeAll3(ForkJoinPool pool) {
    RecursiveAction a = new CheckedRecursiveAction() {
        protected void realCompute() {
            AsyncFib f = new AsyncFib(8);
            FailingAsyncFib g = new FailingAsyncFib(9);
            AsyncFib h = new AsyncFib(7);
            ForkJoinTask[] tasks = { f, g, h };
            shuffle(tasks);
            try {
                invokeAll(tasks[0], tasks[1], tasks[2]);
                shouldThrow();
            } catch (FJException success) {
                checkCompletedAbnormally(g, success);
            }
        }};
    testInvokeOnPool(pool, a);
}

/**
 * invokeAll(collection) throws exception if any task does
 */
public void testAbnormalInvokeAllCollectionSingleton() {
    RecursiveAction a = new CheckedRecursiveAction() {
        protected void realCompute() {
            FailingAsyncFib f = new FailingAsyncFib(8);
            AsyncFib g = new AsyncFib(9);
            AsyncFib h = new AsyncFib(7);
            ForkJoinTask[] tasks = { f, g, h };
            shuffle(tasks);
            try {
                invokeAll(Arrays.asList(tasks));
                shouldThrow();
            } catch (FJException success) {
                checkCompletedAbnormally(f, success);
            }
        }};
    testInvokeOnPool(singletonPool(), a);
}

/**
 * invokeAll(t1, t2) throw exception if any task does
 */
public void testAbnormalInvokeAll2() {
    RecursiveAction a = new CheckedRecursiveAction() {
        protected void realCompute() {
            AsyncFib f = new AsyncFib(8);
            FailingAsyncFib g = new FailingAsyncFib(9);
            ForkJoinTask[] tasks = { f, g };
            shuffle(tasks);
            try {
                invokeAll(tasks);
                shouldThrow();
            } catch (FJException success) {
                checkCompletedAbnormally(g, success);
            }
        }};
    testInvokeOnPool(mainPool(), a);
}

/**
 * timed get of a forked task throws exception when task completes abnormally
 */
public void testAbnormalForkTimedGetCC() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() throws Exception {
            FailingCCF f = new LFCCF(null, 8);
            assertSame(f, f.fork());
            try {
                f.get(LONG_DELAY_MS, MILLISECONDS);
                shouldThrow();
            } catch (ExecutionException success) {
                Throwable cause = success.getCause();
                assertTrue(cause instanceof FJException);
                checkCompletedAbnormally(f, cause);
            }
        }};
    checkInvoke(a);
}

/**
 * timed get of a forked task throws exception when task completes abnormally
 */
public void testAbnormalForkTimedGet() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() throws Exception {
            FailingCCF f = new LFCCF(8);
            assertSame(f, f.fork());
            try {
                f.get(LONG_DELAY_MS, MILLISECONDS);
                shouldThrow();
            } catch (ExecutionException success) {
                Throwable cause = success.getCause();
                assertTrue(cause instanceof FJException);
                checkCompletedAbnormally(f, cause);
            }
        }};
    testInvokeOnPool(mainPool(), a);
}

/**
 * join of a forked task throws exception when task cancelled
 */
public void testCancelledForkJoin() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF f = new LCCF(8);
            assertTrue(f.cancel(true));
            assertSame(f, f.fork());
            try {
                f.join();
                shouldThrow();
            } catch (CancellationException success) {
                checkCancelled(f);
            }
        }};
    testInvokeOnPool(mainPool(), a);
}

/**
 * get of a forked task throws exception when task cancelled
 */
public void testCancelledForkGet() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() throws Exception {
            CCF f = new LCCF(8);
            assertTrue(f.cancel(true));
            assertSame(f, f.fork());
            try {
                f.get();
                shouldThrow();
            } catch (CancellationException success) {
                checkCancelled(f);
            }
        }};
    testInvokeOnPool(mainPool(), a);
}

/**
 * timed get of a forked task throws exception when task cancelled
 */
public void testCancelledForkTimedGet() throws Exception {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() throws Exception {
            CCF f = new LCCF(8);
            assertTrue(f.cancel(true));
            assertSame(f, f.fork());
            try {
                f.get(LONG_DELAY_MS, MILLISECONDS);
                shouldThrow();
            } catch (CancellationException success) {
                checkCancelled(f);
            }
        }};
    testInvokeOnPool(mainPool(), a);
}

/**
 * invokeAll(tasks) with > 2 argument invokes tasks
 */
public void testInvokeAll3() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF f = new LCCF(8);
            CCF g = new LCCF(9);
            CCF h = new LCCF(7);
            invokeAll(f, g, h);
            assertEquals(21, f.number);
            assertEquals(34, g.number);
            assertEquals(13, h.number);
            checkCompletedNormally(f);
            checkCompletedNormally(g);
            checkCompletedNormally(h);
        }};
    testInvokeOnPool(mainPool(), a);
}

/**
 * invokeAll(collection) invokes all tasks in the collection
 */
public void testInvokeAllCollection() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF f = new LCCF(8);
            CCF g = new LCCF(9);
            CCF h = new LCCF(7);
            HashSet set = new HashSet();
            set.add(f);
            set.add(g);
            set.add(h);
            invokeAll(set);
            assertEquals(21, f.number);
            assertEquals(34, g.number);
            assertEquals(13, h.number);
            checkCompletedNormally(f);
            checkCompletedNormally(g);
            checkCompletedNormally(h);
        }};
    testInvokeOnPool(mainPool(), a);
}

/**
 * invokeAll(tasks) with > 2 argument throws exception if any task does
 */
public void testAbnormalInvokeAll3() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF f = new LCCF(8);
            FailingCCF g = new LFCCF(9);
            CCF h = new LCCF(7);
            try {
                invokeAll(f, g, h);
                shouldThrow();
            } catch (FJException success) {
                checkCompletedAbnormally(g, success);
            }
        }};
    testInvokeOnPool(mainPool(), a);
}

/**
 * invokeAll(collection) throws exception if any task does
 */
public void testAbnormalInvokeAllCollection() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            FailingCCF f = new LFCCF(8);
            CCF g = new LCCF(9);
            CCF h = new LCCF(7);
            HashSet set = new HashSet();
            set.add(f);
            set.add(g);
            set.add(h);
            try {
                invokeAll(set);
                shouldThrow();
            } catch (FJException success) {
                checkCompletedAbnormally(f, success);
            }
        }};
    testInvokeOnPool(mainPool(), a);
}

/**
 * get of a forked task throws exception when task completes abnormally
 */
public void testAbnormalForkGetCC() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() throws Exception {
            FailingCCF f = new LFCCF(null, 8);
            assertSame(f, f.fork());
            try {
                f.get();
                shouldThrow();
            } catch (ExecutionException success) {
                Throwable cause = success.getCause();
                assertTrue(cause instanceof FJException);
                checkCompletedAbnormally(f, cause);
            }
        }};
    checkInvoke(a);
}

/**
 * peekNextLocalTask returns most recent unexecuted task.
 */
public void testPeekNextLocalTask() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF g = new LCCF(9);
            assertSame(g, g.fork());
            CCF f = new LCCF(8);
            assertSame(f, f.fork());
            assertSame(f, peekNextLocalTask());
            assertNull(f.join());
            checkCompletedNormally(f);
            helpQuiesce();
            checkCompletedNormally(g);
        }};
    testInvokeOnPool(singletonPool(), a);
}

/**
 * pollNextLocalTask returns most recent unexecuted task without
 * executing it
 */
public void testPollNextLocalTask() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF g = new LCCF(9);
            assertSame(g, g.fork());
            CCF f = new LCCF(8);
            assertSame(f, f.fork());
            assertSame(f, pollNextLocalTask());
            helpQuiesce();
            checkNotDone(f);
            assertEquals(34, g.number);
            checkCompletedNormally(g);
        }};
    testInvokeOnPool(singletonPool(), a);
}

/**
 * peekNextLocalTask returns least recent unexecuted task in async mode
 */
public void testPeekNextLocalTaskAsync() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF g = new LCCF(9);
            assertSame(g, g.fork());
            CCF f = new LCCF(8);
            assertSame(f, f.fork());
            assertSame(g, peekNextLocalTask());
            assertNull(f.join());
            helpQuiesce();
            checkCompletedNormally(f);
            assertEquals(34, g.number);
            checkCompletedNormally(g);
        }};
    testInvokeOnPool(asyncSingletonPool(), a);
}

/**
 * pollNextLocalTask returns least recent unexecuted task without
 * executing it, in async mode
 */
public void testPollNextLocalTaskAsync() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF g = new LCCF(9);
            assertSame(g, g.fork());
            CCF f = new LCCF(8);
            assertSame(f, f.fork());
            assertSame(g, pollNextLocalTask());
            helpQuiesce();
            assertEquals(21, f.number);
            checkCompletedNormally(f);
            checkNotDone(g);
        }};
    testInvokeOnPool(asyncSingletonPool(), a);
}

/**
 * pollTask returns an unexecuted task without executing it, in
 * async mode
 */
public void testPollTaskAsync() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF g = new LCCF(9);
            assertSame(g, g.fork());
            CCF f = new LCCF(8);
            assertSame(f, f.fork());
            assertSame(g, pollTask());
            helpQuiesce();
            assertEquals(21, f.number);
            checkCompletedNormally(f);
            checkNotDone(g);
        }};
    testInvokeOnPool(asyncSingletonPool(), a);
}

/**
 * invokeAll(t1, t2) throw exception if any task does
 */
public void testAbnormalInvokeAll2Singleton() {
    RecursiveAction a = new CheckedRecursiveAction() {
        protected void realCompute() {
            AsyncFib f = new AsyncFib(8);
            FailingAsyncFib g = new FailingAsyncFib(9);
            ForkJoinTask[] tasks = { f, g };
            shuffle(tasks);
            try {
                invokeAll(tasks);
                shouldThrow();
            } catch (FJException success) {
                checkCompletedAbnormally(g, success);
            }
        }};
    testInvokeOnPool(singletonPool(), a);
}

/**
 * get of a forked task throws exception when task completes abnormally
 */
public void testAbnormalForkGetSingleton() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() throws Exception {
            FailingCCF f = new LFCCF(8);
            assertSame(f, f.fork());
            try {
                f.get();
                shouldThrow();
            } catch (ExecutionException success) {
                Throwable cause = success.getCause();
                assertTrue(cause instanceof FJException);
                checkCompletedAbnormally(f, cause);
            }
        }};
    testInvokeOnPool(singletonPool(), a);
}

/**
 * timed get of a forked task throws exception when task completes abnormally
 */
public void testAbnormalForkTimedGetSingleton() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() throws Exception {
            FailingCCF f = new LFCCF(8);
            assertSame(f, f.fork());
            try {
                f.get(LONG_DELAY_MS, MILLISECONDS);
                shouldThrow();
            } catch (ExecutionException success) {
                Throwable cause = success.getCause();
                assertTrue(cause instanceof FJException);
                checkCompletedAbnormally(f, cause);
            }
        }};
    testInvokeOnPool(singletonPool(), a);
}

/**
 * invokeAll(collection) invokes all tasks in the collection
 */
public void testInvokeAllCollectionCC() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF f = new LCCF(null, 8);
            CCF g = new LCCF(null, 9);
            CCF h = new LCCF(null, 7);
            HashSet set = new HashSet();
            set.add(f);
            set.add(g);
            set.add(h);
            invokeAll(set);
            assertEquals(21, f.number);
            assertEquals(34, g.number);
            assertEquals(13, h.number);
            checkCompletedNormally(f);
            checkCompletedNormally(g);
            checkCompletedNormally(h);
        }};
    checkInvoke(a);
}

/**
 * get of a forked task throws exception when task cancelled
 */
public void testCancelledForkGetSingleton() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() throws Exception {
            CCF f = new LCCF(8);
            assertTrue(f.cancel(true));
            assertSame(f, f.fork());
            try {
                f.get();
                shouldThrow();
            } catch (CancellationException success) {
                checkCancelled(f);
            }
        }};
    testInvokeOnPool(singletonPool(), a);
}

/**
 * timed get of a forked task throws exception when task cancelled
 */
public void testCancelledForkTimedGetSingleton() throws Exception {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() throws Exception {
            CCF f = new LCCF(8);
            assertTrue(f.cancel(true));
            assertSame(f, f.fork());
            try {
                f.get(LONG_DELAY_MS, MILLISECONDS);
                shouldThrow();
            } catch (CancellationException success) {
                checkCancelled(f);
            }
        }};
    testInvokeOnPool(singletonPool(), a);
}

/**
 * invokeAll(tasks) with > 2 argument invokes tasks
 */
public void testInvokeAll3CC() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF f = new LCCF(null, 8);
            CCF g = new LCCF(null, 9);
            CCF h = new LCCF(null, 7);
            invokeAll(f, g, h);
            assertEquals(21, f.number);
            assertEquals(34, g.number);
            assertEquals(13, h.number);
            checkCompletedNormally(f);
            checkCompletedNormally(g);
            checkCompletedNormally(h);
        }};
    checkInvoke(a);
}

/**
 * invokeAll(tasks) with > 2 argument invokes tasks
 */
public void testInvokeAll3Singleton() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF f = new LCCF(8);
            CCF g = new LCCF(9);
            CCF h = new LCCF(7);
            invokeAll(f, g, h);
            assertEquals(21, f.number);
            assertEquals(34, g.number);
            assertEquals(13, h.number);
            checkCompletedNormally(f);
            checkCompletedNormally(g);
            checkCompletedNormally(h);
        }};
    testInvokeOnPool(singletonPool(), a);
}

/**
 * invokeAll(collection) invokes all tasks in the collection
 */
public void testInvokeAllCollectionSingleton() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF f = new LCCF(8);
            CCF g = new LCCF(9);
            CCF h = new LCCF(7);
            HashSet set = new HashSet();
            set.add(f);
            set.add(g);
            set.add(h);
            invokeAll(set);
            assertEquals(21, f.number);
            assertEquals(34, g.number);
            assertEquals(13, h.number);
            checkCompletedNormally(f);
            checkCompletedNormally(g);
            checkCompletedNormally(h);
        }};
    testInvokeOnPool(singletonPool(), a);
}

/**
 * invokeAll(tasks) with > 2 argument throws exception if any task does
 */
public void testAbnormalInvokeAll3Singleton() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF f = new LCCF(8);
            FailingCCF g = new LFCCF(9);
            CCF h = new LCCF(7);
            try {
                invokeAll(f, g, h);
                shouldThrow();
            } catch (FJException success) {
                checkCompletedAbnormally(g, success);
            }
        }};
    testInvokeOnPool(singletonPool(), a);
}

/**
 * invokeAll(collection) throws exception if any task does
 */
public void testAbnormalInvokeAllCollectionSingleton() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            FailingCCF f = new LFCCF(8);
            CCF g = new LCCF(9);
            CCF h = new LCCF(7);
            HashSet set = new HashSet();
            set.add(f);
            set.add(g);
            set.add(h);
            try {
                invokeAll(set);
                shouldThrow();
            } catch (FJException success) {
                checkCompletedAbnormally(f, success);
            }
        }};
    testInvokeOnPool(singletonPool(), a);
}

/**
 * invokeAll(tasks) with > 2 argument throws exception if any task does
 */
public void testAbnormalInvokeAll3Singleton() {
    RecursiveAction a = new CheckedRecursiveAction() {
        protected void realCompute() {
            AsyncFib f = new AsyncFib(8);
            FailingAsyncFib g = new FailingAsyncFib(9);
            AsyncFib h = new AsyncFib(7);
            ForkJoinTask[] tasks = { f, g, h };
            shuffle(tasks);
            try {
                invokeAll(tasks);
                shouldThrow();
            } catch (FJException success) {
                checkCompletedAbnormally(g, success);
            }
        }};
    testInvokeOnPool(singletonPool(), a);
}

/**
 * A task submitted after shutdown is rejected
 */
public void testSubmitAfterShutdown() {
    ForkJoinPool p = new ForkJoinPool(1);
    try (PoolCleaner cleaner = cleaner(p)) {
        p.shutdown();
        assertTrue(p.isShutdown());
        try {
            ForkJoinTask<Integer> f = p.submit(new FibTask(8));
            shouldThrow();
        } catch (RejectedExecutionException success) {}
    }
}

/**
 * quietlyInvoke task returns when task completes abnormally
 */
public void testAbnormalQuietlyInvokeCC() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            FailingCCF f = new LFCCF(null, 8);
            f.quietlyInvoke();
            assertTrue(f.getException() instanceof FJException);
            checkCompletedAbnormally(f, f.getException());
        }};
    checkInvoke(a);
}

protected Double compute() {
    double max = speedLimitByLane[laneNumber - 1];
    double min = laneNumber == 1 ? 0.0 : speedLimitByLane[laneNumber - 2];
    if (trafficUnitsNumber < threshold) {
        double speed = FactoryTraffic.getTrafficUnitStream(dateLocation, trafficUnitsNumber)
                .map(TrafficUnitWrapper::new)
                .map(tuw -> tuw.setSpeedModel(speedModel))
                .map(tuw -> calcSpeed(tuw.getVehicle(), tuw.getTraction(), timeSec))
                .mapToDouble(sp -> sp)
                .filter(sp -> sp > min && sp <= max)
                .average()
                .getAsDouble();
        return (double) Math.round(speed);
    }
    else{
        int tun = trafficUnitsNumber / 2;
        AverageSpeed as1 = new AverageSpeed(tun, timeSec, dateLocation, speedLimitByLane, laneNumber, threshold);
        AverageSpeed as2 = new AverageSpeed(tun, timeSec, dateLocation, speedLimitByLane, laneNumber, threshold);
        return ForkJoinTask.invokeAll(List.of(as1, as2))
                .stream()
                .mapToDouble(ForkJoinTask::join)
                .map(Math::round)
                .average()
                .getAsDouble();
    }

}

private static double doInvokeAll(AverageSpeed as1, AverageSpeed as2){
    return ForkJoinTask.invokeAll(List.of(as1, as2))
            .stream()
            .mapToDouble(ForkJoinTask::join)
            .map(Math::round)
            .average()
            .getAsDouble();
}

/**
 * quietlyJoin of a forked task returns when task cancelled
 */
public void testCancelledForkQuietlyJoinCC() {
    ForkJoinTask a = new CheckedRecursiveAction() {
        protected void realCompute() {
            CCF f = new LCCF(null, 8);
            assertTrue(f.cancel(true));
            assertSame(f, f.fork());
            f.quietlyJoin();
            checkCancelled(f);
        }};
    checkInvoke(a);
}