root/trunk/tango/scrapple/thread/ThreadPool.d

/++
    author: Jeff Davey <jeffd@gwava.com>
    license: BSD style

    A basic thread queueing module.

    This specific module pre-detaches all threads up to the size of the thread pool. The idea is that state
    can then be saved on each thread's TLS for caching and other purposes.

    All threads are set to "detach" mode, so that you can close the running program and let the OS clean up
    any thread resources.

    There is also a close function, which specifically waits for all detached threads to move to a closed state
    before returning.

    Usage is fairly straightforward, and both delegates and functions are supported:

    Example:
    ---
    uint a = 0;
    Mutex m = new Mutex();

    void func(void *arg)
    {           
        char[] name = Thread.getThis().name;
        int *data = cast(int *)arg;
        m.lock();
        a++;
        m.unlock();
        Thread.sleep(1);
        return 0;
    }


    ThreadPool pool = new ThreadPool(100, &func);
    if (pool)
    {
        for (uint i = 0; i < 1000; i++)
        {
            int *data = new int();
            *data = i;
            pool.start(cast(void*)(data));
        }
    }
    pool.close();
    assert(a == 1000);
    ---
++/

module tango.scrapple.thread.ThreadPool;

private import tango.core.Thread;
private import tango.core.sync.Mutex;
private import tango.core.sync.Condition;

private class PooledThread
{
    enum State
    {
        IDLE,
        RUNNING,
        SHUTDOWN,
        CLOSED
    };

    this(void delegate(void *) dg, size_t stackSize = 0)
    {
        this._dg = dg;
        this(stackSize);
    }

    this(void (*func)(void *arg), size_t stackSize = 0)
    {
        this._func = func;
        this(stackSize);
    }

    State state()
    {
        return this._state;
    }

    State state(State newState)
    {
        return this._state = newState;
    }

    void join()
    {
        _thread.join();
    }

    void lock()
    {
        _lock.lock();
    }

    bool tryLock()
    {
        return _lock.tryLock();
    }

    void unlock()
    {
        _lock.unlock();
    }

    void* arg()
    {
        return this._arg;
    }

    void* arg(void* newArg)
    {
        return this._arg = newArg;
    }
    
    void notify()
    {
        _signal.notify();
    }

    private:
    Mutex _lock;                        // for condition/mutex pair
    Condition _signal;                  // used to signal work available
    void delegate(void *) _dg = null;   // delegate to call
    void (*_func)(void *) = null;       // or functiont o call
    void* _arg = null;                  // Arg that has been passed
    Thread _thread;                     // Underlying Thread class
    State _state = State.CLOSED;        // Our current state.

    
    void run()
    {
        _lock.lock();
        scope (exit) _lock.unlock();

        for(;;)
        {
            this._state = State.IDLE;
            _signal.wait(); // wait for signal from pool
            if (this._state == State.SHUTDOWN)  // shutting down?
            {
                this._state = State.CLOSED;
                break;
            }

            if (this._state == State.RUNNING)   // let's run the work.
            {
                if (_dg)
                    _dg(_arg);
                else if (_func)
                    _func(_arg);
            }
        }

        return 0;
    }

    this(size_t stackSize)
    {
        this._lock = new Mutex();
        this._signal = new Condition(this._lock);

        _thread = new Thread(&run, stackSize);
        _thread.isDaemon = true;
        _thread.start();
    }

}

/++
    ThreadPool is the thread queueing class.
++/

class ThreadPool
{

    /++
        Construct a ThreadPool. 

        Params:
        poolSize            =   The size of the thread pool. It will 
                                actually create this many threads on startup.

        (function/delegate) =   void func(void *arg) argument is the passed
                                argument from the start function.

        stackSize           =   Size of the stack per thread. As tango
                                doesn't quite support this functionality,
                                it's not working quite yet.
    ++/ 
    this(uint poolSize, void (*func)(void *), size_t stackSize = 0)
    in
    {
        assert(func);
    }
    body
    {
        this._func = func;
        this(poolSize, stackSize);
    }

    this(uint poolSize, void delegate(void *) dg, size_t stackSize = 0)
    in
    {
        assert(dg);
    }
    body
    {
        this._dg = dg;
        this(poolSize, stackSize);      
    }


    /++
        Start a unit of work.

        Params:
        arg                 =   A void * arg that will be passed to the
                                underlying worker thread.

        block               =   If true, the start function will wait
                                for an existing thread to become 
                                available, if all are being used. If
                                false, the start function will return 
                                immediately with a value of false if it
                                could not schedule work, or a vlue of true
                                if it could.
    
    ++/
    bool start(void *arg, bool block = true)
    {
        bool rtn = false;
        for(;;)
        {
            uint i = 0;
            for (; i < _pool.length; i++)
            {
                if (_pool[i].tryLock())
                {
                    scope(exit) _pool[i].unlock();
                    if (_pool[i].state == PooledThread.State.IDLE)
                    {
                        int *data = cast(int *)arg;
                        _pool[i].state = PooledThread.State.RUNNING;
                        _pool[i].arg = arg;
                        _pool[i].notify();
                        rtn = true;
                        break;
                    }
                }
            }

            if (i == _pool.length)
            {
                if (block)
                    Thread.sleep(0.1);
                else
                    break;
            }
            else
                break;
        }

        return rtn;
    }


    /++
      Close the thread pool. This will go through all running threads,
      tell them to shutdown, and wait for that to happen before returning.
    
      Currently, the poll time between thread checks is hard-coded to 100ms
    ++/
    void close()
    {
        if (open)
        {
            for(;;)
            {
                bool allClosed = true;
                for (uint i = 0; i < _pool.length; i++)
                {
                    _pool[i].lock();
                    if (_pool[i].state == PooledThread.State.IDLE)
                    {
                        _pool[i].state = PooledThread.State.SHUTDOWN;       
                        _pool[i].notify();
                    }   
                    if (_pool[i].state != PooledThread.State.CLOSED)
                        allClosed = false;                                  
                    _pool[i].unlock();
                }
                if (allClosed)
                    break;
                else
                    Thread.sleep(.1);
            }
            open = false;
        }
    }


    private:
    void delegate(void *) _dg = null;
    void (*_func)(void *) = null;
    PooledThread[] _pool = null;
    bool open = false;

    this(uint poolSize, size_t stackSize)
    {
        _pool = new PooledThread[poolSize];

        for (uint i = 0; i < poolSize; i++)
        {
            if (_func)
                _pool[i] = new PooledThread(_func, stackSize);
            else
                _pool[i] = new PooledThread(_dg, stackSize);
        }
        open = true;
    }
}

version (Test)
{
    import tetra.util.Test;
    unittest
    {   
        Test.Status basicFunctionality(inout char[][] messages)
        {
            uint a = 0;
            Mutex m = new Mutex();
            void func(void *arg)
            {           
                char[] name = Thread.getThis().name;
                int *data = cast(int *)arg;
                m.lock();
                a++;
                m.unlock();
                Thread.sleep(1);
                return 0;
            }


            ThreadPool pool = new ThreadPool(100, &func);
            if (pool)
            {
                for (uint i = 0; i < 1000; i++)
                {
                    int *data = new int();
                    *data = i;
                    pool.start(cast(void*)(data));
                }
            }
            pool.close();

            if (a == 1000)
                return Test.Status.Success;
            return Test.Status.Failure;
        }

        auto t = new Test("tetra.util.ThreadPool");
        t["Basic Functionality"] = &basicFunctionality;
        t.run();
    }
}