ConcurrentQueue.hh

Go to the documentation of this file.

#ifndef __CONCURRENT_QUEUE_H__
#define __CONCURRENT_QUEUE_H__

#include <queue>
#include <mutex>
#include <condition_variable>
#include <chrono>

template<typename T>

class ConcurrentQueue
{
    std::condition_variable cond_var;
    std::mutex mutex;
    std::queue<T> queue;
    int max_size;

    public: ConcurrentQueue(int size):
        max_size(size) {}

    public: bool enqueue(T data)
    {
        bool success = false;
        std::chrono::milliseconds timeout(200);
        std::unique_lock<std::mutex> lock(mutex);
        if (cond_var.wait_for(lock, timeout, [this](){ return !isFull();}))
        {
            queue.push(data);
            success = true;
        }
        lock.unlock();
        cond_var.notify_all();
        return success;
    }

    public: bool dequeue(T &data)
    {
        bool success = false;
        std::chrono::milliseconds timeout(200);
        std::unique_lock<std::mutex> lock(mutex);
        if (cond_var.wait_for(lock, timeout, [this](){ return !isEmpty();}))
        {
            data = queue.front();
            queue.pop();
            success = true;
        }
        lock.unlock();
        cond_var.notify_all();
        return success;   
    }

    public: void clear()
    {
        std::unique_lock<std::mutex> lock(mutex);
        while(!isEmpty()) {
            queue.pop();
        }
        lock.unlock();
        cond_var.notify_all();
    }
    
    private: bool isFull() const
    {
        bool full = queue.size() >= max_size;
        return queue.size() >= max_size;
    }

    private: bool isEmpty() const
    {
        return queue.size() == 0;
    }

    public: int length() const
    {
        return queue.size();
    }
};

#endif