thread_safe_queue.h

Go to the documentation of this file.

#pragma once
 
#include <condition_variable>
#include <mutex>
#include <optional>
#include <queue>
 
namespace mm::core::containers {

template <typename T>
class ThreadSafeQueue {
 private:
  std::queue<T> queue_;

  size_t capacity_;

  mutable std::mutex mtx_;

  std::condition_variable not_empty_cv_;

  std::condition_variable not_full_cv_;
 
 public:
  explicit ThreadSafeQueue(size_t capacity = 0) : capacity_(capacity) {}
 
  ThreadSafeQueue(const ThreadSafeQueue&) = delete;
  ThreadSafeQueue& operator=(const ThreadSafeQueue&) = delete;

  void push(T value) {
    std::unique_lock<std::mutex> lock(mtx_);
    if (capacity_ > 0) {
      not_full_cv_.wait(lock, [this] { return queue_.size() < capacity_; });
    }
    queue_.push(std::move(value));
    not_empty_cv_.notify_one();
  }

  bool try_push(T value) {
    std::lock_guard<std::mutex> lock(mtx_);
    if (capacity_ > 0 && queue_.size() >= capacity_) return false;
    queue_.push(std::move(value));
    not_empty_cv_.notify_one();
    return true;
  }

  T wait_and_pop() {
    std::unique_lock<std::mutex> lock(mtx_);
    not_empty_cv_.wait(lock, [this] { return !queue_.empty(); });
    T value = std::move(queue_.front());
    queue_.pop();
    not_full_cv_.notify_one();
    return value;
  }

  std::optional<T> try_pop() {
    std::lock_guard<std::mutex> lock(mtx_);
    if (queue_.empty()) return std::nullopt;
    T value = std::move(queue_.front());
    queue_.pop();
    not_full_cv_.notify_one();
    return value;
  }

  bool empty() const {
    std::lock_guard<std::mutex> lock(mtx_);
    return queue_.empty();
  }

  size_t size() const {
    std::lock_guard<std::mutex> lock(mtx_);
    return queue_.size();
  }

  size_t capacity() const { return capacity_; }
 
  // -------- Moodycamel-style API --------
  // Provides methods (enqueue, try_enqueue, emplace, try_emplace, wait_dequeue,
  // try_dequeue, peek) that match moodycamel::ReaderWriterQueue for easy
  // replacement.

  void enqueue(T value) { push(std::move(value)); }

  bool try_enqueue(T value) { return try_push(std::move(value)); }

  template <typename... Args>
  void emplace(Args&&... args) {
    std::unique_lock<std::mutex> lock(mtx_);
    if (capacity_ > 0) {
      not_full_cv_.wait(lock, [this] { return queue_.size() < capacity_; });
    }
    queue_.emplace(std::forward<Args>(args)...);
    not_empty_cv_.notify_one();
  }

  template <typename... Args>
  bool try_emplace(Args&&... args) {
    std::lock_guard<std::mutex> lock(mtx_);
    if (capacity_ > 0 && queue_.size() >= capacity_) {
      return false;
    }
    queue_.emplace(std::forward<Args>(args)...);
    not_empty_cv_.notify_one();
    return true;
  }

  T wait_dequeue() { return wait_and_pop(); }

  bool try_dequeue(T& out) {
    auto opt = try_pop();
    if (!opt) {
      return false;
    }
    out = std::move(*opt);
    return true;
  }

  std::optional<T> try_dequeue() { return try_pop(); }

  bool peek(T& out) const {
    std::lock_guard<std::mutex> lock(mtx_);
    if (queue_.empty()) {
      return false;
    }
    out = queue_.front();  // copy or move depending on T
    return true;
  }

  std::optional<T> peek() const {
    std::lock_guard<std::mutex> lock(mtx_);
    if (queue_.empty()) {
      return std::nullopt;
    }
    return queue_.front();  // copy
  }
};
 
}  // namespace mm::core::containers