/*
 * Written by Dawid Kurzyniec and released to the public domain, as explained
 * at http://creativecommons.org/licenses/publicdomain
 */

package edu.emory.mathcs.util.concurrent;

import java.security.*;
import java.util.*;

import edu.emory.mathcs.backport.java.util.concurrent.*;
import edu.emory.mathcs.backport.java.util.concurrent.atomic.*;

/**
 *
 * @author Dawid Kurzyniec
 * @version 1.0
 */
public class ExecutorUtils {
    private ExecutorUtils() {}

    /**
     * Creates a thread pool that reuses a fixed set of threads
     * operating off a shared unbounded queue. If any thread
     * terminates due to a failure during execution prior to shutdown,
     * a new one will take its place if needed to execute subsequent
     * tasks.
     *
     * @param nThreads the number of threads in the pool
     * @return the newly created thread pool
     */
    public static ExecutorService newFixedSecureThreadPool(int nThreads) {
        return new SecureThreadPoolExecutor(nThreads, nThreads,
                                            0L, TimeUnit.MILLISECONDS,
                                            new LinkedBlockingQueue());
    }

    /**
     * Creates an Executor that uses a single worker thread operating
     * off an unbounded queue. (Note however that if this single
     * thread terminates due to a failure during execution prior to
     * shutdown, a new one will take its place if needed to execute
     * subsequent tasks.)  Tasks are guaranteed to execute
     * sequentially, and no more than one task will be active at any
     * given time. Unlike the otherwise equivalent
     * <tt>newFixedThreadPool(1)</tt> the returned executor is
     * guaranteed not to be reconfigurable to use additional threads.
     *
     * @return the newly created single-threaded Executor
     */
    public static ExecutorService newSecureSingleThreadExecutor() {
        return new DelegatedExecutorService
            (new SecureThreadPoolExecutor(1, 1,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue()));
    }

    /**
     * Creates a thread pool that creates new threads as needed, but
     * will reuse previously constructed threads when they are
     * available.  These pools will typically improve the performance
     * of programs that execute many short-lived asynchronous tasks.
     * Calls to <tt>execute</tt> will reuse previously constructed
     * threads if available. If no existing thread is available, a new
     * thread will be created and added to the pool. Threads that have
     * not been used for sixty seconds are terminated and removed from
     * the cache. Thus, a pool that remains idle for long enough will
     * not consume any resources. Note that pools with similar
     * properties but different details (for example, timeout parameters)
     * may be created using {@link ThreadPoolExecutor} constructors.
     *
     * @return the newly created thread pool
     */
    public static ExecutorService newSecureCachedThreadPool() {
        return new SecureThreadPoolExecutor(0, Integer.MAX_VALUE,
                                            60L, TimeUnit.SECONDS,
                                            new SynchronousQueue());
    }

    /**
     * Returns a runnable wrapper that ensures that executions
     * (even if performed by different threads) inherit current access control
     * context and {@link ThreadContext}. In particular, the runnable will
     * always run with the same access permissions,
     * {@link DelegatableThreadLocal delegatable thread locals},
     * and context class loader, determined
     * at the time of invocation of this method.
     */
    public static Runnable delegatedRunnable(Runnable runnable) {
        return new DelegatedRunnable(runnable, ThreadContext.getContext());
    }

    /**
     * Returns a runnable wrapper that ensures that executions
     * (even if performed by different threads) inherit current access control
     * context and the specified {@link ThreadContext}. In particular, the
     * runnable will always run with the same access permissions,
     * {@link DelegatableThreadLocal delegatable thread locals},
     * and context class loader, determined
     * at the time of invocation of this method.
     */
    public static Runnable delegatedRunnable(Runnable runnable, ThreadContext tc) {
        return new DelegatedRunnable(runnable, tc);
    }

    /**
     * Returns a callable wrapper that ensures that executions
     * (even if performed by different threads) inherit current access control
     * context and {@link ThreadContext}. In particular, the callable will
     * always run with the same access permissions,
     * {@link DelegatableThreadLocal delegatable thread locals},
     * and context class loader, determined
     * at the time of invocation of this method.
     */
    public static Callable delegatedCallable(Callable callable) {
        return new DelegatedCallable(callable, ThreadContext.getContext());
    }

    /**
     * Returns a callable wrapper that ensures that executions
     * (even if performed by different threads) inherit current access control
     * context and the specified {@link ThreadContext}. In particular, the
     * callable will
     * always run with the same access permissions,
     * {@link DelegatableThreadLocal delegatable thread locals},
     * and context class loader, determined
     * at the time of invocation of this method.
     */
    public static Callable delegatedCallable(Callable callable, ThreadContext tc) {
        return new DelegatedCallable(callable, tc);
    }

    /**
     * Returns a thread factory implementation that attempts to ensure that
     * created threads are equivalent regardless of threads that request
     * creation. Precisely, created threads belong to the same thread group,
     * and they inherit a "parent thread context" (access control context,
     * {@link DelegatableThreadLocal}s, etc.) from the invoker of this method
     * rather than from the invoker of {@link ThreadFactory#newThread}.
     * <p>
     * Use with {@link ThreadPoolExecutor}, to ensure
     * deterministic behavior and consistent security properties.
     * Nevertheless, stronger semantics are often neccessary in
     * security-sensitive applications. In particular,
     * it may be required that worker tasks run within access control context and
     * with delegatable locals of the thread that scheduled the task, which
     * may generally be distinct from both the thread pool creator and the worker
     * thread creator. If such security semantics is needed, use
     * {@link SecureThreadPoolExecutor} or one of "newSecure*"  methods.
     */
    public static ThreadFactory safeThreadFactory() {
        return new SafeThreadFactory();
    }

    /**
     * Creates a successfully completed future with specified result.
     * @param result the result of the future
     * @return the completed future
     */
    public static Future completedFuture(Object result) {
        return completedFuture(result, null);
    }

    /**
     * Creates a successfully completed future with specified result and
     * {@link Callback completion callback}. If not null, the callback is
     * immediately notified about the completion.
     * @param result the result of the future
     * @param cb the callback to be notified about successful completion
     * @return the completed future
     */
    public static Future completedFuture(Object result, Callback cb) {
        if (cb != null) cb.completed(result);
        return new CompletedFuture(result, null);
    }

    /**
     * Creates a failed future with specified failure cause.
     * @param cause the cause of failure
     * @return the completed future
     */
    public static Future failedFuture(Throwable cause) {
        return failedFuture(cause, null);
    }

    /**
     * Creates a failed future with specified failure cause and
     * {@link Callback completion callback}. If not null, the callback is
     * immediately notified about the completion.
     * @param cause the cause of failure
     * @param cb the callback to be notified about failure
     * @return the completed future
     */
    public static Future failedFuture(Throwable cause, Callback cb) {
        if (cb != null) cb.failed(cause);
        return new CompletedFuture(null, cause);
    }

    /**
     * Wrapper for a runnable that ensures that further executions
     * (even if performed by different threads) inherit current access control
     * context and {@link ThreadContext}. In particular, the task will always
     * run with the same access permissions,
     * {@link DelegatableThreadLocal delegatable thread locals},
     * and context class loader, determined
     * at the time of creation of this DelegatedRunnable.
     *
     * @author Dawid Kurzyniec
     * @version 1.0
     */
    static class DelegatedRunnable implements Runnable {

        final Runnable runnable;
        final AccessControlContext acc;
        final PrivilegedAction tcWrappedAction;

        /**
         * Construct new DelegatedRunnable wrapper for a given task.
         * @param runnable the task to wrap
         */
        public DelegatedRunnable(final Runnable runnable, ThreadContext tc) {
            this.runnable = runnable;
            this.acc = AccessController.getContext();
            this.tcWrappedAction = tc.wrap(new PrivilegedAction() {
                public Object run() { runnable.run(); return null; }
            });
        }

        /**
         * Execute wrapped runnable within access control
         * context and {@link ThreadContext} associated to this object at
         * the creation time. In particular, the task will run with the same
         * access permissions,
         * {@link DelegatableThreadLocal delegatable thread locals},
         * and context class loader, as seen when this object was created.
         */
        public void run() {
            AccessController.doPrivileged(tcWrappedAction, acc);
        }

        public String toString() { return runnable.toString(); }
    }

    /**
     * Wrapper for a runnable that ensures that further executions
     * (even if performed by different threads) inherit current access control
     * context and {@link ThreadContext}. In particular, the task will always
     * run with the same access permissions,
     * {@link DelegatableThreadLocal delegatable thread locals},
     * and context class loader, determined
     * at the time of creation of this DelegatedRunnable.
     *
     * @author Dawid Kurzyniec
     * @version 1.0
     */
    static class DelegatedCallable implements Callable {
        final Callable callable;
        final AccessControlContext acc;
        final ThreadContext tc;

        /**
         * Construct new DelegatedCallable wrapper for a given task.
         * @param callable the task to wrap
         */
        public DelegatedCallable(Callable callable, ThreadContext tc) {
            this.callable = callable;
            this.acc = AccessController.getContext();
            this.tc = tc;
        }

        /**
         * Execute wrapped callable within access control
         * context and {@link ThreadContext} associated to this object at
         * the creation time. In particular, the task will run with the same
         * access permissions,
         * {@link DelegatableThreadLocal delegatable thread locals},
         * and context class loader, as seen when this object was created.
         */
        public Object call() throws Exception {
            try {
                return AccessController.doPrivileged(new PrivilegedExceptionAction() {
                    public Object run() throws Exception {
                        return tc.perform(callable);
                    }
                }, acc);
            }
            catch (PrivilegedActionException pae) {
                throw pae.getException();
            }
        }

        public String toString() { return callable.toString(); }
    }

    /**
     * Thread factory implementation that attempts to ensure that created threads
     * are equivalent regardless of threads that request creation. Precisely,
     * created threads belong to the same thread group, and they inherit a
     * "parent thread context" (access control context,
     * {@link DelegatableThreadLocal}s, etc.) from the factory creator rather than
     * from the invoker of {@link #newThread}.
     * <p>
     * When used with {@link ThreadPoolExecutor}, is ensures
     * deterministic behavior and consistent security properties.
     * Nevertheless, stronger semantics are often neccessary in
     * security-sensitive applications. In particular,
     * it may be required that worker tasks run within access control context and
     * with delegatable locals of the thread that scheduled the task, which
     * may generally be distinct from both the thread pool creator and the worker
     * thread creator. If such security semantics is needed, use
     * {@link SecureThreadPoolExecutor}.
     *
     * @author Dawid Kurzyniec
     * @version 1.0
     */

    public static class SafeThreadFactory implements ThreadFactory {
        static final AtomicInteger poolNumber = new AtomicInteger();
        final AtomicInteger threadNumber;
        final AccessControlContext acc;
        final ThreadGroup group;
        final ThreadContext tc;
        final String name;
        int idx = 0;

        /**
         * Creates a new tread factory that creates threads within a default
         * thread group. The group of the current thread (invoking this
         * constructor) is used as a default unless explicitly overridden
         * by current security manager.
         */
        public SafeThreadFactory() {
            this(null, "safe-pool-" +
                       poolNumber.getAndIncrement() +
                       "-thread-");
        }

        /**
         * Creates a new tread factory that uses default thread group and
         * specified thread name prefix. The group of the current thread (invoking
         * this constructor) is used as a default unless explicitly overridden
         * by current security manager.
         *
         * @param name the thread name prefix
         */
        public SafeThreadFactory(String name) {
            this(null, name);
        }

        /**
         * Creates a new tread factory that creates threads within the specified
         * thread group.
         *
         * @param group the thread group for created threads
         */
        public SafeThreadFactory(ThreadGroup group) {
            this(group, null);
        }

        /**
         * Creates a new tread factory that creates threads within the specified
         * thread group, and uses specified thread name prefix, appending
         * consecutive numbers to the names of created threads.
         *
         * @param group the thread group for created threads
         * @param name the thread name prefix
         */
        public SafeThreadFactory(ThreadGroup group, String name) {
            this(group, name, true);
        }

        /**
         * Creates a new tread factory that creates threads within the specified
         * thread group, and uses specified thread name prefix.
         *
         * @param group the thread group for created threads
         * @param name the thread name prefix
         */
        public SafeThreadFactory(ThreadGroup group, String name, boolean count) {
            this.threadNumber = (count ? new AtomicInteger() : null);
            if (group == null) {
                // try to determine tg as in java.lang.Thread
                SecurityManager security = System.getSecurityManager();

                if (security != null) {
                    group = security.getThreadGroup();
                }
            }

            if (group == null) {
                group = Thread.currentThread().getThreadGroup();
            }

            this.group = group;
            this.name = name;
            this.acc = AccessController.getContext();
            this.tc = ThreadContext.getContext();
        }

        public Thread newThread(final Runnable command) {
            return (Thread)AccessController.doPrivileged(tc.wrap(new PrivilegedAction() {
                public Object run() {
                    if (name == null) {
                        return new Thread(group, command);
                    }
                    else {
                        String tname = name;
                        if (threadNumber != null) tname += threadNumber.getAndIncrement();
                        return new Thread(group, command, tname);
                    }
                }
            }), acc);
        }

    }

    final static class CompletedFuture implements Future {
        final Object result;
        final Throwable exception;
        CompletedFuture(Object result, Throwable exception) {
            this.result = result;
            this.exception = exception;
        }

        public Object get() throws ExecutionException {
            if (exception != null) {
                throw new ExecutionException("Task has completed abruptly", exception);
            }
            return result;
        }

        public Object get(long timeout, TimeUnit granularity) throws ExecutionException {
            return get();
        }

        /**
         * Always returns true.
         * @return true
         */
        public boolean isDone() {
            return true;
        }

        /**
         * Always returns false.
         * @return false
         */
        public boolean cancel(boolean mayInterruptIfRunning) {
            return false;
        }

        /**
         * Always returns false.
         * @return false
         */
        public boolean isCancelled() {
            return false;
        }
    }


    /**
      * A wrapper class that exposes only the ExecutorService methods
      * of an implementation.
      */
     static class DelegatedExecutorService extends AbstractExecutorService {
         private final ExecutorService e;
         DelegatedExecutorService(ExecutorService executor) { e = executor; }
         public void execute(Runnable command) { e.execute(command); }
         public void shutdown() { e.shutdown(); }
         public List shutdownNow() { return e.shutdownNow(); }
         public boolean isShutdown() { return e.isShutdown(); }
         public boolean isTerminated() { return e.isTerminated(); }
         public boolean awaitTermination(long timeout, TimeUnit unit)
             throws InterruptedException {
             return e.awaitTermination(timeout, unit);
         }
         public Future submit(Runnable task) {
             return e.submit(task);
         }
         public Future submit(Callable task) {
             return e.submit(task);
         }
         public Future submit(Runnable task, Object result) {
             return e.submit(task, result);
         }
         public List invokeAll(Collection tasks) throws InterruptedException {
             return e.invokeAll(tasks);
         }
         public List invokeAll(Collection tasks, long timeout, TimeUnit unit)
             throws InterruptedException {
             return e.invokeAll(tasks, timeout, unit);
         }
         public Object invokeAny(Collection tasks)
             throws InterruptedException, ExecutionException {
             return e.invokeAny(tasks);
         }
         public Object invokeAny(Collection tasks, long timeout, TimeUnit unit)
             throws InterruptedException, ExecutionException, TimeoutException {
             return e.invokeAny(tasks, timeout, unit);
         }
     }

     /**
      * A wrapper class that exposes only the ExecutorService and
      * ScheduleExecutor methods of a ScheduledExecutorService implementation.
      */
     static class DelegatedScheduledExecutorService
             extends DelegatedExecutorService
             implements ScheduledExecutorService {
         private final ScheduledExecutorService e;
         DelegatedScheduledExecutorService(ScheduledExecutorService executor) {
             super(executor);
             e = executor;
         }
         public ScheduledFuture schedule(Runnable command, long delay,  TimeUnit unit) {
             return e.schedule(command, delay, unit);
         }
         public ScheduledFuture schedule(Callable callable, long delay, TimeUnit unit) {
             return e.schedule(callable, delay, unit);
         }
         public ScheduledFuture scheduleAtFixedRate(Runnable command, long initialDelay,  long period, TimeUnit unit) {
             return e.scheduleAtFixedRate(command, initialDelay, period, unit);
         }
         public ScheduledFuture scheduleWithFixedDelay(Runnable command, long initialDelay,  long delay, TimeUnit unit) {
             return e.scheduleWithFixedDelay(command, initialDelay, delay, unit);
         }
     }
}
