concurrent_container.hpp

#pragma once
 
#include <chrono>
#include <condition_variable>
#include <mutex>
 
namespace fly::detail {
 
template <typename T, typename Container>
class ConcurrentContainer
{
public:
    using size_type = typename Container::size_type;
    using value_type = T;
 
    virtual ~ConcurrentContainer() = default;
 
    void push(T &&item);
 
    void pop(T &item);
 
    template <typename R, typename P>
    bool pop(T &item, std::chrono::duration<R, P> duration);
 
    bool empty() const;
 
    size_type size() const;
 
protected:
    virtual void push_internal(T &&item) = 0;
 
    virtual void pop_internal(T &item) = 0;
 
    mutable std::mutex m_container_mutex;
    Container m_container;
 
private:
    std::condition_variable m_push_condition;
};
 
//==================================================================================================
template <typename T, typename Container>
void ConcurrentContainer<T, Container>::push(T &&item)
{
    {
        std::unique_lock<std::mutex> lock(m_container_mutex);
        push_internal(std::move(item));
    }
 
    m_push_condition.notify_one();
}
 
//==================================================================================================
template <typename T, typename Container>
void ConcurrentContainer<T, Container>::pop(T &item)
{
    std::unique_lock<std::mutex> lock(m_container_mutex);
 
    while (m_container.empty())
    {
        m_push_condition.wait(lock);
    }
 
    pop_internal(item);
}
 
//==================================================================================================
template <typename T, typename Container>
template <typename R, typename P>
bool ConcurrentContainer<T, Container>::pop(T &item, std::chrono::duration<R, P> wait_time)
{
    std::unique_lock<std::mutex> lock(m_container_mutex);
 
    auto empty_test = [this] {
        return !m_container.empty();
    };
 
    bool item_popped = m_push_condition.wait_for(lock, wait_time, std::move(empty_test));
 
    if (item_popped)
    {
        pop_internal(item);
    }
 
    return item_popped;
}
 
//==================================================================================================
template <typename T, typename Container>
bool ConcurrentContainer<T, Container>::empty() const
{
    std::unique_lock<std::mutex> lock(m_container_mutex);
    return m_container.empty();
}
 
//==================================================================================================
template <typename T, typename Container>
auto ConcurrentContainer<T, Container>::size() const -> size_type
{
    std::unique_lock<std::mutex> lock(m_container_mutex);
    return m_container.size();
}
 
} // namespace fly::detail