source: trunk/FACT++/src/queue.h@ 16090

#ifndef FACT_Queue
#define FACT_Queue

#include <thread>
#include <condition_variable>

template<class T>
class Queue : std::deque<T>
{
    std::mutex fMutex;        // Mutex needed for the conditional
    std::condition_variable fCond; // Conditional

    enum state_t
    {
        kRun,
        kWait,
        kStop
    };

    state_t fState;               // Stop signal for the thread

    typedef std::function<void(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);
        while (1)
        {
            while (std::deque<T>::empty() && fState==kRun)
                fCond.wait(lock);

            if (fState==kStop)
                break;

            if (fState==kWait && std::deque<T>::empty())
                break;

            const T val = std::deque<T>::front();
            std::deque<T>::pop_front();

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

            if (fCallback)
                fCallback(val);

            lock.lock();
        }
    }

public:
    Queue(const callback &f) : fState(kRun), fCallback(f)
    {
        fThread = std::thread(std::bind(&Queue::Thread, this));
    }
    ~Queue()
    {
        stop();
        join();
    }

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

        std::deque<T>::push_back(val);
        fCond.notify_one();
    }

    void wait()
    {
        fMutex.lock();
        fState = kWait;
        fCond.notify_one();
        fMutex.unlock();
    }

    void stop()
    {
        fMutex.lock();
        fState = kStop;
        fCond.notify_one();
        fMutex.unlock();
    }

    void join()
    {
        // This can happen is the thread is not running anymore
        try
        {
            fThread.join();
        }
        catch (const std::system_error &)
        {
        }
    }
};

#endif