WorkPile.h

// $Id: WorkPile.h 21 2010-09-05 04:18:17Z cschwarz1 $

#ifndef BASE_WORKPILE_H
#define BASE_WORKPILE_H

#include <cerrno>
#include <deque>
#include <list>
#include "Condition.h"
#include "Thread.h"

namespace base {

    template<class T> class WorkPile: public std::deque<T>, public Condition {
    public:
        typedef std::deque<T> DEQUE;   


        WorkPile(unsigned wmlow = 0, unsigned wmhigh = 0) {
            _isblocked = false;
            _workers = 0;
            _wmlow = wmlow;
            _wmhigh = wmhigh ? wmhigh : wmlow;
            if (_wmhigh) {
                if (!_wmlow)
                    _wmlow = 1;
                else if (_wmlow > _wmhigh)
                    _wmlow = _wmhigh;
                _wmcond = new Condition(_mutex);
            } else
                _wmcond = NULL;
        }

        ~WorkPile() {
            if (_wmcond)
                delete _wmcond;
        }


        T dequeue() {
            T ret;

            lock();
            if (DEQUE::empty()) {
                unlock();
                throw IOException(IOException::errThread, ENOENT, "WorkPile is empty");
            } else {
                if (DEQUE::size() == _wmlow)
                    _wmcond->signal();
                ret = DEQUE::front();
                DEQUE::pop_front();
            }
            unlock();
            return ret;
        }


        std::list<T> dequeue(unsigned count) {
            std::list<T> ret;

            lock();
            if (count) {
                if (_wmlow && DEQUE::size() > _wmlow && DEQUE::size() < _wmlow + count)
                    _wmcond->signal();
                while (count && !DEQUE::empty()) {
                    ret.push_back(DEQUE::front());
                    DEQUE::pop_front();
                    count--;
                }
            } else {
                if (_wmlow)
                    _wmcond->signal();
                while (!DEQUE::empty()) {
                    ret.push_back(DEQUE::front());
                    DEQUE::pop_front();
                }
            }
            unlock();
            return ret;
        }


        T dequeue(Thread *worker) {
            T ret;

            lock();
            if (DEQUE::empty()) {
                _workers++;
                do
                    wait();
                while (DEQUE::empty() && !worker->isCancelled());
                _workers--;
            }
            if (DEQUE::empty()) {
                unlock();
                throw IOException(IOException::errThread, ENOENT, "WorkPile is empty, the thread has received a cancellation request");
            } else {
                if (DEQUE::size() == _wmlow)
                    _wmcond->signal();
                ret = DEQUE::front();
                DEQUE::pop_front();
            }
            unlock();
            return ret;
        }


        size_t enqueue(T t, bool block = true) {
            size_t ret;

            lock();
            if (_isblocked || (_wmhigh && DEQUE::size() >= _wmhigh)) {
                if (!block) {
                    unlock();
                    throw IOException(IOException::errThread, ENOENT, "WorkPile is full");
                }
                _isblocked = true;
                while (DEQUE::size() >= _wmlow)
                    _wmcond->wait();
                _isblocked = false;
            }
            push_back(t);
            if (_workers)
                signal();
            ret = DEQUE::size() > _workers ? DEQUE::size() - _workers : 0;
            unlock();
            return ret;
        }


        unsigned getWorkerCount() {
            unsigned ret;

            lock();
            ret = _workers;
            unlock();
            return ret;
        }


        size_t size() {
            size_t ret;

            lock();
            ret = DEQUE::size();
            unlock();
            return ret;
        }

    private:
        bool       _isblocked; 
        unsigned   _workers;   
        Condition *_wmcond;    
        unsigned   _wmhigh;    
        unsigned   _wmlow;     

        DISALLOW_COPY_CONSTRUCTOR_AND_ASSIGNMENT(WorkPile);
    };
}

#endif