tthread.cc

线程池 线程池 线程池

//
// Project   : threadpool
// File      : TThread.cc
// Author    : Ronald Kriemann
// Purpose   : baseclass for a thread-able class
//
// 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 <unistd.h>
#include <signal.h>
#include <time.h>
#include <sched.h>
#include <string.h>

#include <iostream>

#include "TThread.hh"

using std::cerr;

//
// routine to call TThread::run() method
//
extern "C"
void *
_run_thread ( void *arg )
{
    if (arg != NULL)
    {
        ((TThread*) arg)->run();
        ((TThread*) arg)->reset_running();
    }// if

    return NULL;
}

////////////////////////////////////////////
//
// constructor and destructor
//
    
TThread::TThread ( int thread_no )
        : _running( false ), _thread_no(thread_no)
{
}

TThread::~TThread ()
{
    // request cancellation of the thread if running
    if ( _running )
        cancel();
}

////////////////////////////////////////////
//
// access local data
//

void
TThread::set_thread_no ( int n )
{
    _thread_no = n;
}
    
////////////////////////////////////////////
//
// thread management
//

//
// create thread (actually start it)
//
void 
TThread::create ( bool detached, bool sscope )
{
    if ( ! _running )
    {
        int status;
        
        if ((status = pthread_attr_init( & _thread_attr )) != 0)
        {
            cout << "(TThread) create : pthread_attr_init ("
                 << strerror( status ) << ")" << endl;
            return;
        }// if

        if ( detached )
        {
            // detache created thread from calling thread
            if ((status = pthread_attr_setdetachstate( & _thread_attr, PTHREAD_CREATE_DETACHED )) != 0)
            {
                cout << "(TThread) create : pthread_attr_setdetachstate ("
                     << strerror( status ) << ")" << endl;
                return;
            }// if
        }// if

        if ( sscope )
        {
            // use system-wide scheduling for thread
            if ((status = pthread_attr_setscope( & _thread_attr, PTHREAD_SCOPE_SYSTEM )) != 0 )
            {
                cout << "(TThread) create : pthread_attr_setscope ("
                     << strerror( status ) << ")" << endl;
                return;
            }// if
        }// if
            
#if defined(_POSIX_THREAD_PRIORITY_SCHEDULING) && defined(SUNOS)
        //
        // adjust thread-scheduling for Solaris
        //
        
        struct sched_param  t_param;
        
        t_param.sched_priority = sched_get_priority_min( SCHED_RR );
        
       if ((status = pthread_attr_setschedpolicy(  & _thread_attr, SCHED_RR )) != 0)
           cout << "(TThread) create : pthread_attr_setschedpolicy ("
                << strerror( status ) << ")" << endl;
        
       if ((status = pthread_attr_setschedparam(   & _thread_attr, & t_param )) != 0)
           cout << "(TThread) create : pthread_attr_setschedparam ("
                << strerror( status ) << ")" << endl;

       if ((status = pthread_attr_setinheritsched( & _thread_attr, PTHREAD_EXPLICIT_SCHED )) != 0)
           cout << "(TThread) create : pthread_attr_setinheritsched ("
                << strerror( status ) << ")" << endl;
#endif

#ifdef HPUX
       // on HP-UX we increase the stack-size for a stable behaviour
       // (need much memory for this !!!)
       pthread_attr_setstacksize( & _thread_attr, 32 * 1024 * 1024 );
#endif
        
       if ((status = pthread_create( & _thread_id, & _thread_attr, _run_thread, this ) != 0))
           cout << "(TThread) create : pthread_create ("
                << strerror( status ) << ")" << endl;
       else
           _running = true;
    }// if
    else
        cerr << "(TThread) create : thread is already running" << endl;
}

//
// detach thread
//
void 
TThread::detach ()
{
    if ( _running )
    {
        int status;
        
        // detach thread
        if ((status = pthread_detach( _thread_id )) != 0)
            cout << "(TThread) detach : pthread_detach ("
                 << strerror( status ) << ")" << endl;
    }// if
}

//
// synchronise with thread (wait until finished)
//
void 
TThread::join ()
{
    if ( _running )
    {
        int status;
    
        // wait for thread to finish
        if ((status = pthread_join( _thread_id, NULL )) != 0)
            cout << "(TThread) join : pthread_join ("
                 << strerror( status ) << ")" << endl;

        _running = false;
    }// if
}

//
// terminate thread
//
void 
TThread::exit ()
{
    if ( _running && (pthread_self() == _thread_id))
    {
        void  * ret_val = NULL;

        pthread_exit( ret_val );
        
        _running = false;
    }// if
}

//
// request cancellation of thread
//
void 
TThread::cancel ()
{
    if ( _running )
    {
        int status;
        
        if ((status = pthread_cancel( _thread_id )) != 0)
            cout << "(TThread) cancel : pthread_cancel ("
                 << strerror( status ) << ")" << endl;
    }// if
}

//
// send signal signo to thread
//
void 
TThread::kill ( int signo )
{
    if ( _running )
    {
        int status;
        
        if ((status = pthread_kill( _thread_id, signo )) != 0)
            cout << "(TThread) kill : pthread_kill ("
                 << strerror( status ) << ")" << endl;
        else
            _running = false;
    }// if
}

//
// put thread to sleep (milli + nano seconds)
//
void
TThread::sleep ( long m_sec, long nano_sec )
{
    if ( _running )
    {
        struct timespec  interval;

        if ( m_sec < 0 )    m_sec = 0;
        if ( nano_sec < 0 ) nano_sec = 0;

        while ( nano_sec >= 1000000 )
        {
            m_sec++;
            nano_sec -= 1000000;
        }// while
        
        interval.tv_sec  = m_sec / 1000;
        interval.tv_nsec = (m_sec % 1000) * 1000000 + nano_sec;
    
        if (nanosleep( & interval, 0 ) != 0)
            cout << "(TThread) delay : error in nanosleep" << endl;
    }// if
}



////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////
//
// class for a semaphore
//
////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////

//
// reset counter
//
void
TSemaphore::init ( int c )
{
    _mutex.lock();

    _value = c;

    _mutex.unlock();
    _cond.signal();
}

//
// increment counter
//
void
TSemaphore::inc ()
{
    _mutex.lock();

    _value++;

    _mutex.unlock();
    _cond.signal();
}
    
//
// decrement counter
//
void
TSemaphore::dec ()
{
    _mutex.lock();

    while ( _value <= 0 )
        _cond.wait( _mutex );

    _value--;

    _mutex.unlock();
}



////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////
//
// class for a counter (anti-semaphore)
//
////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////

//
// decrement counter
//
void
TCounter::dec ( int d, int c )
{
    _mutex.lock();

    _counter -= d;

    if ( _counter > 0 )
    {
        TMutex  lmu;

        // insert mutex in queue of waiting mutices and lock it
        _mutex_list.append( & lmu );
        lmu.lock();

        _mutex.unlock();

        // wait until released
        lmu.lock();
    }// if
    else
    {
        release_all();
        _counter = c;
        _mutex.unlock();
    }// else
}

//
// reset counter
//
void
TCounter::init ( int c )
{
    _counter = c;

    _mutex.lock();
    _mutex_list.remove_all();
    _mutex.unlock();
}

//
// release all mutices in list
//
void
TCounter::release_all ()
{
    while ( _mutex_list.size() > 0 )
        _mutex_list.behead()->unlock();
}