/*

 * Copyright (c) 2003, 2015, Oracle and/or its affiliates. All rights reserved.

 * 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.  Oracle designates this

 * particular file as subject to the "Classpath" exception as provided

 * by Oracle in the LICENSE file that accompanied this code.

 *

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

 */

package com.sun.corba.se.impl.orbutil.threadpool;

import java.io.IOException;

import java.io.Closeable;

import java.security.AccessController;

import java.security.PrivilegedAction;

import java.util.List;

import java.util.ArrayList;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.concurrent.atomic.AtomicLong;

import com.sun.corba.se.spi.orbutil.threadpool.NoSuchWorkQueueException;

import com.sun.corba.se.spi.orbutil.threadpool.ThreadPool;

import com.sun.corba.se.spi.orbutil.threadpool.Work;

import com.sun.corba.se.spi.orbutil.threadpool.WorkQueue;

import com.sun.corba.se.impl.orbutil.ORBConstants;

import com.sun.corba.se.impl.orbutil.threadpool.WorkQueueImpl;

import com.sun.corba.se.spi.monitoring.MonitoringConstants;

import com.sun.corba.se.spi.monitoring.MonitoredObject;

import com.sun.corba.se.spi.monitoring.MonitoringFactories;

import com.sun.corba.se.spi.orb.ORB;

import com.sun.corba.se.spi.monitoring.LongMonitoredAttributeBase;

import com.sun.corba.se.impl.logging.ORBUtilSystemException;

import com.sun.corba.se.impl.orbutil.ORBConstants;

import com.sun.corba.se.spi.logging.CORBALogDomains;

public class ThreadPoolImpl implements ThreadPool

{

    // serial counter useful for debugging

    private static AtomicInteger threadCounter = new AtomicInteger(0);

    private static final ORBUtilSystemException wrapper =

        ORBUtilSystemException.get(CORBALogDomains.RPC_TRANSPORT);

    // Any time currentThreadCount and/or availableWorkerThreads is updated

    // or accessed this ThreadPool's WorkQueue must be locked. And, it is

    // expected that this ThreadPool's WorkQueue is the only object that

    // updates and accesses these values directly and indirectly though a

    // call to a method in this ThreadPool. If any call to update or access

    // those values must synchronized on this ThreadPool's WorkQueue.

    private WorkQueue workQueue;

    // Stores the number of available worker threads

    private int availableWorkerThreads = 0;

    // Stores the number of threads in the threadpool currently

    private int currentThreadCount = 0;

    // Minimum number of worker threads created at instantiation of the threadpool

    private int minWorkerThreads = 0;

    // Maximum number of worker threads in the threadpool

    private int maxWorkerThreads = 0;

    // Inactivity timeout value for worker threads to exit and stop running

    private long inactivityTimeout;

    // Indicates if the threadpool is bounded or unbounded

    private boolean boundedThreadPool = false;

    // Running count of the work items processed

    // Set the value to 1 so that divide by zero is avoided in

    // averageWorkCompletionTime()

    private AtomicLong processedCount = new AtomicLong(1);

    // Running aggregate of the time taken in millis to execute work items

    // processed by the threads in the threadpool

    private AtomicLong totalTimeTaken = new AtomicLong(0);

    // Name of the ThreadPool

    private String name;

    // MonitoredObject for ThreadPool

    private MonitoredObject threadpoolMonitoredObject;

    // ThreadGroup in which threads should be created

    private ThreadGroup threadGroup;

    Object workersLock = new Object();

    List<WorkerThread> workers = new ArrayList<>();

    /**

     * This constructor is used to create an unbounded threadpool

     */

    public ThreadPoolImpl(ThreadGroup tg, String threadpoolName) {

        inactivityTimeout = ORBConstants.DEFAULT_INACTIVITY_TIMEOUT;

        maxWorkerThreads = Integer.MAX_VALUE;

        workQueue = new WorkQueueImpl(this);

        threadGroup = tg;

        name = threadpoolName;

        initializeMonitoring();

    }

    /**

     * This constructor is used to create an unbounded threadpool

     * in the ThreadGroup of the current thread

     */

    public ThreadPoolImpl(String threadpoolName) {

        this( Thread.currentThread().getThreadGroup(), threadpoolName ) ;

    }

    /**

     * This constructor is used to create bounded threadpool

     */

    public ThreadPoolImpl(int minSize, int maxSize, long timeout,

                                            String threadpoolName)

    {

        minWorkerThreads = minSize;

        maxWorkerThreads = maxSize;

        inactivityTimeout = timeout;

        boundedThreadPool = true;

        workQueue = new WorkQueueImpl(this);

        name = threadpoolName;

        for (int i = 0; i < minWorkerThreads; i++) {

            createWorkerThread();

        }

        initializeMonitoring();

    }

    // Note that this method should not return until AFTER all threads have died.

    public void close() throws IOException {

        // Copy to avoid concurrent modification problems.

        List<WorkerThread> copy = null;

        synchronized (workersLock) {

            copy = new ArrayList<>(workers);

        }

        for (WorkerThread wt : copy) {

            wt.close();

            while (wt.getState() != Thread.State.TERMINATED) {

                try {

                    wt.join();

                } catch (InterruptedException exc) {

                    wrapper.interruptedJoinCallWhileClosingThreadPool(exc, wt, this);

                }

            }

        }

        threadGroup = null;

    }

    // Setup monitoring for this threadpool

    private void initializeMonitoring() {

        // Get root monitored object

        MonitoredObject root = MonitoringFactories.getMonitoringManagerFactory().

                createMonitoringManager(MonitoringConstants.DEFAULT_MONITORING_ROOT, null).

                getRootMonitoredObject();

        // Create the threadpool monitoring root

        MonitoredObject threadPoolMonitoringObjectRoot = root.getChild(

                    MonitoringConstants.THREADPOOL_MONITORING_ROOT);

        if (threadPoolMonitoringObjectRoot == null) {

            threadPoolMonitoringObjectRoot =  MonitoringFactories.

                    getMonitoredObjectFactory().createMonitoredObject(

                    MonitoringConstants.THREADPOOL_MONITORING_ROOT,

                    MonitoringConstants.THREADPOOL_MONITORING_ROOT_DESCRIPTION);

            root.addChild(threadPoolMonitoringObjectRoot);

        }

        threadpoolMonitoredObject = MonitoringFactories.

                    getMonitoredObjectFactory().

                    createMonitoredObject(name,

                    MonitoringConstants.THREADPOOL_MONITORING_DESCRIPTION);

        threadPoolMonitoringObjectRoot.addChild(threadpoolMonitoredObject);

        LongMonitoredAttributeBase b1 = new

            LongMonitoredAttributeBase(MonitoringConstants.THREADPOOL_CURRENT_NUMBER_OF_THREADS,

                    MonitoringConstants.THREADPOOL_CURRENT_NUMBER_OF_THREADS_DESCRIPTION) {

                public Object getValue() {

                    return new Long(ThreadPoolImpl.this.currentNumberOfThreads());

                }

            };

        threadpoolMonitoredObject.addAttribute(b1);

        LongMonitoredAttributeBase b2 = new

            LongMonitoredAttributeBase(MonitoringConstants.THREADPOOL_NUMBER_OF_AVAILABLE_THREADS,

                    MonitoringConstants.THREADPOOL_CURRENT_NUMBER_OF_THREADS_DESCRIPTION) {

                public Object getValue() {

                    return new Long(ThreadPoolImpl.this.numberOfAvailableThreads());

                }

            };

        threadpoolMonitoredObject.addAttribute(b2);

        LongMonitoredAttributeBase b3 = new

            LongMonitoredAttributeBase(MonitoringConstants.THREADPOOL_NUMBER_OF_BUSY_THREADS,

                    MonitoringConstants.THREADPOOL_NUMBER_OF_BUSY_THREADS_DESCRIPTION) {

                public Object getValue() {

                    return new Long(ThreadPoolImpl.this.numberOfBusyThreads());

                }

            };

        threadpoolMonitoredObject.addAttribute(b3);

        LongMonitoredAttributeBase b4 = new

            LongMonitoredAttributeBase(MonitoringConstants.THREADPOOL_AVERAGE_WORK_COMPLETION_TIME,

                    MonitoringConstants.THREADPOOL_AVERAGE_WORK_COMPLETION_TIME_DESCRIPTION) {

                public Object getValue() {

                    return new Long(ThreadPoolImpl.this.averageWorkCompletionTime());

                }

            };

        threadpoolMonitoredObject.addAttribute(b4);

        LongMonitoredAttributeBase b5 = new

            LongMonitoredAttributeBase(MonitoringConstants.THREADPOOL_CURRENT_PROCESSED_COUNT,

                    MonitoringConstants.THREADPOOL_CURRENT_PROCESSED_COUNT_DESCRIPTION) {

                public Object getValue() {

                    return new Long(ThreadPoolImpl.this.currentProcessedCount());

                }

            };

        threadpoolMonitoredObject.addAttribute(b5);

        // Add the monitored object for the WorkQueue

        threadpoolMonitoredObject.addChild(

                ((WorkQueueImpl)workQueue).getMonitoredObject());

    }

    // Package private method to get the monitored object for this

    // class

    MonitoredObject getMonitoredObject() {

        return threadpoolMonitoredObject;

    }

    public WorkQueue getAnyWorkQueue()

    {

        return workQueue;

    }

    public WorkQueue getWorkQueue(int queueId)

        throws NoSuchWorkQueueException

    {

        if (queueId != 0)

            throw new NoSuchWorkQueueException();

        return workQueue;

    }

    /**

     * To be called from the workqueue when work is added to the

     * workQueue. This method would create new threads if required

     * or notify waiting threads on the queue for available work

     */

    void notifyForAvailableWork(WorkQueue aWorkQueue) {

        synchronized (aWorkQueue) {

            if (availableWorkerThreads < aWorkQueue.workItemsInQueue()) {

                createWorkerThread();

            } else {

                aWorkQueue.notify();

            }

        }

    }

    private Thread createWorkerThreadHelper( String name ) {

        // Thread creation needs to be in a doPrivileged block

        // if there is a non-null security manager for two reasons:

        // 1. The creation of a thread in a specific ThreadGroup

        //    is a privileged operation.  Lack of a doPrivileged

        //    block here causes an AccessControlException

        //    (see bug 6268145).

        // 2. We want to make sure that the permissions associated

        //    with this thread do NOT include the permissions of

        //    the current thread that is calling this method.

        //    This leads to problems in the app server where

        //    some threads in the ThreadPool randomly get

        //    bad permissions, leading to unpredictable

        //    permission errors (see bug 6021011).

        //

        //    A Java thread contains a stack of call frames,

        //    one for each method called that has not yet returned.

        //    Each method comes from a particular class.  The class

        //    was loaded by a ClassLoader which has an associated

        //    CodeSource, and this determines the Permissions

        //    for all methods in that class.  The current

        //    Permissions for the thread are the intersection of

        //    all Permissions for the methods on the stack.

        //    This is part of the Security Context of the thread.

        //

        //    When a thread creates a new thread, the new thread

        //    inherits the security context of the old thread.

        //    This is bad in a ThreadPool, because different

        //    creators of threads may have different security contexts.

        //    This leads to occasional unpredictable errors when

        //    a thread is re-used in a different security context.

        //

        //    Avoiding this problem is simple: just do the thread

        //    creation in a doPrivileged block.  This sets the

        //    inherited security context to that of the code source

        //    for the ORB code itself, which contains all permissions

        //    in either Java SE or Java EE.

        WorkerThread thread = new WorkerThread(threadGroup, name);

        synchronized (workersLock) {

            workers.add(thread);

        }

        // The thread must be set to a daemon thread so the

        // VM can exit if the only threads left are PooledThreads

        // or other daemons.  We don't want to rely on the

        // calling thread always being a daemon.

        // Note that no exception is possible here since we

        // are inside the doPrivileged block.

        thread.setDaemon(true);

        wrapper.workerThreadCreated(thread, thread.getContextClassLoader());

        thread.start();

        return null;

    }

    /**

     * To be called from the workqueue to create worker threads when none

     * available.

     */

    void createWorkerThread() {

        final String name = getName();

        synchronized (workQueue) {

            try {

                if (System.getSecurityManager() == null) {

                    createWorkerThreadHelper(name);

                } else {

                    // If we get here, we need to create a thread.

                    AccessController.doPrivileged(

                            new PrivilegedAction() {

                        public Object run() {

                            return createWorkerThreadHelper(name);

                        }

                    }

                    );

                }

            } catch (Throwable t) {

                // Decrementing the count of current worker threads.

                // But, it will be increased in the finally block.

                decrementCurrentNumberOfThreads();

                wrapper.workerThreadCreationFailure(t);

            } finally {

                incrementCurrentNumberOfThreads();

            }

        }

    }

    public int minimumNumberOfThreads() {

        return minWorkerThreads;

    }

    public int maximumNumberOfThreads() {

        return maxWorkerThreads;

    }

    public long idleTimeoutForThreads() {

        return inactivityTimeout;

    }

    public int currentNumberOfThreads() {

        synchronized (workQueue) {

            return currentThreadCount;

        }

    }

    void decrementCurrentNumberOfThreads() {

        synchronized (workQueue) {

            currentThreadCount--;

        }

    }

    void incrementCurrentNumberOfThreads() {

        synchronized (workQueue) {

            currentThreadCount++;

        }

    }

    public int numberOfAvailableThreads() {

        synchronized (workQueue) {

            return availableWorkerThreads;

        }

    }

    public int numberOfBusyThreads() {

        synchronized (workQueue) {

            return (currentThreadCount - availableWorkerThreads);

        }

    }

    public long averageWorkCompletionTime() {

        synchronized (workQueue) {

            return (totalTimeTaken.get() / processedCount.get());

        }

    }

    public long currentProcessedCount() {

        synchronized (workQueue) {

            return processedCount.get();

        }

    }

    public String getName() {

        return name;

    }

    /**

    * This method will return the number of WorkQueues serviced by the threadpool.

    */

    public int numberOfWorkQueues() {

        return 1;

    }

    private static synchronized int getUniqueThreadId() {

        return ThreadPoolImpl.threadCounter.incrementAndGet();

    }

    /**

     * This method will decrement the number of available threads

     * in the threadpool which are waiting for work. Called from

     * WorkQueueImpl.requestWork()

     */

    void decrementNumberOfAvailableThreads() {

        synchronized (workQueue) {

            availableWorkerThreads--;

        }

    }

    /**

     * This method will increment the number of available threads

     * in the threadpool which are waiting for work. Called from

     * WorkQueueImpl.requestWork()

     */

    void incrementNumberOfAvailableThreads() {

        synchronized (workQueue) {

            availableWorkerThreads++;

        }

    }

    {

        private Work currentWork;

        private int threadId = 0; // unique id for the thread

        private volatile boolean closeCalled = false;

        private String threadPoolName;

        // name seen by Thread.getName()

        private StringBuffer workerThreadName = new StringBuffer();

        WorkerThread(ThreadGroup tg, String threadPoolName) {

            super(tg, "Idle");

            this.threadId = ThreadPoolImpl.getUniqueThreadId();

            this.threadPoolName = threadPoolName;

            setName(composeWorkerThreadName(threadPoolName, "Idle"));

        }

        public synchronized void close() {

            closeCalled = true;

            interrupt();

        }

        private void resetClassLoader() {

        }

        private void performWork() {

            long start = System.currentTimeMillis();

            try {

                currentWork.doWork();

            } catch (Throwable t) {

                wrapper.workerThreadDoWorkThrowable(this, t);

            }

            long elapsedTime = System.currentTimeMillis() - start;

            totalTimeTaken.addAndGet(elapsedTime);

            processedCount.incrementAndGet();

        }

        public void run() {

            try  {

                while (!closeCalled) {

                    try {

                        currentWork = ((WorkQueueImpl)workQueue).requestWork(

                            inactivityTimeout);

                        if (currentWork == null)

                            continue;

                    } catch (InterruptedException exc) {

                        wrapper.workQueueThreadInterrupted( exc, getName(),

                           Boolean.valueOf(closeCalled));