/*

 * 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/

 * Other contributors include Andrew Wright, Jeffrey Hayes,

 * Pat Fisher, Mike Judd.

 */

import static java.util.concurrent.TimeUnit.MILLISECONDS;

import static java.util.concurrent.TimeUnit.SECONDS;

import java.util.ArrayList;

import java.util.Collection;

import java.util.Collections;

import java.util.List;

import java.util.concurrent.ArrayBlockingQueue;

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.Future;

import java.util.concurrent.FutureTask;

import java.util.concurrent.LinkedBlockingQueue;

import java.util.concurrent.RejectedExecutionHandler;

import java.util.concurrent.RunnableFuture;

import java.util.concurrent.SynchronousQueue;

import java.util.concurrent.ThreadFactory;

import java.util.concurrent.ThreadLocalRandom;

import java.util.concurrent.ThreadPoolExecutor;

import java.util.concurrent.TimeoutException;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.concurrent.locks.Condition;

import java.util.concurrent.locks.ReentrantLock;

import junit.framework.Test;

import junit.framework.TestSuite;

public class ThreadPoolExecutorSubclassTest extends JSR166TestCase {

    public static void main(String[] args) {

        main(suite(), args);

    }

    public static Test suite() {

        return new TestSuite(ThreadPoolExecutorSubclassTest.class);

    }

    static class CustomTask<V> implements RunnableFuture<V> {

        final Callable<V> callable;

        final ReentrantLock lock = new ReentrantLock();

        final Condition cond = lock.newCondition();

        boolean done;

        boolean cancelled;

        V result;

        Thread thread;

        Exception exception;

        CustomTask(Callable<V> c) {

            if (c == null) throw new NullPointerException();

            callable = c;

        }

        CustomTask(final Runnable r, final V res) {

            if (r == null) throw new NullPointerException();

            callable = new Callable<V>() {

                public V call() throws Exception { r.run(); return res; }};

        }

        public boolean isDone() {

            lock.lock(); try { return done; } finally { lock.unlock() ; }

        }

        public boolean isCancelled() {

            lock.lock(); try { return cancelled; } finally { lock.unlock() ; }

        }

        public boolean cancel(boolean mayInterrupt) {

            lock.lock();

            try {

                if (!done) {

                    cancelled = true;

                    done = true;

                    if (mayInterrupt && thread != null)

                        thread.interrupt();

                    return true;

                }

                return false;

            }

            finally { lock.unlock() ; }

        }

        public void run() {

            lock.lock();

            try {

                if (done)

                    return;

                thread = Thread.currentThread();

            }

            finally { lock.unlock() ; }

            V v = null;

            Exception e = null;

            try {

                v = callable.call();

            }

            catch (Exception ex) {

                e = ex;

            }

            lock.lock();

            try {

                if (!done) {

                    result = v;

                    exception = e;

                    done = true;

                    thread = null;

                    cond.signalAll();

                }

            }

            finally { lock.unlock(); }

        }

        public V get() throws InterruptedException, ExecutionException {

            lock.lock();

            try {

                while (!done)

                    cond.await();

                if (cancelled)

                    throw new CancellationException();

                if (exception != null)

                    throw new ExecutionException(exception);

                return result;

            }

            finally { lock.unlock(); }

        }

        public V get(long timeout, TimeUnit unit)

            throws InterruptedException, ExecutionException, TimeoutException {

            long nanos = unit.toNanos(timeout);

            lock.lock();

            try {

                while (!done) {

                    if (nanos <= 0L)

                        throw new TimeoutException();

                    nanos = cond.awaitNanos(nanos);

                }

                if (cancelled)

                    throw new CancellationException();

                if (exception != null)

                    throw new ExecutionException(exception);

                return result;

            }

            finally { lock.unlock(); }

        }

    }

    static class CustomTPE extends ThreadPoolExecutor {

        protected <V> RunnableFuture<V> newTaskFor(Callable<V> c) {

            return new CustomTask<V>(c);

        }

        protected <V> RunnableFuture<V> newTaskFor(Runnable r, V v) {

            return new CustomTask<V>(r, v);

        }

        CustomTPE(int corePoolSize,

                  int maximumPoolSize,

                  long keepAliveTime,

                  TimeUnit unit,

                  BlockingQueue<Runnable> workQueue) {

            super(corePoolSize, maximumPoolSize, keepAliveTime, unit,

                  workQueue);

        }

        CustomTPE(int corePoolSize,

                  int maximumPoolSize,

                  long keepAliveTime,

                  TimeUnit unit,

                  BlockingQueue<Runnable> workQueue,

                  ThreadFactory threadFactory) {

        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,

             threadFactory);

        }

        CustomTPE(int corePoolSize,

                  int maximumPoolSize,

                  long keepAliveTime,

                  TimeUnit unit,

                  BlockingQueue<Runnable> workQueue,

                  RejectedExecutionHandler handler) {

        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,

              handler);

        }

        CustomTPE(int corePoolSize,

                  int maximumPoolSize,

                  long keepAliveTime,

                  TimeUnit unit,

                  BlockingQueue<Runnable> workQueue,

                  ThreadFactory threadFactory,

                  RejectedExecutionHandler handler) {

            super(corePoolSize, maximumPoolSize, keepAliveTime, unit,

              workQueue, threadFactory, handler);

        }

        final CountDownLatch beforeCalled = new CountDownLatch(1);

        final CountDownLatch afterCalled = new CountDownLatch(1);

        final CountDownLatch terminatedCalled = new CountDownLatch(1);

        public CustomTPE() {

            super(1, 1, LONG_DELAY_MS, MILLISECONDS, new SynchronousQueue<Runnable>());

        }

        protected void beforeExecute(Thread t, Runnable r) {

            beforeCalled.countDown();

        }

        protected void afterExecute(Runnable r, Throwable t) {

            afterCalled.countDown();

        }

        protected void terminated() {

            terminatedCalled.countDown();

        }

        public boolean beforeCalled() {

            return beforeCalled.getCount() == 0;

        }

        public boolean afterCalled() {

            return afterCalled.getCount() == 0;

        }

        public boolean terminatedCalled() {

            return terminatedCalled.getCount() == 0;

        }

    }

    static class FailingThreadFactory implements ThreadFactory {

        int calls = 0;

        public Thread newThread(Runnable r) {

            if (++calls > 1) return null;

            return new Thread(r);

        }

    }

    /**

     * execute successfully executes a runnable

     */

    public void testExecute() throws InterruptedException {

        final ThreadPoolExecutor p =

            new CustomTPE(1, 1,

                          2 * LONG_DELAY_MS, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10));

        try (PoolCleaner cleaner = cleaner(p)) {

            final CountDownLatch done = new CountDownLatch(1);

            final Runnable task = new CheckedRunnable() {

                public void realRun() { done.countDown(); }};

            p.execute(task);

            await(done);

        }

    }

    /**

     * getActiveCount increases but doesn't overestimate, when a

     * thread becomes active

     */

    public void testGetActiveCount() throws InterruptedException {

        final CountDownLatch done = new CountDownLatch(1);

        final ThreadPoolExecutor p =

            new CustomTPE(2, 2,

                          LONG_DELAY_MS, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10));

        try (PoolCleaner cleaner = cleaner(p, done)) {

            final CountDownLatch threadStarted = new CountDownLatch(1);

            assertEquals(0, p.getActiveCount());

            p.execute(new CheckedRunnable() {

                public void realRun() throws InterruptedException {

                    threadStarted.countDown();

                    assertEquals(1, p.getActiveCount());

                    await(done);

                }});

            await(threadStarted);

            assertEquals(1, p.getActiveCount());

        }

    }

    /**

     * prestartCoreThread starts a thread if under corePoolSize, else doesn't

     */

    public void testPrestartCoreThread() {

        final ThreadPoolExecutor p =

            new CustomTPE(2, 6,

                          LONG_DELAY_MS, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10));

        try (PoolCleaner cleaner = cleaner(p)) {

            assertEquals(0, p.getPoolSize());

            assertTrue(p.prestartCoreThread());

            assertEquals(1, p.getPoolSize());

            assertTrue(p.prestartCoreThread());

            assertEquals(2, p.getPoolSize());

            assertFalse(p.prestartCoreThread());

            assertEquals(2, p.getPoolSize());

            p.setCorePoolSize(4);

            assertTrue(p.prestartCoreThread());

            assertEquals(3, p.getPoolSize());

            assertTrue(p.prestartCoreThread());

            assertEquals(4, p.getPoolSize());

            assertFalse(p.prestartCoreThread());

            assertEquals(4, p.getPoolSize());

        }

    }

    /**

     * prestartAllCoreThreads starts all corePoolSize threads

     */

    public void testPrestartAllCoreThreads() {

        final ThreadPoolExecutor p =

            new CustomTPE(2, 6,

                          LONG_DELAY_MS, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10));

        try (PoolCleaner cleaner = cleaner(p)) {

            assertEquals(0, p.getPoolSize());

            p.prestartAllCoreThreads();

            assertEquals(2, p.getPoolSize());

            p.prestartAllCoreThreads();

            assertEquals(2, p.getPoolSize());

            p.setCorePoolSize(4);

            p.prestartAllCoreThreads();

            assertEquals(4, p.getPoolSize());

            p.prestartAllCoreThreads();

            assertEquals(4, p.getPoolSize());

        }

    }

    /**

     * getCompletedTaskCount increases, but doesn't overestimate,

     * when tasks complete

     */

    public void testGetCompletedTaskCount() throws InterruptedException {

        final ThreadPoolExecutor p =

            new CustomTPE(2, 2,

                          LONG_DELAY_MS, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10));

        try (PoolCleaner cleaner = cleaner(p)) {

            final CountDownLatch threadStarted = new CountDownLatch(1);

            final CountDownLatch threadProceed = new CountDownLatch(1);

            final CountDownLatch threadDone = new CountDownLatch(1);

            assertEquals(0, p.getCompletedTaskCount());

            p.execute(new CheckedRunnable() {

                public void realRun() throws InterruptedException {

                    threadStarted.countDown();

                    assertEquals(0, p.getCompletedTaskCount());

                    await(threadProceed);

                    threadDone.countDown();

                }});

            await(threadStarted);

            assertEquals(0, p.getCompletedTaskCount());

            threadProceed.countDown();

            await(threadDone);

            long startTime = System.nanoTime();

            while (p.getCompletedTaskCount() != 1) {

                if (millisElapsedSince(startTime) > LONG_DELAY_MS)

                    fail("timed out");

                Thread.yield();

            }

        }

    }

    /**

     * getCorePoolSize returns size given in constructor if not otherwise set

     */

    public void testGetCorePoolSize() {

        final ThreadPoolExecutor p =

            new CustomTPE(1, 1,

                          LONG_DELAY_MS, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10));

        try (PoolCleaner cleaner = cleaner(p)) {

            assertEquals(1, p.getCorePoolSize());

        }

    }

    /**

     * getKeepAliveTime returns value given in constructor if not otherwise set

     */

    public void testGetKeepAliveTime() {

        final ThreadPoolExecutor p =

            new CustomTPE(2, 2,

                          1000, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10));

        try (PoolCleaner cleaner = cleaner(p)) {

            assertEquals(1, p.getKeepAliveTime(SECONDS));

        }

    }

    /**

     * getThreadFactory returns factory in constructor if not set

     */

    public void testGetThreadFactory() {

        final ThreadFactory threadFactory = new SimpleThreadFactory();

        final ThreadPoolExecutor p =

            new CustomTPE(1, 2,

                          LONG_DELAY_MS, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10),

                          threadFactory,

                          new NoOpREHandler());

        try (PoolCleaner cleaner = cleaner(p)) {

            assertSame(threadFactory, p.getThreadFactory());

        }

    }

    /**

     * setThreadFactory sets the thread factory returned by getThreadFactory

     */

    public void testSetThreadFactory() {

        final ThreadPoolExecutor p =

            new CustomTPE(1, 2,

                          LONG_DELAY_MS, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10));

        try (PoolCleaner cleaner = cleaner(p)) {

            ThreadFactory threadFactory = new SimpleThreadFactory();

            p.setThreadFactory(threadFactory);

            assertSame(threadFactory, p.getThreadFactory());

        }

    }

    /**

     * setThreadFactory(null) throws NPE

     */

    public void testSetThreadFactoryNull() {

        final ThreadPoolExecutor p =

            new CustomTPE(1, 2,

                          LONG_DELAY_MS, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10));

        try (PoolCleaner cleaner = cleaner(p)) {

            try {

                p.setThreadFactory(null);

                shouldThrow();

            } catch (NullPointerException success) {}

        }

    }

    /**

     * getRejectedExecutionHandler returns handler in constructor if not set

     */

    public void testGetRejectedExecutionHandler() {

        final RejectedExecutionHandler handler = new NoOpREHandler();

        final ThreadPoolExecutor p =

            new CustomTPE(1, 2,

                          LONG_DELAY_MS, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10),

                          handler);

        try (PoolCleaner cleaner = cleaner(p)) {

            assertSame(handler, p.getRejectedExecutionHandler());

        }

    }

    /**

     * setRejectedExecutionHandler sets the handler returned by

     * getRejectedExecutionHandler

     */

    public void testSetRejectedExecutionHandler() {

        final ThreadPoolExecutor p =

            new CustomTPE(1, 2,

                          LONG_DELAY_MS, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10));

        try (PoolCleaner cleaner = cleaner(p)) {

            RejectedExecutionHandler handler = new NoOpREHandler();

            p.setRejectedExecutionHandler(handler);

            assertSame(handler, p.getRejectedExecutionHandler());

        }

    }

    /**

     * setRejectedExecutionHandler(null) throws NPE

     */

    public void testSetRejectedExecutionHandlerNull() {

        final ThreadPoolExecutor p =

            new CustomTPE(1, 2,

                          LONG_DELAY_MS, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10));

        try (PoolCleaner cleaner = cleaner(p)) {

            try {

                p.setRejectedExecutionHandler(null);

                shouldThrow();

            } catch (NullPointerException success) {}

        }

    }

    /**

     * getLargestPoolSize increases, but doesn't overestimate, when

     * multiple threads active

     */

    public void testGetLargestPoolSize() throws InterruptedException {

        final int THREADS = 3;

        final CountDownLatch done = new CountDownLatch(1);

        final ThreadPoolExecutor p =

            new CustomTPE(THREADS, THREADS,

                          LONG_DELAY_MS, MILLISECONDS,

                          new ArrayBlockingQueue<Runnable>(10));

        try (PoolCleaner cleaner = cleaner(p, done)) {

            assertEquals(0, p.getLargestPoolSize());

            final CountDownLatch threadsStarted = new CountDownLatch(THREADS);

            for (int i = 0; i < THREADS; i++)

                p.execute(new CheckedRunnable() {

                    public void realRun() throws InterruptedException {

                        threadsStarted.countDown();

                        await(done);

                        assertEquals(THREADS, p.getLargestPoolSize());

                    }});

            await(threadsStarted);

            assertEquals(THREADS, p.getLargestPoolSize());

        }

        assertEquals(THREADS, p.getLargestPoolSize());

    }

    /**