/*
 * @(#)ThreadPool.java      0.9.0 06/04/2000 - 15:13:54
 *
 * Copyright (C) 2000,,2003 2002 Matt Albrecht
 * groboclown@users.sourceforge.net
 * http://groboutils.sourceforge.net
 *
 *  Permission is hereby granted, free of charge, to any person obtaining a
 *  copy of this software and associated documentation files (the "Software"),
 *  to deal in the Software without restriction, including without limitation
 *  the rights to use, copy, modify, merge, publish, distribute, sublicense,
 *  and/or sell copies of the Software, and to permit persons to whom the 
 *  Software is furnished to do so, subject to the following conditions:
 *
 *  The above copyright notice and this permission notice shall be included in 
 *  all copies or substantial portions of the Software. 
 *
 *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 
 *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 
 *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL 
 *  THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 
 *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING 
 *  FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER 
 *  DEALINGS IN THE SOFTWARE.
 */

package net.sourceforge.groboutils.util.thread.v1;


import net.sourceforge.groboutils.util.datastruct.v1.SynchQueue;

import java.util.Vector;


/**
 * A pool of QueueThread instances, each containing an instance of
 * an ObjectListener implemented class.  The Class to be the listener
 * is passed into the constructor.  Requirements for the Class are:
 * 1. it implements QueueThread.ObjectListener, 2. it has a public
 * constructor without any parameters.
 * <P>
 * The pool handles menial tasks such as:
 * <ol>
 *      <li>Growing the thread pool if the number of waiting objects
 *          is above a threshold number, up to a maximum number of
 *          threads.
 *      <li>Finding the thread with the fewest number of waiting objects.
 *      <li>Optimization of determining which thread to pass events to.
 * </ol>
 * <P>
 * The pool gets much of its functionality by sharing a single SynchQueue
 * between all of its threads.
 *
 * @author   Matt Albrecht <a href="mailto:groboclown@users.sourceforge.net">groboclown@users.sourceforge.net</a>
 * @since    June 4, 2000
 * @version  $Date: 2003/02/10 22:52:49 $
 */
public class ThreadPool
{
    //----------------------------
    // Public data
    
    
    //----------------------------
    // Private data
    
    private Class m_objListenerClass = null;
    private Object m_objListenerInitData = null;
    
    private QueueThread[] m_pool = null;
    private SynchQueue m_sharedQueue = new SynchQueue();
    
    private int m_maxThreads = 10;
    private int m_numThreads = 0;
    private int m_depthThreshold = 5;
    
    //----------------------------
    // constructors
    
    /**
     * Default constructor
     */
    public ThreadPool( Class objectListenerClass )
    {
        this( objectListenerClass, null, 1, 10 );
    }
    
    
    /**
     * 
     */
    public ThreadPool( Class objectListenerClass, int maxThreads )
    {
        this( objectListenerClass, null, 1, maxThreads );
    }
    
    
    /**
     *
     * @param initData if the given objectListenerClass is an instance
     *      of ThreadObjectListener, then the initData will be passed
     *      into the initialize( Object ) method.
     */
    public ThreadPool( Class objectListenerClass, Object initData )
    {
        this( objectListenerClass, initData, 1, 10 );
    }
    
    
    /**
     * 
     */
    public ThreadPool( Class objectListenerClass, Object initData,
            int maxThreads )
    {
        this( objectListenerClass, initData, 1, maxThreads );
    }
    
    
    /**
     * 
     */
    public ThreadPool( Class objectListenerClass, Object initData,
            int startingThreadCount, int maxThreads )
    {
        this.m_objListenerClass = objectListenerClass;
        this.m_objListenerInitData = initData;
        try
        {
            createObjectListenerInstance();
        }
        catch (Exception ex)
        {
ex.printStackTrace();
            throw new IllegalArgumentException( "Class "+objectListenerClass+
                " does not create ObjectListener instances");
        }
        
        setMaximumThreadCount( maxThreads );
        this.m_pool = new QueueThread[ maxThreads ];
        while (this.m_numThreads < startingThreadCount)
        {
            addNewThread();
        }
    }
    
    
    //----------------------------
    // Public methods
    
    /**
     * 
     */
    public void setDepthThreshold( int threshold )
    {
        if (threshold < 1)
        {
            throw new IllegalArgumentException("threshold "+threshold+
                " is too low");
        }
        this.m_depthThreshold = threshold;
    }
    
    /**
     * 
     */
    public int getObjectDepth()
    {
        return this.m_sharedQueue.size();
    }
    
    /**
     * Adds the given object into the shared queue, so that the next
     * available thread will process it.
     */
    public void addObject( Object o )
    {
        checkThreshold();
        this.m_sharedQueue.enqueue( o );
    }
    
    /**
     * 
     */
    public int getThreadCount()
    {
        return this.m_numThreads;
    }
    
    /**
     * 
     */
    public int getMaximumThreadCount()
    {
        return this.m_maxThreads;
    }
    
    /**
     * 
     */
    public void setMaximumThreadCount( int max )
    {
        if (max < 1)
        {
            throw new IllegalArgumentException("maximum count "+max+
                " is out of bounds" );
        }
        this.m_maxThreads = max;
    }
    
    
    /**
     * Waits for all expecting objects in the queue to be processed,
     * and for each thread to finish processing an object.
     */
    public void waitForThreadsToFinish()
    {
        // wait for the SynchQueue to empty
        while (getObjectDepth() > 0)
        {
            Thread.yield();
        }
        
        Vector v = new Vector();
        synchronized (v)
        {
            // find all threads which are still processing objects
            for (int i = this.m_numThreads; --i >= 0;)
            {
                if (this.m_pool[i].isProcessingObjects())
                {
                    v.addElement( this.m_pool[i] );
                }
            }
            
            // wait for all threads to finish processing their objects
            QueueThread qt;
            while (v.size() > 0)
            {
                Thread.yield();
                for (int i = v.size(); --i >= 0;)
                {
                    qt = (QueueThread)v.elementAt(i);
                    if (!qt.isProcessingObjects())
                    {
                        v.removeElementAt(i);
                        // don't need to adjust i because
                        // we're procressing backwards.
                    }
                }
            }
        }
    }
    
    /**
     * Stops all threads.
     */
    public synchronized void stopThreads()
    {
        for (int i = this.m_numThreads; --i >= 0;)
        {
            if (this.m_pool[i] != null)
                this.m_pool[i].stop();
        }
    }
    
    /**
     * Suspends all threads.
     */
    public synchronized void suspendThreads()
    {
        for (int i = this.m_numThreads; --i >= 0;)
        {
            if (this.m_pool[i] != null)
                this.m_pool[i].suspend();
        }
    }
    
    /**
     * Resumes all threads.
     */
    public synchronized void resumeThreads()
    {
        for (int i = this.m_numThreads; --i >= 0;)
        {
            if (this.m_pool[i] != null)
                this.m_pool[i].resume();
        }
    }
    
    
    
    
    //----------------------------
    // Protected methods

    /**
     * If there are not enough threads, then add one into the
     * internal array, start the thread, and return the created
     * thread.
     *
     * @return the new thread, or <tt>null</tt> if the pool has
     *      exceeded its maximum thread count.
     */
    protected synchronized QueueThread addNewThread()
    {
        QueueThread qt = null;
        if (this.m_numThreads < this.m_maxThreads)
        {
            qt = this.m_pool[ this.m_numThreads++ ] =
                new QueueThread( createObjectListenerInstance(),
                this.m_sharedQueue );
            qt.start();
        }
        return qt;
    }
    
    
    /**
     * Checks if the depth on the shared queue is too deep (beyond the
     * threshold), and if so, creates a new thread to help deal with the
     * situation.
     */
    protected void checkThreshold()
    {
        if (this.m_sharedQueue.size() > this.m_depthThreshold)
        {
            addNewThread();
        }
    }
    
    /**
     * Create an instance of the basic object listener class, as given
     * in the constructor.
     *
     * @exception IllegalStateException thrown if there is an error
     *      creating a new instance of the class.
     */
    protected IObjectListener createObjectListenerInstance()
    {
        try
        {
//System.out.println("Creating an instance of class "+this.m_objListenerClass+
//", modifiers = "+(java.lang.reflect.Modifier.toString(
//this.m_objListenerClass.getModifiers())));
            IObjectListener ol = (IObjectListener)
                this.m_objListenerClass.newInstance();
            if (ol instanceof IThreadObjectListener)
            {
                ((IThreadObjectListener)ol).initialize(
                    this.m_objListenerInitData );
            }
            return ol;
        }
        catch (InstantiationException ie)
        {
            throw new IllegalStateException("could not instantiate from class "+
                this.m_objListenerClass.getName()+
                ": general instantiation exception "+ie.getMessage());
        }
        catch (IllegalAccessException iae)
        {
            throw new IllegalStateException("could not instantiate from class "+
                this.m_objListenerClass.getName()+
                ": could not access constructor "+iae.getMessage());
        }
        catch (ClassCastException cce)
        {
            throw new IllegalStateException("could not instantiate from class "+
                this.m_objListenerClass.getName()+": instance of wrong type "+
                cce.getMessage());
        }
    }
    
    //----------------------------
    // Private methods
}