/*
 * 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);
    }
}