PollManager.C

// -*- C++ -*-
 
//  Copyright (c) 2006-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.
 
// 
//
// PollManager.C
//
// Event Service Manager used in a polling fashion
//
// @VERSION@
 
#include <isode/base/logging.h>
 
#include "AsyncEventRef.h"
 
namespace Event {
    class PollManager : public Manager {
    private:
        pthread_mutex_t    _mutex;
        pthread_cond_t     _cond;
 
        const timespec           *_duetime; 
        
        AsyncEventPriorityQueue  *_mainQ;   
        AsyncEventTimerQueue     *_timerQ;  
        AsyncEventPlainQueue     *_atExitQ; 
 
        pthread_t             _thread;      
        bool                  _busy;        
 
        // 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 ();
            }
        }
 
    public:
        PollManager ();
 
        // Create a instance
        static Manager *Create ();
 
        // Is threading
        virtual bool IsThreaded () { return false; }
 
        // Run a thread - does nothing
        void Run (bool ) {}
 
        // Poll the event manager
        bool Poll (const timespec *when,
                   bool immediate,
                   pthread_mutex_t *mutex,
                   pthread_cond_t *cond);
 
        // 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() { return false; }
 
        // Set the thread limits
        // activethreads - number of threads 'active' at once
        // maxthreads - limit on the total number of threads
        virtual void SetThreads (int , int ) {}
 
        // Get the allowed active thread limit
        virtual int GetActiveLimit () { return 0; }
 
        // 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 _duetime; }
    };
 
    static PollManager *threadinstance;
 
    PollManager::PollManager () : _busy(false)
    {
        pthread_mutex_init (&_mutex, NULL);
        pthread_cond_init (&_cond, NULL);
 
        _mainQ   = new AsyncEventPriorityQueue;
        _timerQ  = new AsyncEventTimerQueue;
        _atExitQ = new AsyncEventPlainQueue;
 
        LOG_DEBUG (("PollManager created"));
    }
 
    Manager *PollManager::Create () {
        if ( threadinstance != NULL )
            return threadinstance;
 
        threadinstance = new PollManager ();
 
        return threadinstance;
    }
 
    bool PollManager::Poll (const timespec *when,
                            bool immediate,
                            pthread_mutex_t *clientmutex,
                            pthread_cond_t *)
    {
        if ( pthread_mutex_lock (&_mutex) != 0 )
            return false;
 
        // Unlock the client mutex for the duration of this call
        // So that stuff can be delivered to it via events
        if ( clientmutex ) pthread_mutex_unlock (clientmutex);
 
        // If the current thread is the busy thread, then recursing,
        // which is allowed.
        if ( _busy && !pthread_equal(_thread, pthread_self()) ) {
            // Someone else is executing the manager
            if ( !immediate ) {
                if ( when != 0 )
                    pthread_cond_timedwait (&_cond, &_mutex, when);
                else
                    pthread_cond_wait (&_cond, &_mutex);
            }
 
        } else {
 
            _busy = true;
            _thread = pthread_self();
 
            struct timespec now;
            getNow (now);
 
            if ( _timerQ->HasEvents () ) {
 
                AsyncEvent *tevent;
 
                // Get any due events onto the main event queue
                while ( (tevent = _timerQ->First (now)) != 0 ) {
                    LOG_DEBUG (("timerQ event %p due", tevent));
                    _mainQ->Insert (tevent, 0);
                }
            }
 
            _duetime = 0;
 
            const struct timespec *nextdue = _timerQ->NextPrio();
 
            if ( nextdue != 0 )
                _duetime = nextdue;
 
            if ( when != 0 ) {
                if ( _duetime == 0 || timespec_cmp (when, _duetime) )
                    _duetime = when;
            }
 
            LOG_DEBUG (("PollManager: events=%d",_mainQ->HasEvents()));
 
            if ( _mainQ->HasEvents() ) {
 
                AsyncEvent *event = _mainQ->First();
 
                Receiver *rcvr = event ? event->GetReceiver() : 0;
 
                // Client mutex is unlocked while events are delivered
                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 (("Manager: queue %p", event));
                    rcvr->Queue (event);
 
                    pthread_mutex_unlock (&_mutex);
 
                    rcvr->Process ();
 
                } else {
 
                    pthread_mutex_unlock (&_mutex);
 
                    LOG_DEBUG (("Manager: deliver %p", event));
 
                    event->Deliver();
                }
 
                if ( pthread_mutex_lock(&_mutex) != 0 ) {
                    return false;
                }
 
            } else if ( _duetime != 0 ) {
                // Nothing to process until duetime, so wait
                (void) pthread_cond_timedwait (&_cond, &_mutex, _duetime);
            }
 
            _busy = false;
        }
 
        bool retval = ( _mainQ->HasEvents() &&
                        *(_mainQ->NextPrio()) < (unsigned) MAXPRIO );
 
        // Tell any other threads that something has happened
        pthread_cond_broadcast (&_cond);
 
        pthread_mutex_unlock (&_mutex);
 
        if ( clientmutex ) pthread_mutex_lock (clientmutex);
 
        return retval;
    }
 
    void PollManager::Queue (AsyncEvent *event,
                             unsigned priority,
                             bool)
    {
        LOG_DEBUG (("Queue: %p prio=%u", event, priority));
 
        if ( pthread_mutex_lock (&_mutex) != 0 )
            return;
 
        _deQueue (event);
 
        _mainQ->Insert (event, priority);
 
        pthread_mutex_unlock (&_mutex);
 
    }
 
    void PollManager::QueueAtExit (AsyncEvent *event)
    {
        LOG_DEBUG (("QueueAtExit: %p", event));
 
        if ( pthread_mutex_lock (&_mutex) != 0 )
            return;
 
        _deQueue (event);
 
        _atExitQ->Insert (event);
 
        pthread_mutex_unlock (&_mutex);
    }
 
    void PollManager::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 )
            return;
 
        _deQueue (event);
 
        _timerQ->Insert (event, *duetime);
 
        pthread_mutex_unlock (&_mutex);
    }
 
    void PollManager::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 PollManager::Dequeue (AsyncEvent *event)
    {
        if ( pthread_mutex_lock (&_mutex) != 0 )
            return;
 
        _deQueue (event);
 
        pthread_mutex_unlock (&_mutex);
    }
 
    void PollManager::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);
        }
 
        // 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 ) {
            // Need to unlock during delivery
            pthread_mutex_unlock (&_mutex);
 
            event->Deliver ();
 
            if ( pthread_mutex_lock (&_mutex) != 0 ) {
                pthread_exit(0);
            }
        }
 
        pthread_mutex_unlock (&_mutex);
    }
 
    Manager *CreatePollManager ()
    {
        return PollManager::Create ();
    }
}