CircularQueue.hpp

Go to the documentation of this file.

#ifndef IV_SRC_CONTAINERS_CIRCULARQUEUE_HPP_
#define IV_SRC_CONTAINERS_CIRCULARQUEUE_HPP_
 
#include <array>
#include <condition_variable>
#include <functional>
#include <mutex>
#include <type_traits>
 
namespace iv::containers
{
 
template<typename T, uint64_t Size>
class CircularQueue
{
public:
    CircularQueue() = default;
    CircularQueue(const CircularQueue &) = delete;
 
    virtual ~CircularQueue()
    {
        clear();
    }
 
    CircularQueue &operator=(const CircularQueue &other) = delete;
 
    void pop(T &item)
    {
        std::unique_lock<std::mutex> lock(m_mutex);
 
        if (unblockingEmpty())
        {
            m_cond.wait(lock);
        }
 
        item = m_data[m_front];
        unblockingPop();
    }
 
    void pop()
    {
        std::unique_lock<std::mutex> lock(m_mutex);
 
        if (unblockingEmpty())
        {
            m_cond.wait(lock);
        }
 
        unblockingPop();
    }
 
    bool popWithFilter(T &item, const std::function<bool(const T &)> &filter)
    {
        std::unique_lock<std::mutex> lock(m_mutex);
 
        if (unblockingEmpty())
        {
            m_cond.wait(lock);
        }
 
        while (not unblockingEmpty() and not filter(m_data[m_front]))
        {
            unblockingPop();
        }
 
        if (not unblockingEmpty())
        {
            item = m_data[m_front];
            unblockingPop();
            return true;
        }
 
        return false;
    }
 
    void get(T &item)
    {
        std::unique_lock<std::mutex> lock(m_mutex);
 
        if (unblockingEmpty())
        {
            m_cond.wait(lock);
        }
 
        item = m_data[m_front];
    }
 
    virtual bool push(const T &item)
    {
        std::unique_lock<std::mutex> lock(m_mutex);
 
        if (not unblockingFull())
        {
            m_data[m_rear] = item;
            m_rear = (m_rear + 1) % Size;
            ++m_occupied;
            m_cond.notify_one();
 
            return true;
        }
 
        return false;
    }
 
    void clear()
    {
        std::unique_lock<std::mutex> lock(m_mutex);
 
        while (not unblockingEmpty())
        {
            if constexpr (std::is_pointer_v<T>)
            {
                delete &m_data[m_front];
                m_data[m_front] = nullptr;
            }
 
            unblockingPop();
        }
    }
 
    bool empty()
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        return unblockingEmpty();
    }
 
    bool full()
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        return unblockingFull();
    }
 
    uint64_t size()
    {
        return Size;
    }
 
    uint64_t occupied()
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        return m_occupied;
    }
 
private:
    [[nodiscard]] bool unblockingFull() const
    {
        return m_occupied == Size;
    }
 
    [[nodiscard]] bool unblockingEmpty() const
    {
        return m_occupied == 0;
    }
 
    void unblockingPop()
    {
        m_front = (m_front + 1) % Size;
        --m_occupied;
    }
 
    static_assert(std::is_default_constructible_v<T>, "T must be default constructible");
    static_assert(Size >= 50, "Size is too small!");
    static_assert(Size <= 5000, "Size is too long!");
 
    std::array<T, Size> m_data;
    uint64_t m_front {0};
    uint64_t m_rear {0};
    uint64_t m_occupied {0};
    std::mutex m_mutex;
    std::condition_variable_any m_cond;
};
 
template<typename T, uint64_t Size>
class OverwritingCircularQueue : public CircularQueue<T, Size>
{
public:
    OverwritingCircularQueue() = default;
    OverwritingCircularQueue(const OverwritingCircularQueue &) = delete;
 
    bool push(const T &item) override
    {
        bool didPop {false};
        while (not CircularQueue<T, Size>::push(item))
        {
            didPop = true;
            CircularQueue<T, Size>::pop();
        }
 
        return didPop;
    }
};
 
}// namespace iv::containers
 
#endif//IV_SRC_CONTAINERS_CIRCULARQUEUE_HPP_