Event::ThreadManager Class Reference

Event manager which is multi-threaded. More...

Inheritance diagram for Event::ThreadManager:

Public Member Functions
void	Manager ()
void	Timer ()
virtual bool	IsThreaded ()
virtual void	Run (bool isthread)
virtual bool	Poll (const timespec *when, bool immediate, pthread_mutex_t *mutex, pthread_cond_t *cond)
virtual void	Terminate (unsigned int millisecs)
virtual void	Blocking ()
virtual void	Unblocked ()
virtual bool	LongLived ()
virtual void	SetThreads (int activethreads, int maxthreads)
virtual int	GetActiveLimit ()
virtual void	Queue (AsyncEvent *event, unsigned priority, bool nostart=false)
virtual void	QueueAt (AsyncEvent *event, const struct timespec *abstime)
virtual void	QueueAt (AsyncEvent *event, unsigned int millisec)
virtual void	Dequeue (AsyncEvent *event)
virtual void	QueueAtExit (AsyncEvent *event)
virtual const struct timespec *	NextEventDue ()

Static Public Member Functions
static Manager *	Create ()

Detailed Description

Event manager which is multi-threaded.

Definition at line 36 of file ThreadManager.C.

Constructor & Destructor Documentation

ThreadManager()

Event::ThreadManager::ThreadManager

(

)

Definition at line 198 of file ThreadManager.C.

    {
        pthread_mutex_init (&_mutex, NULL);
        pthread_cond_init (&_waitQ, NULL);
        pthread_cond_init (&_timeQ, NULL);
        pthread_cond_init (&_termwait, NULL);
 
        pthread_key_create (&_llkey, NULL);
 
        pthread_attr_init (&_threadattrs);
 
        const size_t TWOMEG = (2*1024*1024);
        size_t default_stacksize = 0;
 
        if ( pthread_attr_getstacksize (&_threadattrs,
                                        &default_stacksize) != 0 )
            default_stacksize = TWOMEG;
 
        if ( thread_stacksize != 0 ) {
            default_stacksize = thread_stacksize;
        } else if ( default_stacksize < TWOMEG ) {
            default_stacksize = TWOMEG;
        }
 
        if ( default_stacksize < PTHREAD_STACK_MIN )
            default_stacksize = PTHREAD_STACK_MIN;
 
        pthread_attr_setstacksize (&_threadattrs, default_stacksize);
 
        pthread_attr_setdetachstate (&_threadattrs, PTHREAD_CREATE_DETACHED);
 
        _running     = false;
        _terminating = false;
 
        _nthreads    = 0;
        _nactive     = 0;
        _nblocked    = 0;
        _ntimer      = 0;
        _nwait       = 0;
 
        _duetime.tv_sec = 0;
        _duetime.tv_nsec = 0;
 
        long config_cpu, online_cpu;
 
        isode_cpucount (&config_cpu, &online_cpu, 0, 0);
 
        if ( num_threads != 0 )
            online_cpu = num_threads;
 
        _activelimit   = online_cpu > 2 ? online_cpu : 2;
        _threadlimit   = _activelimit > 10 ? 2*_activelimit : 20;
 
        if ( _threadlimit < thread_limit )
            _threadlimit = thread_limit;
 
        _mainQ   = new AsyncEventPriorityQueue;
        _timerQ  = new AsyncEventTimerQueue;
        _atExitQ = new AsyncEventPlainQueue;
 
        LOG_DEBUG (("ThreadManager created"));
    }

Member Function Documentation

Create()

Manager * Event::ThreadManager::Create ( )

static

Definition at line 261 of file ThreadManager.C.

                                    {
        if ( threadinstance != NULL )
            return threadinstance;
 
        threadinstance = new ThreadManager ();
 
        return threadinstance;
    }

Manager()

void Event::ThreadManager::Manager

(

)

Definition at line 335 of file ThreadManager.C.

    {
        if ( pthread_mutex_lock (&_mutex) != 0 ) {
            _nactive--;
            _nthreads--;
            return;
        }
 
        while ( !_terminating && _nthreads <= _threadlimit ) {
 
            LOG_DEBUG (("Manager: active=(%d:%d) events=%d",
                        _nactive, _activelimit, _mainQ->HasEvents()));
 
            if ( _nactive <= _activelimit && _mainQ->HasEvents() ) {
 
                AsyncEvent *event = _mainQ->First();
 
                // Unlocks mutex
                eventDeliver (event);
 
                if ( pthread_mutex_lock(&_mutex) != 0 ) {
                    _nactive--;
                    _nthreads--;
                    return;
                }
 
                // Check to see if thread was executing long lived event
                if ( pthread_getspecific (_llkey) != NULL ) {
                    LOG_DEBUG (("Manager: %p longlived (%d)",
                                event, _activelimit));
 
                    pthread_setspecific (_llkey, NULL);
                    if ( _activelimit > 1 ) _activelimit--;
                }
 
            } else {
 
                // Wait for something to do
 
                _nactive--;
                _nwait++;
 
                LOG_DEBUG (("Manager: wait on queue"));
 
                int ret = pthread_cond_wait (&_waitQ, &_mutex);
 
                _nwait--;
                _nactive++;
 
                if ( ret != 0 )
                    break;
            }
        }
 
        _nthreads--;
        _nactive--;
 
        if ( _terminating )
            pthread_cond_signal (&_termwait);
 
        pthread_mutex_unlock (&_mutex);
 
        LOG_DEBUG (("Manager: thread exiting (n=%d, active=%d)",
                    _nthreads, _nactive));
 
    }

Timer()

void Event::ThreadManager::Timer

(

)

Definition at line 429 of file ThreadManager.C.

    {
        LOG_DEBUG (("Timer() starting"));
 
        if ( pthread_mutex_lock (&_mutex) != 0 ) {
            _ntimer--;
            _nthreads--;
            return;
        }
 
        while ( !_terminating ) {
            struct timespec now;
            getNow (now);
 
            LOG_DEBUG (("Timer: timer events: %d", _timerQ->HasEvents ()));
            if ( _timerQ->HasEvents () ) {
 
                AsyncEvent *tevent;
 
                while ( (tevent = _timerQ->First (now)) != 0 ) {
                    LOG_DEBUG (("timerQ event %p due", tevent));
                    _mainQ->Insert (tevent, 0);
                    if ( _running && _nactive < _activelimit )
                        startThread();
                }
            }
 
            if ( _timerQ->HasEvents () ) {
                _duetime = *_timerQ->NextPrio();
 
            } else {
                // Wake up in 10 secs
                now.tv_sec += 10;
                _duetime = now;
            }
 
            LOG_DEBUG (("Timer: waiting until %ld.%ld",
                        (long)_duetime.tv_sec, (long)_duetime.tv_nsec));
 
            int ret = pthread_cond_timedwait (&_timeQ, &_mutex, &_duetime);
 
            if (  ret != 0 && ret != ETIMEDOUT ) {
                break;
            }
        }
 
        LOG_DEBUG (("Timer() stopping"));
 
        _ntimer--;
        _nthreads--;
        pthread_mutex_unlock (&_mutex);
    }

IsThreaded()

virtual bool Event::ThreadManager::IsThreaded ( )

inlinevirtual

Definition at line 123 of file ThreadManager.C.

{ return true; }

Run()

void Event::ThreadManager::Run ( bool  isthread )

virtual

Definition at line 310 of file ThreadManager.C.

    {
        if ( pthread_mutex_lock (&_mutex) != 0 ) {
            return;
        }
 
        // Start Manager
        _running = true;
 
        if ( isthread ) {
            // This thread becomes one of the threads
            _nthreads++;
            _nactive++;
 
        } else {
            // May need to start a thread
            startThread ();
        }
 
        pthread_mutex_unlock (&_mutex);
 
        if ( isthread )
            Manager ();
    }

Poll()

virtual bool Event::ThreadManager::Poll ( const timespec *  when, bool  immediate, pthread_mutex_t *  mutex, pthread_cond_t *  cond  )

inlinevirtual

Definition at line 133 of file ThreadManager.C.

                                                 {
            if ( immediate || mutex == 0 ) return false;
 
            Blocking();
 
            if ( when )
                pthread_cond_timedwait (cond, mutex, when);
            else
                pthread_cond_wait (cond, mutex);
 
            Unblocked();
 
            return false;
        }

Terminate()

void Event::ThreadManager::Terminate ( unsigned int  millisecs )

virtual

Definition at line 665 of file ThreadManager.C.

    {
        LOG_DEBUG (("Manager: Terminate in %dms", millisecs));
 
        struct timespec now;
 
        getNow (now);
 
        addMillisecs (now, millisecs);
 
        if ( pthread_mutex_lock (&_mutex) != 0 ) {
            pthread_exit(0);
        }
 
        _terminating = true;
 
        // Wake up all waiters
        pthread_cond_broadcast (&_waitQ);
        pthread_cond_broadcast (&_timeQ);
 
        // Cancel all queued events
        AsyncEvent    *event;
 
        while ( (event = _mainQ->First()) != 0 )
            _deQueue (event);
 
        while ( (event = _timerQ->First()) != 0 )
            _deQueue (event);
 
        // Deliver all at exit events
        // in the reverse order to their being added
        while ( (event = _atExitQ->Last()) != 0 ) {
            // Unlocks mutex
            eventDeliver (event);
 
            if ( pthread_mutex_lock (&_mutex) != 0 ) {
                pthread_exit(0);
            }
        }
 
        // Wait for all threads to terminate
        while ( _nthreads > 0 &&
                pthread_cond_timedwait (&_termwait, &_mutex, &now) == 0 )
            /* empty */;
 
        pthread_mutex_unlock (&_mutex);
    }

Blocking()

void Event::ThreadManager::Blocking ( )

virtual

Definition at line 482 of file ThreadManager.C.

    {
        LOG_DEBUG (("Blocking"));
 
        if ( pthread_mutex_lock (&_mutex) != 0 ) {
            _nthreads--;
            _nactive--;
            pthread_exit(0);
        }
 
        _nblocked++;
        _nactive--;
 
        // Since _nactive has fallen, another thread might be able to start
        // or may need to be created.
        startThread ();
 
        pthread_mutex_unlock (&_mutex);
    }

Unblocked()

void Event::ThreadManager::Unblocked ( )

virtual

Definition at line 502 of file ThreadManager.C.

    {
        LOG_DEBUG (("Unblocked"));
 
        if ( pthread_mutex_lock (&_mutex) != 0 ) {
            _nthreads--;
            _nblocked--;
            pthread_exit(0);
        }
 
        _nblocked--;
        _nactive++;
 
        pthread_mutex_unlock (&_mutex);
    }

LongLived()

bool Event::ThreadManager::LongLived ( )

virtual

Definition at line 518 of file ThreadManager.C.

    {
        LOG_DEBUG (("LongLived"));
 
        if ( pthread_mutex_lock (&_mutex) != 0 ) {
            _nactive--;
            _nthreads--;
            _nblocked--;
            pthread_exit(0);
        }
 
        if ( _terminating ) {
            pthread_mutex_unlock (&_mutex);
            return true;
        }
 
        // If the total thread limit has been reached, then do nothing.
        // Otherwise, so long as we have not already marked the thread as
        // long lived, then do so and start it.
        if ( _activelimit < _threadlimit &&
             pthread_getspecific (_llkey) == NULL ) {
            // This is a long-lived thread possibly blocking other events
            // We have not previously marked it as such
 
            // mark this thread as long-lived
            if (pthread_setspecific (_llkey, (void *)1) == 0) {
                _activelimit++;
                startThread ();
            }
        }
 
        pthread_mutex_unlock (&_mutex);
        return false;
    }

SetThreads()

void Event::ThreadManager::SetThreads ( int  activethreads, int  maxthreads  )

virtual

Definition at line 402 of file ThreadManager.C.

    {
        LOG_DEBUG (("SetThreads: %d %d",activethreads, maxthreads));
 
        if ( pthread_mutex_lock (&_mutex) != 0 )
            return;
 
        if ( activethreads > 1 )
            _activelimit = activethreads;
 
        if ( maxthreads > _activelimit ) {
            _threadlimit = maxthreads;
        } else if ( maxthreads == 0 ) {
            if ( _activelimit > 10 )
                _threadlimit = 2*_activelimit;
            else
                _threadlimit = 20;
 
            if ( _threadlimit < thread_limit )
                _threadlimit = thread_limit;
        }
 
        startThread ();
 
        pthread_mutex_unlock (&_mutex);
    }

GetActiveLimit()

virtual int Event::ThreadManager::GetActiveLimit ( )

inlinevirtual

Definition at line 170 of file ThreadManager.C.

{ return _activelimit; }

Queue()

void Event::ThreadManager::Queue ( AsyncEvent *  event, unsigned  priority, bool  nostart = false  )

virtual

Definition at line 553 of file ThreadManager.C.

    {
        LOG_DEBUG (("Queue: %p prio=%u nostart=%d", event, priority, nostart));
 
        if ( pthread_mutex_lock (&_mutex) != 0 ) {
            _nthreads--;
            _nblocked--;
            pthread_exit(0);
        }
 
        _deQueue (event);
 
        if ( !_terminating ) {
            _mainQ->Insert (event, priority);
 
            if ( !nostart )
                startThread ();
        }
 
        pthread_mutex_unlock (&_mutex);
    }

QueueAt()

void Event::ThreadManager::QueueAt ( AsyncEvent *  event, const struct timespec *  abstime  )

virtual

Definition at line 601 of file ThreadManager.C.

    {
        LOG_DEBUG (("Queue: %p @ %d.%9.9ld",
                    event, (int)duetime->tv_sec, duetime->tv_nsec));
 
        if ( pthread_mutex_lock (&_mutex) != 0 ) {
            _nthreads--;
            _nblocked--;
            pthread_exit(0);
        }
 
        _deQueue (event);
 
        if (!_terminating ) {
            _timerQ->Insert (event, *duetime);
 
            if ( _ntimer <= 0 ) {
                // Need to start a thread to handle the timer
                pthread_t tid;
 
                // What should we do with errors here?
                if ( pthread_create (&tid, &_threadattrs, Timer_fnx, 0) == 0 ) {
                    // Counters are incremented here as it may be while
                    // until the actual thread gets a go.
                    _ntimer++;
                    _nthreads++;
                }
 
            } else if ( timespec_cmp (duetime, &_duetime) ) {
                // This event due before timer thread wakes up,
                // so wake it up now to recalculate
                pthread_cond_signal (&_timeQ);
            }
        }
 
        pthread_mutex_unlock (&_mutex);
    }

Dequeue()

void Event::ThreadManager::Dequeue ( AsyncEvent *  event )

virtual

Definition at line 652 of file ThreadManager.C.

    {
        if ( pthread_mutex_lock (&_mutex) != 0 ) {
            _nthreads--;
            _nblocked--;
            pthread_exit(0);
        }
 
        _deQueue (event);
 
        pthread_mutex_unlock (&_mutex);
    }

QueueAtExit()

void Event::ThreadManager::QueueAtExit ( AsyncEvent *  event )

virtual

Definition at line 577 of file ThreadManager.C.

    {
        LOG_DEBUG (("QueueAtExit: %p", event));
 
        if ( pthread_mutex_lock (&_mutex) != 0 ) {
            _nthreads--;
            _nblocked--;
            pthread_exit(0);
        }
 
        _deQueue (event);
 
        if ( _terminating ) {
            // Unlocks mutex
            eventDeliver (event);
 
        } else {
            _atExitQ->Insert (event);
            pthread_mutex_unlock (&_mutex);
 
        }
 
    }

NextEventDue()

virtual const struct timespec * Event::ThreadManager::NextEventDue ( )

inlinevirtual

Definition at line 193 of file ThreadManager.C.

{ return 0; }

---

The documentation for this class was generated from the following file:

• ThreadManager.C