flow_concurrent_flex_queue.h

Go to the documentation of this file.

#pragma once
#include <condition_variable>
#include <memory>
#include <mutex>
#include <optional>
 
namespace flow {

  template <typename T>
  class ConcurrentFlexQueue {
    struct Node {
      std::optional<T> value;
      std::unique_ptr<Node> next;
    };
 
    std::unique_ptr<Node> head_;
    Node* tail_;
    mutable std::mutex headMux_;
    mutable std::mutex tailMux_;
    std::condition_variable blocked_;
 
  public:
    ConcurrentFlexQueue()
      : head_(std::make_unique<Node>()), tail_(head_.get()) {
    }
 
    ConcurrentFlexQueue(const ConcurrentFlexQueue&) = delete;
    ConcurrentFlexQueue(ConcurrentFlexQueue&&) = delete;
    ConcurrentFlexQueue& operator=(const ConcurrentFlexQueue&) = delete;
    ConcurrentFlexQueue& operator=(ConcurrentFlexQueue&&) = delete;

    bool empty() const {
      std::scoped_lock lock(headMux_, tailMux_);
      return head_.get() == tail_;
    }

    void push(const T& value) {
      auto dummy = std::make_unique<Node>();
      {
        std::lock_guard lock(tailMux_);
        tail_->value = value;
        tail_->next = std::move(dummy);
        tail_ = tail_->next.get();
      }
      blocked_.notify_one();
    }

    void push(T&& value) {
      auto dummy = std::make_unique<Node>();
      {
        std::lock_guard lock(tailMux_);
        tail_->value = std::move(value);
        tail_->next = std::move(dummy);
        tail_ = tail_->next.get();
      }
      blocked_.notify_one();
    }

    template <typename... Args>
    void emplace(Args&&... args) {
      auto dummy = std::make_unique<Node>();
      {
        std::lock_guard lock(tailMux_);
        tail_->value.emplace(std::forward<Args>(args)...);
        tail_->next = std::move(dummy);
        tail_ = tail_->next.get();
      }
      blocked_.notify_one();
    }

    std::optional<T> tryFront() const {
      std::unique_lock lock(headMux_);
      if (head_.get() == tail_) {
        return std::nullopt;
      }
      return head_->value;
    }

    std::optional<T> tryPop() {
      std::unique_lock lock(headMux_);
      if (head_.get() == tail_) {
        return std::nullopt;
      }
 
      auto oldHead = std::move(head_);
      head_ = std::move(oldHead->next);
      return std::move(oldHead->value);
    }

    T waitPop() {
      std::unique_lock lock(headMux_);
      blocked_.wait(lock, [&]() { return head_.get() != tail_; });
 
      auto oldHead = std::move(head_);
      head_ = std::move(oldHead->next);
      return oldHead->value.value();
    }
  };
}