queue/queue.h

/*--------------------------------------------------------------------------

  This is part of the Multithreading system of the Karoo project.

  (C) 2006,2007,2008 Brian Modra <brian@zwartberg.com>

  This library is free software; you can redistribute them 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 library 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 these libraries; if not, write to the Free Software Foundation,
  Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

--------------------------------------------------------------------------*/
#ifndef queue_h
#define queue_h

#ifndef _REENTRANT
#define _REENTRANT
#endif

#include "config.h"
#include "base.h"
#include "log.h"
#include "util.h"
#include <map>
#include <list>
#include <pthread.h>
#include <setjmp.h>
#include <signal.h>
#include <time.h>
#include <iostream>

using std::multimap;
using std::list;

namespace karoo {

    class pebble;
    class exepool;

    extern void initQueues();
    extern void closeQueues();

    typedef struct {
        double mean;    // average time spent in running each pebble 
        double sd;              // standard deviation in time spent running pebbles
        double min;             // minimum time spent running the fastest pebble
        double max;             // maximum time spent running the slowest pebble

        double mean_delay;      // average latency when running pebbles that had a timestamp
        double sd_delay;        // standard deviation in latency when running pebbles that had a timestamp
        double min_delay;       // minimum latency when running pebbles that had a timestamp
        double max_delay;       // maximum latency when running pebbles that had a timestamp

        int busy_percent;       // the busy percentage. Zero means the queue is way under-utilised. 100% is not necesarily a problem, though it likely is... but the latency statistics are more indicative of an overload problem.

        int num_iterations;     // the number of times this queue has iterated through its run loop

        int num_errors; // the number of caught exceptions from pebbles running in this queue.

        bool is_stopped;        // true if the queue is stopped

        double min_idle;        // minimum number of seconds spent waiting
        double max_idle;        // maximum number of seconds spent waiting

        int queued;             // number of pebbles currently queued
        int waiting;    // number of pebbles currently waiting for their timestamp to come up
        int running;    // number of pebbles currently runing (can only be 1 or zero)
        int timeouts;   // number of times the queue has timed out while waiting for pebbles to run.
    } qstats;

    struct exeque_comp {
        bool operator() (const struct timespec& lhs, const struct timespec& rhs) const;
    };



    class exeque : public referable
    {
    private:
        pthread_t thread;
        bool taken_over;
        mutable sem_t stopped;
        karoo_mutex stop_sem;
        karoo_mutex delete_sem;
        mutable karoo_mutex stats_sem;

        bool refusing;
        bool stopping;
        long granularity;
        exepool* delete_after_stop;
        friend class exepool;
        void scheduleDelete(exepool* parent);

        double* time_samples;
        double* idle_time_samples;
        double* delayed_time_samples;
        int time_sample_pos;
        mutable int time_sample_size;
        int time_sample_alloc_size;
        int timeouts;
        int iters;
        int errors;
        double idle_time;
        int run_count;
        int running;
        time_t stopped_time;

        struct timespec run_time;

        sigjmp_buf env;

    protected:
        multimap<const struct timespec,pebble*,exeque_comp> inbox;
        mutable karoo_mutex inbox_sem;
        karoo_cond inbox_notifier;
        //bool waiting;

        log logger;

    public:

        exeque(long granularity = 0, int time_sample_alloc_size = 100);
        virtual ~exeque();

        void setLogger(const log& logger) { this->logger = logger; }

        bool start(long granularity = 5, bool take_over = false);

        void stop(bool block = false);

        void join();


        bool isStopped(long wait_usec = 0) const;

        void refuse();



        void add(pebble* p);
        void add(const struct timespec& ts, pebble* p);
        void add(int after_seconds, pebble* p, long after_nano_seconds = 0);

        void run();

        void reset(long granularity = 0, int time_sample_alloc_size = 100);

        const qstats& getStats(qstats& qs, bool reset = false) const;

        int getBusyPercent(bool reset = false);
        int getRunCount(bool reset = false);
        int getErrorCount(bool reset = false);
        int getTimeoutCount(bool reset = false);
        int getWaitingCount() const;
        int getQueuedCount() const;
        int getImminentCount(unsigned usecs = 0) const;

        bool hasBeenIdleSince(const struct timespec& comp) const;

        pthread_t getThread() { return thread; }

        sigjmp_buf& getSigLongJmpEnv() { return env; }
    };

    class pebble: public referable
    {
    private:
        friend class exeque;
        karoo_mutex sem;
        bool dereference_scheduled;
    protected:
        exeque* q;
        text exception;
        bool error;
    public:
        pebble();
        virtual ~pebble();
        virtual void run() = 0;
        exeque* getQueue() { return q; }
        const text& getException() const { return exception; }
        bool getSuccess() const { return !error; }
        void dereferenceAfterRun() { dereference_scheduled = true; }
    };

    class poolstats {
    private:
        int size;
        qstats* qs;
        friend class exepool;
    public:
        poolstats() { size = 0; qs = NULL; }
        poolstats(const poolstats& ref);
        virtual ~poolstats();
    
        poolstats& operator=(const poolstats& ref);
        const qstats& operator[](int i) const { return qs[i]; }
        qstats& operator[](int i) { return qs[i]; }
        int getSize() const { return size; }
    };


    class exepool : public referable
    {
    private:
        mutable karoo_mutex sem;
        friend class poolpebble;
        void schedule(pebble* p);
    protected:
        long granularity;
        int time_sample_alloc_size;
        int alloc_size;
        int size;
        exeque* feed;
        exeque** qs;
        log logger;
        karoo_mutex auto_sem;
        list<exeque*> auto_scheduled;
        list<exeque*> auto_stopped;
    public:
        exepool(int alloc_size, long granularity = 5, int time_sample_alloc_size = 100); // granularity = 0 means use the default of 5 seconds
        virtual ~exepool();

        void cleanupLater(exeque* q);

        void cleanupAutoDeletes(bool force);

        void setLogger(const log& logger);

        void stop(bool block = false);
        bool isStopped(long wait_usec = 0) const;

        int getImminentCount(unsigned usecs = 0) const;

        void add(pebble* p);
        void add(const struct timespec& ts, pebble* p);
        void add(int after_seconds, pebble* p);

        const poolstats& getStats(poolstats& ps, bool reset = false) const;
        int getSize() const;
    };

    class poolpebble: public pebble
    {
    private:
    protected:
        exepool* pool;
        pebble* p;
    public:
        poolpebble(exepool* pool, pebble* p);
        virtual ~poolpebble();
        void run();
    };

    template <typename T>
    class synchronised
    {
    private:
        T val;
        mutable karoo_mutex sem;
    public:
        synchronised<T>() { MUTEX_INIT(sem); }
        synchronised<T>(const T& val) { MUTEX_INIT(sem); this->val = val; }
        synchronised<T>(const synchronised<T>& ref) { MUTEX_INIT(sem); this->val = (T)ref; }
        virtual ~synchronised<T>() { MUTEX_DESTROY(sem); }

        T operator=(const T& val) {
            MUTEX_LOCK(sem);
            try {
                this->val = val;
                MUTEX_UNLOCK(sem);
                return val;
            }
            catch (...) {
                try { MUTEX_UNLOCK(sem); } catch(...) {}
                throw;
            }
        }

        T operator=(const synchronised<T>& ref) {
            T v = (T)ref;
            MUTEX_LOCK(sem);
            try {
                val = v;
                MUTEX_UNLOCK(sem);
                return v;
            } catch (...) {
                try { MUTEX_UNLOCK(sem); } catch(...) {}
                throw;
            }
        }

        operator T() const {
            MUTEX_LOCK(sem);
            try {
                T v = val;
                MUTEX_UNLOCK(sem);
                return v;
            }
            catch (...) {
                try { MUTEX_UNLOCK(sem); } catch(...) {}
                throw;
            }
        }

        T operator++(int) {
            MUTEX_LOCK(sem);
            try {
                T v = val;
                val++;
                MUTEX_UNLOCK(sem);
                return v;
            } catch (...) {
                try { MUTEX_UNLOCK(sem); } catch(...) {}
                throw;
            }
        }

        T operator++() {
            MUTEX_LOCK(sem);
            try {
                val++;
                T v = val;
                MUTEX_UNLOCK(sem);
                return v;
            } catch (...) {
                try { MUTEX_UNLOCK(sem); } catch(...) {}
                throw;
            }
        }
    };

    extern std::ostream& operator<<(std::ostream& out, const qstats& qs);

}

#endif

---