flow_concurrent_queue.h

Go to the documentation of this file.

#pragma once
#include <condition_variable>
#include <deque>
#include <mutex>
#include <optional>
#include <queue>
 
namespace flow {
  template <typename T, typename Container = std::deque<T>>
  class ConcurrentQueue {
    using allocator_type = typename Container::allocator_type;
 
    mutable std::mutex mux_;
    std::condition_variable blocked_;
    std::queue<T, Container> queue_;
 
  public:
    ConcurrentQueue() = default;

    explicit ConcurrentQueue(const allocator_type& allocator)
      : queue_(allocator) {
    }
 
    ConcurrentQueue(const ConcurrentQueue&) = delete;
    ConcurrentQueue(ConcurrentQueue&&) = delete;
    ConcurrentQueue& operator=(const ConcurrentQueue&) = delete;
    ConcurrentQueue& operator=(ConcurrentQueue&&) = delete;

    bool empty() const {
      std::lock_guard lock(mux_);
      return queue_.empty();
    }

    std::size_t size() const {
      std::lock_guard lock(mux_);
      return queue_.size();
    }

    void push(const T& value) {
      {
        std::lock_guard lock(mux_);
        queue_.push(value);
      }
      blocked_.notify_one();
    }

    void push(T&& value) {
      {
        std::lock_guard lock(mux_);
        queue_.push(std::move(value));
      }
      blocked_.notify_one();
    }

    template <typename... Args>
    void emplace(Args&&... args) {
      {
        std::lock_guard lock(mux_);
        queue_.emplace(std::forward<Args>(args)...);
      }
      blocked_.notify_one();
    }

    std::optional<T> tryFront() const {
      std::lock_guard lock(mux_);
      if (queue_.empty()) {
        return std::nullopt;
      }
      return queue_.front();
    }

    std::optional<T> tryPop() {
      std::lock_guard lock(mux_);
      if (queue_.empty()) {
        return std::nullopt;
      }
      std::optional<T> value = std::move(queue_.front());
 
      queue_.pop();
      return value;
    }

    T waitPop() {
      std::unique_lock lock(mux_);
      blocked_.wait(lock, [&]() { return !queue_.empty(); });
 
      T value = std::move(queue_.front());
      queue_.pop();
      return value;
    }
  };
}