mpsc_ring_queue.h

Go to the documentation of this file.

 
#ifndef CPPCOMMON_THREADS_MPSC_RING_QUEUE_H
#define CPPCOMMON_THREADS_MPSC_RING_QUEUE_H
 
#include "threads/spin_lock.h"
#include "threads/spsc_ring_queue.h"
#include "time/timestamp.h"
 
#include <cassert>
#include <cstdio>
#include <functional>
#include <memory>
#include <thread>
#include <utility>
#include <vector>
 
namespace CppCommon {
 
 
template<typename T>
class MPSCRingQueue
{
public:
 
    explicit MPSCRingQueue(size_t capacity, size_t concurrency = std::thread::hardware_concurrency());
    MPSCRingQueue(const MPSCRingQueue&) = delete;
    MPSCRingQueue(MPSCRingQueue&&) = delete;
    ~MPSCRingQueue() = default;
 
    MPSCRingQueue& operator=(const MPSCRingQueue&) = delete;
    MPSCRingQueue& operator=(MPSCRingQueue&&) = delete;
 
    explicit operator bool() const noexcept { return !empty(); }
 
    bool empty() const noexcept { return (size() == 0); }
    size_t capacity() const noexcept { return _capacity; }
    size_t concurrency() const noexcept { return _concurrency; }
    size_t size() const noexcept;
 
 
    bool Enqueue(const T& item);
 
    bool Enqueue(T&& item);
 
 
    bool Dequeue(T& item);
 
 
    bool Dequeue(const std::function<void(const T&)>& handler = [](const int&){});
 
private:
    struct Producer
    {
        SpinLock lock;
        SPSCRingQueue<T> queue;
 
        Producer(size_t capacity) : queue(capacity) {}
    };
 
    size_t _capacity;
    size_t _concurrency;
    std::vector<std::shared_ptr<Producer>> _producers;
    size_t _consumer;
};
 
} // namespace CppCommon
 
#include "mpsc_ring_queue.inl"
 
#endif // CPPCOMMON_THREADS_MPSC_RING_QUEUE_H