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_;

  bool aborted_ = false;

  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 abort() {
    std::unique_lock<std::mutex> lock(mtx_);
    aborted_ = true;
    not_empty_cv_.notify_all();  // Force wake-up
    not_full_cv_.notify_all();
  }

  bool aborted() const {
    std::lock_guard<std::mutex> lock(mtx_);
    return aborted_;
  }

  bool push(T value) {
    std::unique_lock<std::mutex> lock(mtx_);
 
    if (capacity_ > 0) {
      not_full_cv_.wait(
          lock, [this] { return queue_.size() < capacity_ || aborted_; });
    }
 
    if (aborted_) {
      return false;  // Do not push if aborted
    }
 
    queue_.push(std::move(value));
    not_empty_cv_.notify_one();
    return true;
  }

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

  void wait_for_data() {
    std::unique_lock<std::mutex> lock(mtx_);
    not_empty_cv_.wait(lock, [this] { return !queue_.empty() || aborted_; });
  }

  void notify_all() {
    std::lock_guard<std::mutex> lock(mtx_);
    not_empty_cv_.notify_all();
    not_full_cv_.notify_all();
  }

  std::optional<T> wait_and_pop() {
    std::unique_lock<std::mutex> lock(mtx_);
 
    not_empty_cv_.wait(lock, [this] { return !queue_.empty() || aborted_; });
 
    if (queue_.empty()) {
      return std::nullopt;
    }
 
    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>
  bool emplace(Args&&... args) {
    std::unique_lock<std::mutex> lock(mtx_);
 
    if (capacity_ > 0) {
      not_full_cv_.wait(
          lock, [this] { return queue_.size() < capacity_ || aborted_; });
    }
 
    if (aborted_) {
      return false;
    }
 
    queue_.emplace(std::forward<Args>(args)...);
    not_empty_cv_.notify_one();
    return true;
  }

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

  std::optional<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
  }

  template <typename Predicate>
  void removeIf(Predicate pred) {
    std::lock_guard<std::mutex> lock(mtx_);
    std::queue<T> newQueue;
 
    while (!queue_.empty()) {
      T& item = queue_.front();
      if (!pred(item)) {
        newQueue.push(std::move(item));
      }
      queue_.pop();
    }
 
    queue_ = std::move(newQueue);
  }
};
 
}  // namespace mm::core::containers