Syncmsg.h

Go to the documentation of this file.

// -*- 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.
 
 
// 
//
//
// @VERSION@
 
#ifndef _SYNCMSG_H_
#define _SYNCMSG_H_
 
#include <errno.h>
#include <pthread.h>
#include "timeutil.h"
#include "EventSvc.h"
 
template<class M, class C> class SyncEvent : public Event::AsyncEvent {
 private:
    M       _msg;
    C      *_target;
    bool    _onheap;
 
 public:
    inline SyncEvent () : _target(0), _onheap(0) {}
 
    inline SyncEvent (M *msg, C *target) :
        _msg (*msg), _target(target), _onheap(true) { }
 
    virtual ~SyncEvent () { }
 
    inline void Init (C *rcvr) {
        SetReceiver (rcvr);
        _target = rcvr;
    }
 
    virtual void Deliver () {
        _target->actualDeliver (&_msg);
    }
 
    virtual bool ShouldDelete () const { return _onheap; }
 
    inline void Send (M &data) {
        _msg = data;
        ::Event::Manager::GetManager()->Queue (this);
    }
 
    inline void Send (const timespec *when, M &data ) {
        _msg = data;
 
        ::Event::Manager::GetManager()->QueueAt (this, when);
    }
 
    inline void Sendms (unsigned millisecs, M &data) {
        _msg = data;
 
        ::Event::Manager::GetManager()->QueueAt (this, millisecs);
    }
 
    inline void SendAtExit (M &data) {
        _msg = data;
        ::Event::Manager::GetManager()->QueueAtExit (this);
    }
 
};
 
 
class SyncEventQueue : Event::AsyncEventRef {
private:
    // A simple, doubly linked list
    SyncEventQueue   *_next;
    SyncEventQueue   *_prev;
    SyncEventQueue   *_free;
    SyncEventQueue  **_freelist;
 
    SyncEventQueue (Event::AsyncEvent *event) : Event::AsyncEventRef (event) {}
 
public:
    SyncEventQueue () {
        _next = _prev = this;
        _freelist = &_free;
        _free = 0;
    }
 
    ~SyncEventQueue () {
        // Only delete items if root element
        if ( _freelist == &_free ) {
            SyncEventQueue *q1 = _free;
            while ( q1 ) {
                SyncEventQueue *q2 = q1;
                q1 = q1->_free;
                delete q2;
            }
        }
    }
 
    void Insert (Event::AsyncEvent *event) {
        SyncEventQueue *newq = _free;
 
        if ( newq ) {
            _free = newq->_free;
            newq->Set (event);
        } else {
            newq = new SyncEventQueue (event);
            newq->_freelist = _freelist;
        }
 
        _prev->_next = newq;
        newq->_prev = _prev;
        newq->_next = this;
        _prev = newq;
    }
 
    virtual void extract () {
        _prev->_next = _next;
        _next->_prev = _prev;
        _free = *_freelist;
        *_freelist = this;
    }
 
    Event::AsyncEvent * First () {
        if ( _next == this )
            return 0;
 
        return _next->Extract();
    }
};
 
//
 
template<class C> class Syncmsgobj : public C {
 private:
    pthread_mutex_t              _mutex;
    SyncEventQueue               _queue;
    bool                         _busy;
    bool                         _die;
    bool                         _dying;
    int                          _qcount;
 
 protected:
    inline void dieAux () {
        int rc = pthread_mutex_lock (&_mutex);
 
        _die = true;
 
        bool killit = !_busy && !_dying && _qcount <= 0;
 
        if ( killit )
            _dying = true;
 
        if ( rc == 0 ) pthread_mutex_unlock (&_mutex);
 
        if ( killit ) delete this;
    }
 
 public:
    inline Syncmsgobj () : _busy(false), _die(false), _dying(false), _qcount(0) {
        pthread_mutex_init (&_mutex, 0);
    }
 
    virtual ~Syncmsgobj () {
        Event::AsyncEvent *item;
 
        while ( (item = _queue.First()) != 0 )
            if ( item->ShouldDelete() )
                delete item;
 
        pthread_mutex_destroy (&_mutex);
    }
 
    // Always in a pair with a Process
    virtual void Queue (Event::AsyncEvent *event) {
        if ( pthread_mutex_lock (&_mutex) != 0 )
            return;
 
        _queue.Insert (event);
 
        _qcount++;
 
        pthread_mutex_unlock (&_mutex);    
    }
 
    virtual void Extract (Event::AsyncEvent *event) {
        if ( pthread_mutex_lock (&_mutex) != 0 )
            return;
 
        Event::AsyncEventRef *ref = event->Reference();
 
        if ( ref != 0 )
            ref->Extract();
 
        pthread_mutex_unlock (&_mutex);    
    }
 
    // ALways follows a Queue
    virtual void Process () {
        if ( pthread_mutex_lock (&_mutex) != 0 )
            return;
 
        _qcount--;
 
        if ( _busy ) {
 
            pthread_mutex_unlock (&_mutex);
 
            return;
        }
 
        _busy = true;
 
        Event::AsyncEvent *item;
 
        while ( (item = _queue.First ()) ) {
 
            if ( _die ) {
                if ( item->ShouldDelete() )
                    delete item;
                continue;
            }
 
            pthread_mutex_unlock (&_mutex);
 
            // As the delivery of the item may delete the underlying object,
            // we need to find out whether this should delete it beforehand
            bool shouldDelete = item->ShouldDelete();
 
            item->Deliver ();
 
            if ( shouldDelete )
                delete item;
 
            if ( pthread_mutex_lock (&_mutex) != 0 )
                return;
        }
 
        _busy = false;
 
        bool killit = _die && !_dying && _qcount <= 0;
 
        if ( killit )
            _dying = true;
 
        pthread_mutex_unlock (&_mutex);
 
        if ( killit )
            delete this;
    }
 
    template <class M> inline void msgDeliver (M *msg) {
 
        if ( pthread_mutex_lock (&_mutex) != 0 )
            return;
 
        if ( _die || _busy ) {
 
            // Don't bother to queue this message if the entity is going away
            if ( !_die )
                _queue.Insert (new SyncEvent<M,C> (msg, this));
                
            pthread_mutex_unlock (&_mutex);
 
            return;
        }
 
        _busy = true;
        pthread_mutex_unlock (&_mutex);
 
        this->actualDeliver (msg);
 
        if ( pthread_mutex_lock (&_mutex) != 0 )
            return;
 
        Event::AsyncEvent *item;
 
        while ( (item = _queue.First ()) ) {
 
            if ( _die ) {
                if ( item->ShouldDelete() )
                    delete item;
                continue;
            }
 
            pthread_mutex_unlock (&_mutex);
 
            bool shouldDelete =  item->ShouldDelete();
 
            item->Deliver ();
 
            if ( shouldDelete )
                delete item;
 
            if ( pthread_mutex_lock (&_mutex) != 0 )
                return;
        }
 
        _busy = false;
 
        bool killit = _die && !_dying && _qcount <= 0;
 
        if ( killit )
            _dying = true;
 
        pthread_mutex_unlock (&_mutex);
 
        if ( killit ) delete this;
    }
};
 
#endif  /* _SYNCMSG_H_ */