wait_queue.hpp

 
#ifndef WAIT_QUEUE_HPP_INCLUDED
#define WAIT_QUEUE_HPP_INCLUDED
 
#include <cassert> // assert
#include <deque>
#include <mutex> // std::scoped_lock, std::mutex
#include <condition_variable>
#include <stop_token> // std::stop_source, std::stop_token
#include <optional>
#include <utility> // std::move, std::move_if_noexcept, std::forward
#include <type_traits> // for requires clauses and noexcept specs
// #include <concepts> 
 
namespace chops {
 
// requirements for wait_queue container
 
template <typename Ctr, typename T>
concept supports_push_back = requires (Ctr ctr, T val) {
  ctr.push_back(val);
};
 
template <typename Ctr, typename ... Args>
concept supports_emplace_back = requires (Ctr ctr, Args&& ... args) {
  ctr.emplace_back(args ...);
};
 
template <typename Ctr>
concept supports_empty = requires (Ctr ctr) {
  ctr.empty();
};
 
template <typename Ctr>
concept supports_pop_front = requires (Ctr ctr) {
  ctr.pop_front();
};
 
template <typename Ctr>
concept supports_size = requires (Ctr ctr) {
  ctr.size();
};
 
template <typename T, typename Container = std::deque<T> >
  requires std::is_copy_constructible_v<T> || std::is_move_constructible_v<T>
class wait_queue {
private:
  mutable std::mutex              m_mut;
  std::optional<std::stop_source> m_stop_src;
  std::stop_token                 m_stop_tok;
  std::condition_variable_any     m_data_cond;
  Container                       m_data_queue;
 
  using lock_guard = std::scoped_lock<std::mutex>;
 
public:
 
  using size_type = typename Container::size_type;
  using value_type = T;
 
public:
 
  wait_queue()
    requires std::is_default_constructible_v<Container>
    // noexcept(std::is_nothrow_constructible_v<Container>)
      : m_stop_src(std::stop_source{}), m_stop_tok((*m_stop_src).get_token()) 
  {
    assert(empty());
    assert(size() == size_type(0));
    assert(!stop_requested());
  }
 
  wait_queue(std::stop_token stop_tok)
    requires std::is_default_constructible_v<Container>
    // noexcept(std::is_nothrow_constructible_v<Container, std::stop_token>)
      : m_stop_tok(stop_tok) 
  {
    assert(empty());
    assert(size() == size_type(0));
  }
 
  wait_queue(Container&& container)
    requires std::is_move_constructible_v<Container> ||
             std::is_copy_constructible_v<Container>
    // noexcept(std::is_nnthrow_constructible_v<Container, Container&&>)
      : m_stop_src(std::stop_source{}), m_stop_tok((*m_stop_src).get_token()),
    m_data_queue(std::move(container)) 
  {
    // not easily assertible until contracts added to C++
  }
 
  wait_queue(std::stop_token stop_tok, Container&& container)
    requires std::is_move_constructible_v<Container> ||
             std::is_copy_constructible_v<Container>
    // noexcept(std::is_nothrow_constructible_v<Container, std::stop_token, Container&&>)
      : m_stop_tok(stop_tok), m_data_queue(std::move(container)) 
  {
  }
 
  wait_queue(size_type sz)
    requires std::is_constructible_v<Container, size_type>
    // noexcept(std::is_nothrow_constructible_v<Container, size_type>)
      : m_stop_src(std::stop_source{}), m_stop_tok((*m_stop_src).get_token()),
    m_data_queue(sz) 
  {
    assert((sz != size_type(0)) || empty());
    assert((size() == size_type(0)) || (size() == sz));
  }
 
  wait_queue(std::stop_token stop_tok, size_type sz)
    requires std::is_constructible_v<Container, std::stop_token, size_type>
    // noexcept(std::is_nothrow_constructible_v<Container, std::stop_token, size_type>)
      : m_stop_tok((*m_stop_src).get_token()), m_data_queue(sz) 
  {
    assert((sz != size_type(0)) || empty());
    assert((size() == size_type(0)) || (size() == sz));
  }
 
  // disallow copy or move construction of the entire object
  wait_queue(const wait_queue&) = delete;
  wait_queue(wait_queue&&) = delete;
 
  // disallow copy or move assigment of the entire object
  wait_queue& operator=(const wait_queue&) = delete;
  wait_queue& operator=(wait_queue&&) = delete;
 
  // modifying methods
 
  auto request_stop() noexcept 
        -> bool 
  {
    if (m_stop_src) {
      return (*m_stop_src).request_stop();
    }
    return false;
  }
 
  auto push(const T& val) /* noexcept(std::is_nothrow_copy_constructible_v<T>) */
        -> bool 
    requires supports_push_back<Container, T>
 
  {
    if (m_stop_tok.stop_requested()) {
      return false;
    }
    lock_guard lk{m_mut};
    m_data_queue.push_back(val);
    m_data_cond.notify_one();
    return true;
 
  }
 
  auto push(T&& val) /* noexcept(std::is_nothrow_move_constructible_v<T>) */ 
        -> bool
    requires supports_push_back<Container, T>
 
  {
    if (m_stop_tok.stop_requested()) {
      return false;
    }
    lock_guard lk{m_mut};
    m_data_queue.push_back(std::move(val));
    m_data_cond.notify_one();
    return true;
 
  }
 
  template <typename ... Args>
  auto emplace_push(Args &&... args) /* noexcept(std::is_nothrow_constructible_v<T, Args...>)*/
        -> bool 
    requires supports_emplace_back<Container, Args...>
 
  {
    if (m_stop_tok.stop_requested()) {
      return false;
    }
    lock_guard lk{m_mut};
    m_data_queue.emplace_back(std::forward<Args>(args)...);
    m_data_cond.notify_one();
    return true;
  }
 
  [[nodiscard]] auto wait_and_pop() /* noexcept(std::is_nothrow_constructible_v<T>) */
        -> std::optional<T> 
      requires supports_empty<Container> && supports_pop_front<Container>
 
  {
    std::unique_lock<std::mutex> lk{m_mut};
    if (!m_data_cond.wait ( lk, m_stop_tok, [this] { return !m_data_queue.empty(); } )) {
      return std::optional<T> {}; // queue was request to stop, no data available
    }
    assert(!m_data_queue.empty());
#ifndef NDEBUG
    const auto old_size = m_data_queue.size();
#endif
    std::optional<T> val {std::move_if_noexcept(m_data_queue.front())}; // move construct if possible
    m_data_queue.pop_front();
    assert(m_data_queue.size() + 1u == old_size);
    return val;
 
  }
 
  [[nodiscard]] auto try_pop() /* noexcept(std::is_nothrow_constructible_v<T>) */
        -> std::optional<T> 
      requires supports_empty<Container> && supports_pop_front<Container>
  {
    if (m_stop_tok.stop_requested()) {
      return std::optional<T> {};
    }
    lock_guard lk{m_mut};
    if (m_data_queue.empty()) {
      return std::optional<T> {};
    }
#ifndef NDEBUG
    const auto old_size = m_data_queue.size();
#endif
    std::optional<T> val {std::move_if_noexcept(m_data_queue.front())}; // move construct if possible
    m_data_queue.pop_front();
    assert(m_data_queue.size() + 1u == old_size);
    return val;
 
  }
 
  // non-modifying methods
 
  template <typename F>
  auto apply(F&& func) const /* noexcept(std::is_nothrow_invocable_v<F&&, const T&>) */
        -> void 
      requires std::is_invocable_v<F, T>
 
  {
    lock_guard lk{m_mut};
    for (const T& elem : m_data_queue) {
      func(elem);
    }
 
  }
 
  [[nodiscard]] auto stop_requested() const noexcept
        -> bool 
 
  {
    return m_stop_tok.stop_requested();
  }
 
  [[nodiscard]] auto empty() const /* noexcept */
        -> bool 
      requires supports_empty<Container>
 
  {
    lock_guard lk{m_mut};
    return m_data_queue.empty();
 
  }
 
  [[nodiscard]] auto size() const /* noexcept */
        -> size_type 
      requires supports_size<Container>
 
  {
    lock_guard lk{m_mut};
    return m_data_queue.size();
 
  }
 
};
 
} // end namespace
 
#endif