source: trunk/Mars/mcore/Queue.h@ 17221

#ifndef FACT_Queue
#define FACT_Queue

#include <list>
#include <thread>
#include <condition_variable>

template<class T>
class Queue
{
    size_t fSize;                 // Only necessary for before C++11
    bool   fSort;                 // Sort the list before processing

    std::list<T> fList;

    std::mutex fMutex;             // Mutex needed for the conditional
    std::condition_variable fCond; // Conditional

    enum state_t
    {
        kIdle,
        kRun,
        kStop,
        kAbort,
        kTrigger
    };

    state_t fState;               // Stop signal for the thread

    typedef std::function<bool(const T &)> callback;
    callback fCallback;       // Callback function called by the thread

    std::thread fThread;      // Handle to the thread

    void Thread()
    {
        std::unique_lock<std::mutex> lock(fMutex);

        // No filling allowed by default (the queue is
        // always processed until it is empty)
        size_t allowed = 0;

        while (1)
        {
            while (fSize==allowed && fState==kRun)
                fCond.wait(lock);

            // Check if the State flag has been changed
            if (fState==kAbort)
                break;

            if (fState==kStop && fList.empty())
                break;

            // If thread got just woken up, move back the state to kRun
            if (fState == kTrigger)
                fState = kRun;

            // Could have been a fState==kTrigger case
            if (fList.empty())
                continue;

            // During the unlocked state, fSize might change.
            // The current size of the queue needs to be saved.
            allowed = fSize;

            // get the first entry from the (sorted) list
            const auto it = fSort ? min_element(fList.begin(), fList.end()) : fList.begin();

            // Theoretically, we can loose a signal here, but this is
            // not a problem, because then we detect a non-empty queue
            lock.unlock();

            // If the first event in the queue could not be processed,
            // no further processing makes sense until a new event has
            // been posted (or the number of events in the queue has
            // changed)  [allowed>0], in the case processing was
            // successfull [alloed==0], the next event will be processed
            // immediately.
            if (!fCallback || !fCallback(*it))
                allowed = 0;

            lock.lock();

            // Whenever an event was successfully processed, allowed
            // is larger than zero and thus the event will be popped
            if (allowed==0)
                continue;

            if (fSort)
                fList.erase(it);
            else
                fList.pop_front() ;

            fSize--;
            allowed--;

        }

        fList.clear();
        fSize = 0;

        fState = kIdle;
    }

public:
    Queue(const callback &f, bool sort=false, bool startup=true) : fSize(0), fSort(sort), fState(kIdle), fCallback(f)
    {
        if (startup)
            start();
    }

    Queue(const Queue<T>& q) : fSize(0), fSort(q.fSort), fState(kIdle), fCallback(q.fCallback)
    {
    }

    Queue<T>& operator = (const Queue<T>& q)
    {
        fSize     = 0;
        fSort     = q.fSort;
        fState    = kIdle;
        fCallback = q.fCallback;
        return *this;
    }

    ~Queue()
    {
        wait(true);
    }

    bool start()
    {
        const std::lock_guard<std::mutex> lock(fMutex);
        if (fState!=kIdle)
            return false;

        fState = kRun;
        fThread = std::thread(std::bind(&Queue::Thread, this));
        return true;
    }

    bool stop()
    {
        const std::lock_guard<std::mutex> lock(fMutex);
        if (fState==kIdle)
            return false;

        fState = kStop;
        fCond.notify_one();

        return true;
    }

    bool abort()
    {
        const std::lock_guard<std::mutex> lock(fMutex);
        if (fState==kIdle)
            return false;

        fState = kAbort;
        fCond.notify_one();

        return true;
    }

    bool wait(bool abrt=false)
    {
        {
            const std::lock_guard<std::mutex> lock(fMutex);
            if (fState==kIdle)
                return false;

            if (fState==kRun)
            {
                fState = abrt ? kAbort : kStop;
                fCond.notify_one();
            }
        }

        fThread.join();
        return true;
    }

    bool post(const T &val)
    {
        const std::lock_guard<std::mutex> lock(fMutex);

        if (fState==kIdle)
            return false;

        fList.push_back(val);
        fSize++;

        fCond.notify_one();

        return true;
    }

    bool notify()
    {
        const std::lock_guard<std::mutex> lock(fMutex);
        if (fState!=kRun)
            return false;

        fState = kTrigger;
        fCond.notify_one();

        return true;
    }

#ifdef __GXX_EXPERIMENTAL_CXX0X__
    template<typename... _Args>
        bool emplace(_Args&&... __args)
    {
        const std::lock_guard<std::mutex> lock(fMutex);
        if (fState==kIdle)
            return false;

        fList.emplace_back(__args...);
        fSize++;

        fCond.notify_one();

        return true;
    }

    bool post(T &&val) { return emplace(std::move(val)); }
#endif

#ifdef __GXX_EXPERIMENTAL_CXX0X__
    bool move(std::list<T>&& x, typename std::list<T>::iterator i)
#else
    bool move(std::list<T>& x, typename std::list<T>::iterator i)
#endif
    {
        const std::lock_guard<std::mutex> lock(fMutex);
        if (fState==kIdle)
            return false;

        fList.splice(fList.end(), x, i);
        fSize++;

        fCond.notify_one();

        return true;
    }

#ifdef __GXX_EXPERIMENTAL_CXX0X__
    bool move(std::list<T>& x, typename std::list<T>::iterator i) { return move(std::move(x), i); }
#endif

    size_t size() const
    {
        return fSize;
    }

    bool empty() const
    {
        return fSize==0;
    }
};

#endif