ThreadManager.C

// -*- C++ -*-
 
//  Copyright (c) 2005-2010, Isode Limited, London, England.
//  All rights reserved.
//                                                                       
//  Acquisition and use of this software and related materials for any      
//  purpose requires a written licence agreement from Isode Limited,
//  or a written licence from an organisation licenced by Isode Limited
//  to grant such a licence.
 
// 
//
// Manager.C
//
// Event Service Manager
//
// @VERSION@
 
#include <errno.h>
 
#include <isode/base/logging.h>
#include <isode/base/util.h>
#include <isode/compat/tailor.h>
 
#include "AsyncEventRef.h"
 
extern "C" {
    static void * Event_fnx (void *context ARGNOTUSED);
    static void * Timer_fnx (void *context ARGNOTUSED);
}
 
namespace Event {
    AsyncEventPlainQueue *AsyncEventPlainQueue::_freelist;
 
    class ThreadManager : public Manager {
    private:
        pthread_mutex_t    _mutex;
        pthread_cond_t     _waitQ;
        pthread_cond_t     _timeQ;
        pthread_cond_t     _termwait;
 
        pthread_attr_t        _threadattrs; 
        pthread_key_t         _llkey;       
 
        timespec              _duetime;     
        bool                  _running;     
        bool                  _terminating; 
        volatile int          _nthreads;    
        volatile int          _nactive;     
        volatile int          _nblocked;    
        volatile int          _ntimer;      
        volatile int          _nwait;       
 
        int                   _activelimit; 
        int                   _threadlimit; 
 
        AsyncEventPriorityQueue  *_mainQ;   
        AsyncEventTimerQueue     *_timerQ;  
        AsyncEventPlainQueue     *_atExitQ; 
 
        // Wake or create a thread
        inline void startThread ();
 
        // Internal dequeue of event
        inline void _deQueue (AsyncEvent *event) {
            AsyncEventRef *eventref = event->Reference();
 
            if ( eventref == 0 )
                // Not queued
                return;
 
            Receiver *rcvr = event->GetReceiver();
            
            if ( rcvr != 0 ) {
                // Might be on the receiver's internal queue
                // So need to dequeue with that locked
                rcvr->Extract (event);
            } else {
                eventref->Extract ();
            }
        }
 
        // Deliver an event
        // This unlocks the mutex
        inline void eventDeliver (AsyncEvent *event) {
            Receiver *rcvr = event->GetReceiver();
 
            if ( rcvr != 0 ) {
                // Queue event to receiver object prior to processing
                // This avoids the race condition where the event
                // object is being handled by one thread, and so cannot
                // be cancelled, but another thread wants to delete it.
 
                LOG_DEBUG (("eventDeliver: queue %p", event));
                rcvr->Queue (event);
 
                pthread_mutex_unlock (&_mutex);
 
                rcvr->Process ();
 
            } else {
 
                pthread_mutex_unlock (&_mutex);
                LOG_DEBUG (("eventDeliver: deliver %p", event));                
                event->Deliver();
            }
        }
 
    public:
        ThreadManager ();
 
        // Create a instance
        static Manager *Create ();
 
        // Run an event manager thread
        void Manager ();
 
        // Run a thread handing timed events
        void Timer ();
 
        // Is threading
        virtual bool IsThreaded () { return true; }
 
        // Start running the event service
        // If isthread is true, the event service use the current thread as
        // one of its threads.
        // Unless external threads inject events after this point,
        // this will never stop.
        virtual void Run (bool isthread);
 
        // Polling
        virtual bool Poll (const timespec *when,
                           bool immediate,
                           pthread_mutex_t *mutex,
                           pthread_cond_t *cond) {
            if ( immediate || mutex == 0 ) return false;
 
            Blocking();
 
            if ( when )
                pthread_cond_timedwait (cond, mutex, when);
            else
                pthread_cond_wait (cond, mutex);
 
            Unblocked();
 
            return false;
        }
 
        // Stop the event service.
        virtual void Terminate(unsigned int millisecs);
 
        // Signal that the thread is about to block for some time
        virtual void Blocking();
 
        // Signal that the thread is no longer blocking
        virtual void Unblocked();
 
        // Say a long-lived event
        // Returns true if terminating
        virtual bool LongLived();
 
        // Set the thread limits
        // activethreads - number of threads 'active' at once
        // maxthreads - limit on the total number of threads
        virtual void SetThreads (int activethreads, int maxthreads);
 
        // Get the allowed active thread limit
        virtual int GetActiveLimit () { return _activelimit; }
 
        // Queue an event for immediate execution with priority
        // Lower numerical values indicates higher priority
        virtual void Queue (AsyncEvent *event,
                            unsigned priority,
                            bool nostart = false);
 
        // Queue an event for execution at a specific time in the future
        virtual void QueueAt (AsyncEvent *event, const struct timespec *abstime);
 
        // Queue an event for execution after some interval
        virtual void QueueAt (AsyncEvent *event, unsigned int millisec);
 
        // Remove an event from its queue via its reference
        virtual void Dequeue (AsyncEvent *event);
 
        // Queue an event to be triggered when terminating
        virtual void QueueAtExit (AsyncEvent *event);
 
        // Return the time at which the next timed event is due or NULL
        // In a multithreaded environment this is not used as
        // We use a thread for the timer, rather than poll() or whatever.
        virtual const struct timespec *NextEventDue () { return 0; }
    };
 
    static ThreadManager *threadinstance;
 
    ThreadManager::ThreadManager ()
    {
        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"));
    }
 
    Manager *ThreadManager::Create () {
        if ( threadinstance != NULL )
            return threadinstance;
 
        threadinstance = new ThreadManager ();
 
        return threadinstance;
    }
 
    // The conditions under which a thread can be started are
    // closely related to the conditions under which the main Manager loop
    // will do something other than queue the thread on the default _waitQ
    // Note that we will be making the started thread 'active',
    // So this needs to be allowed for with in the use of _nactive
 
    inline void ThreadManager::startThread()
    {
        LOG_DEBUG (("startThread: active=%d limit=%d events=%d",
                    _nactive, _activelimit, (int) _mainQ->HasEvents()));
 
        // If started thread will exceed active limit or nothing to do
        if ( !_running || _nactive >= _activelimit ||
             !_mainQ->HasEvents() )
            return;
 
        LOG_DEBUG (("startThread: timer=%d nwait=%d threads=%d limit=%d",
                    _ntimer, _nwait, _nthreads, _threadlimit));
 
        if ( _nwait > 0 ) {
            // Wake one thread.
            pthread_cond_signal (&_waitQ);
 
        } else if ( _nthreads < _threadlimit ) {
            // No threads to do the work; create another
            pthread_t tid;
 
            // What should we do with errors here?
            if ( pthread_create (&tid, &_threadattrs, Event_fnx, 0) == 0 ) {
                // Counters are incremented here as it may be while
                // until the actual thread gets a go.
                // Threads are created 'active' as they can be active
                // as soon as they start up
                _nactive++;
                _nthreads++;
 
            }
        }
    }
 
    void ThreadManager::Run (bool isthread)
    {
        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 ();
    }
 
    void ThreadManager::Manager ()
    {
        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));
 
    }
 
    void ThreadManager::SetThreads (int activethreads, int maxthreads)
    {
        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);
    }
 
    void ThreadManager::Timer()
    {
        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);
    }
 
    void ThreadManager::Blocking ()
    {
        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);
    }
 
    void ThreadManager::Unblocked ()
    {
        LOG_DEBUG (("Unblocked"));
 
        if ( pthread_mutex_lock (&_mutex) != 0 ) {
            _nthreads--;
            _nblocked--;
            pthread_exit(0);
        }
 
        _nblocked--;
        _nactive++;
 
        pthread_mutex_unlock (&_mutex);
    }
 
    bool ThreadManager::LongLived()
    {
        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;
    }
 
    void ThreadManager::Queue (AsyncEvent *event,
                               unsigned priority,
                               bool nostart)
    {
        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);
    }
 
    void ThreadManager::QueueAtExit (AsyncEvent *event)
    {
        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);
 
        }
 
    }
 
    void ThreadManager::QueueAt (AsyncEvent *event, const struct timespec *duetime)
    {
        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);
    }
 
    void ThreadManager::QueueAt (AsyncEvent *event, unsigned millisecs)
    {
        LOG_DEBUG (("Queue: %p in %dms", event, millisecs));
 
        struct timespec now;
 
        getNow (now);
 
        addMillisecs (now, millisecs);
 
        QueueAt (event, &now);
    }
 
    void ThreadManager::Dequeue (AsyncEvent *event)
    {
        if ( pthread_mutex_lock (&_mutex) != 0 ) {
            _nthreads--;
            _nblocked--;
            pthread_exit(0);
        }
 
        _deQueue (event);
 
        pthread_mutex_unlock (&_mutex);
    }
 
    void ThreadManager::Terminate (unsigned int millisecs)
    {
        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);
    }
 
    Manager *CreateThreadManager ()
    {
        return ThreadManager::Create ();
    }
}
 
static void * Event_fnx (void *context ARGNOTUSED) {
    ::Event::threadinstance->Manager ();
    return 0;
}
 
static void * Timer_fnx (void *context ARGNOTUSED) {
    ::Event::threadinstance->Timer ();
    return 0;
}