tthreadpool.hh

线程池 线程池 线程池

#ifndef __TTHREADPOOL_HH
#define __TTHREADPOOL_HH

//
// Project : threadpool
// File    : thread_pool.hh
// Purpose : class for managing a pool of threads
//
// Copyright (C) 2003 - Ronald Kriemann
//
// This library is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 2.1 of the License, or (at your option) any later version.
//
// This program 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 Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public
// License along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
//

#include "TArray.hh"
#include "TSLL.hh"
#include "TThread.hh"

//
// takes jobs and executes them in threads
//
class TThreadPool
{
public:
    /////////////////////////////////////
    //
    // local types
    //

    ///////////////////////////////////////////
    //
    // class for a job in the pool
    //

    class TPoolThr;

    class TJob
    {
    protected:
        // number of processor this job was assigned to
        int         _job_no;

        // associated thread
        TPoolThr  * _pool_thr;
        
    public:
        // constructor
        TJob ( int n ) : _job_no(n), _pool_thr(NULL) {}
        
        // running method
        virtual void run ( void * ptr ) = 0;
        
        // access processor-number
        int        job_no   () const { return _job_no; }
        TPoolThr * pool_thr ()       { return _pool_thr; }

        void set_pool_thr ( TPoolThr * t ) { _pool_thr = t; }
        
        // compare if given proc-no is local one
        bool on_proc ( int p ) const { return ((p == -1) || (_job_no == -1) || (p == _job_no)); }
    };

    //
    // thread handled by threadpool
    //

    class TPoolThr : public TThread
    {
    protected:
        // pool we are in
        TThreadPool    * _pool;

        // job to run and data for it
        TJob           * _job;
        void           * _data_ptr;
        
        // condition and mutex for job-waiting and pool-synchronisation
        TCondition       _work_cond,  _sync_cond;
        TMutex           _work_mutex, _sync_mutex;

        // indicates work-in-progress, end-of-thread and a mutex for this
        bool             _work, _end;
        TMutex           _var_mutex;

        TMutex           _del_mutex;
        
    public:
        // constructor
        TPoolThr ( int n, TThreadPool * p )
                : TThread(n), _pool(p), _job(NULL), _data_ptr(NULL),
                  _work(false), _end(false)
        {}

        ~TPoolThr ();
        
        // running method
        void run ();

        void set_end  ( bool   f ) { _var_mutex.lock(); _end  = f; _var_mutex.unlock(); }
        void set_work ( bool   f ) { _var_mutex.lock(); _work = f; _var_mutex.unlock(); }
        void set_job  ( TJob * j,
                        void * p ) { _var_mutex.lock(); _job  = j; _data_ptr = p; _var_mutex.unlock(); }

        bool is_working () const { return _job != NULL; }

        TJob        * job        () { return _job; }
        
        TCondition  & work_cond  () { return _work_cond; }
        TMutex      & work_mutex () { return _work_mutex; }

        TCondition  & sync_cond  () { return _sync_cond; }
        TMutex      & sync_mutex () { return _sync_mutex; }
    };
    
protected:
    // maximum degree of parallelism
    uint                  _max_parallel;

    // array of threads, handled by pool
    TArray< TPoolThr * >  _threads;
    
    // list of idle threads
    TSLL< TPoolThr * >    _idle_threads;

    // mutex for accessing data
    TMutex                _idle_mutex, _list_mutex;

    // semaphore for accessing idle-list
    TCondition            _idle_cond;
    
public:
    ///////////////////////////////////////////////
    //
    // constructor and destructor
    //

    TThreadPool ( uint max_p );

    ~TThreadPool ();

    ///////////////////////////////////////////////
    //
    // access local variables
    //

    uint max_parallel () const { return _max_parallel; }
    
    ///////////////////////////////////////////////
    //
    // run, stop and synch with job
    //

    void  run  ( TJob * job, void * ptr = NULL );
    void  sync ( TJob * job );
    
    void  sync_all ();

    ///////////////////////////////////////////////
    //
    // manage pool threads
    //

    // return idle thread form pool
    TPoolThr * get_idle ();

    // insert idle thread into pool
    void append_idle ( TPoolThr * t );
};

//
// global thread-pool
//

extern TThreadPool * thread_pool;

//
// init global thread_pool
//

extern int init_thread_pool ( uint p );

#endif  // __TTHREADPOOL_HH