Writing a thread-safe ring queue in C++17

Question

I tried implementing a thread-safe ring queue in C++. I'm totally new to move semantics and C++11/14/17 in general.

#ifndef THREAD_SAFE_RING_QUEUE_HPP_
#define THREAD_SAFE_RING_QUEUE_HPP_

#include <mutex>
#include <optional>
#include <vector>

namespace structure {
class MaximumCapacityReachedError : public std::runtime_error {
 public:
  MaximumCapacityReachedError(std::size_t capacity)
      : runtime_error("Structure instance is full (" +
                      std::to_string(capacity) + ")") {}
};

class EmptyError : public std::runtime_error {
 public:
  EmptyError() : runtime_error("Structure instance is empty") {}
};

template <class T>
class RingQueue {
 public:
  static constexpr std::size_t kDefaultCapacity_ = 10;

  RingQueue(std::size_t = kDefaultCapacity_);

  bool IsEmpty() const;
  std::size_t GetSize() const;
  void Push(T&& element);
  void TryPush(T&& element);
  T& ConsumeNext();
  std::optional<T> TryConsumeNext();
  void Clear();

  std::size_t GetCapacity() const noexcept;
  bool IsFull() const noexcept;

 private:
  mutable std::recursive_mutex mutex_;
  std::size_t capacity_ = 0;
  std::size_t head_ = 0;
  std::size_t tail_ = 0;
  std::vector<T> elements_;
  bool is_full_ = false;
};

template <class T>
inline RingQueue<T>::RingQueue(std::size_t capacity)
    : capacity_(capacity),
      elements_(std::vector<T>(capacity_)),
      is_full_(capacity == 0) {}

template <class T>
inline bool RingQueue<T>::IsEmpty() const {
  std::scoped_lock<std::recursive_mutex> lock(mutex_);
  return capacity_ == 0 || (!is_full_ && head_ == tail_);
}

template <class T>
inline std::size_t RingQueue<T>::GetSize() const {
  std::scoped_lock<std::recursive_mutex> lock(mutex_);

  if (is_full_) {
    return capacity_;
  }

  if (tail_ >= head_) {
    return tail_ - head_;
  }

  return capacity_ - (head_ - tail_);
}

template <class T>
inline void RingQueue<T>::Push(T&& element) {
  std::scoped_lock<std::recursive_mutex> lock(mutex_);

  if (IsFull()) {
    throw MaximumCapacityReachedError(capacity_);
  }

  elements_[tail_] = std::forward<T>(element);
  tail_ = (tail_ + 1) % capacity_;
  is_full_ = head_ == tail_;
}

template <class T>
inline void RingQueue<T>::TryPush(T&& element) {
  std::scoped_lock<std::recursive_mutex> lock(mutex_);

  if (IsFull()) {
    return;
  }

  elements_[tail_] = std::forward<T>(element);
  tail_ = (tail_ + 1) % capacity_;
  is_full_ = head_ == tail_;
}

template <class T>
inline T& RingQueue<T>::ConsumeNext() {
  std::scoped_lock<std::recursive_mutex> lock(mutex_);

  if (IsEmpty()) {
    throw EmptyError();
  }

  const auto previous_head = head_;
  head_ = (head_ + 1) % capacity_;
  is_full_ = false;
  return std::forward<T>(elements_[previous_head]);
}

template <class T>
inline std::optional<T> RingQueue<T>::TryConsumeNext() {
  std::scoped_lock<std::recursive_mutex> lock(mutex_);

  if (IsEmpty()) {
    return std::nullopt;
  }

  const auto previous_head = head_;
  head_ = (head_ + 1) % capacity_;
  is_full_ = false;
  return std::forward<T>(elements_[previous_head]);
}

template <class T>
inline void RingQueue<T>::Clear() {
  std::scoped_lock<std::recursive_mutex> lock(mutex_);

  head_ = 0;
  tail_ = 0;
  is_full_ = false;
}

template <class T>
inline std::size_t RingQueue<T>::GetCapacity() const noexcept {
  std::scoped_lock<std::recursive_mutex> lock(mutex_);
  return capacity_;
}

template <class T>
inline bool RingQueue<T>::IsFull() const noexcept {
  std::scoped_lock<std::recursive_mutex> lock(mutex_);
  return is_full_;
}
}  // namespace structure

#endif  // THREAD_SAFE_RING_QUEUE_HPP_

I have tested my code and it seems to work, but so many things can go wrong...

EDIT: here are the tests that I've done using Catch2. DummyObject is just a class containing an int (the value may be passed in the constructor, and is 0 by default).

#include <memory>
#include "catch.hpp"
#include "dummies.hpp"
#include "utils/structure/ring_queue.hpp"

TEST_CASE("Ring queue creation with default capacity", "[structure]") {
  auto queue = structure::RingQueue<std::shared_ptr<dummy::DummyObject>>();

  SECTION("Properties after creation with default capacity.") {
    REQUIRE(queue.GetCapacity() == queue.kDefaultCapacity_);
    REQUIRE(queue.GetSize() == 0);
    REQUIRE(!queue.IsFull());
    REQUIRE(queue.IsEmpty());
  }
}

TEST_CASE("Ring queue creation with specific capacity", "[structure]") {
  const std::size_t capacity = 15;

  auto queue =
      structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(capacity);

  SECTION("Properties after creation with specific capacity.") {
    REQUIRE(queue.GetCapacity() == capacity);
    REQUIRE(queue.GetSize() == 0);
    REQUIRE(!queue.IsFull());
    REQUIRE(queue.IsEmpty());
  }
}

TEST_CASE("Ring queue push operations in single thread", "[structure]") {
  const std::size_t capacity = 15;

  SECTION("Properties after pushing one element.") {
    auto queue = structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(
        capacity);

    const auto dummy1 = dummy::DummyObject(0);
    queue.Push(std::make_unique<dummy::DummyObject>(0));

    REQUIRE(queue.GetSize() == 1);
    REQUIRE(!queue.IsFull());
    REQUIRE(!queue.IsEmpty());
  }

  SECTION("Properties after pushing two elements.") {
    auto queue = structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(
        capacity);

    queue.Push(std::make_unique<dummy::DummyObject>(0));
    queue.Push(std::make_unique<dummy::DummyObject>(0));
    REQUIRE(queue.GetSize() == 2);
    REQUIRE(!queue.IsFull());
    REQUIRE(!queue.IsEmpty());
  }

  SECTION("Properties after pushing the maximum amount of elements.") {
    auto queue =
        structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(1);

    queue.Push(std::make_unique<dummy::DummyObject>(0));
    REQUIRE(queue.GetSize() == 1);
    REQUIRE(queue.IsFull());
    REQUIRE(!queue.IsEmpty());
  }

  SECTION("Properties after pushing too many elements.") {
    auto queue =
        structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(1);

    queue.Push(std::make_unique<dummy::DummyObject>(0));

    bool is_exception = false;

    try {
      queue.Push(std::make_unique<dummy::DummyObject>(0));
    } catch (const structure::MaximumCapacityReachedError& error) {
      is_exception = true;
    }

    REQUIRE(is_exception);
    REQUIRE(queue.GetSize() == 1);
    REQUIRE(queue.IsFull());
    REQUIRE(!queue.IsEmpty());
  }

  SECTION("Specific error case: pushing one element in a 0-capacity queue.") {
    auto queue =
        structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(0);

    bool is_exception = false;

    try {
      queue.Push(std::make_unique<dummy::DummyObject>(0));
    } catch (const structure::MaximumCapacityReachedError& error) {
      is_exception = true;
    }

    REQUIRE(is_exception);
  }

  SECTION("Try pushing an element.") {
    std::size_t capacity = 5;
    auto queue = structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(
        capacity);

    for (std::size_t index = 0; index < capacity + 1; index++) {
      queue.TryPush(std::make_unique<dummy::DummyObject>(0));
    }

    REQUIRE(queue.GetSize() == capacity);
  }
}

TEST_CASE("Ring queue consume operations in single thread",
          "[structure]") {
  const std::size_t capacity = 15;

  SECTION("Properties after consuming one element.") {
    auto queue = structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(
        capacity);

    queue.Push(std::make_unique<dummy::DummyObject>(42));
    REQUIRE(queue.ConsumeNext()->GetValue() == 42);
    REQUIRE(queue.GetSize() == 0);
    REQUIRE(!queue.IsFull());
    REQUIRE(queue.IsEmpty());
  }

  SECTION("Properties after consuming two elements.") {
    auto queue = structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(
        capacity);

    queue.Push(std::make_unique<dummy::DummyObject>(42));
    REQUIRE(queue.ConsumeNext()->GetValue() == 42);
    REQUIRE(queue.GetSize() == 0);
    REQUIRE(!queue.IsFull());
    REQUIRE(queue.IsEmpty());
  }

  SECTION("Properties after consuming one element in a two-element queue.") {
    auto queue = structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(
        capacity);

    queue.Push(std::make_unique<dummy::DummyObject>(0));
    queue.Push(std::make_unique<dummy::DummyObject>(0));
    REQUIRE(queue.ConsumeNext());
    REQUIRE(queue.GetSize() == 1);
    REQUIRE(!queue.IsFull());
    REQUIRE(!queue.IsEmpty());
  }

  SECTION("Properties after consuming too many elements.") {
    auto queue =
        structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(1);

    queue.Push(std::make_unique<dummy::DummyObject>(42));
    queue.ConsumeNext();
    bool is_exception = false;

    try {
      queue.ConsumeNext();
    } catch (const structure::EmptyError& error) {
      is_exception = true;
    }

    REQUIRE(is_exception);
    REQUIRE(queue.GetSize() == 0);
    REQUIRE(!queue.IsFull());
    REQUIRE(queue.IsEmpty());
  }

  SECTION("Order is respected.") {
    auto queue =
        structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(3);

    queue.Push(std::make_unique<dummy::DummyObject>(1));
    queue.Push(std::make_unique<dummy::DummyObject>(2));
    queue.Push(std::make_unique<dummy::DummyObject>(3));

    REQUIRE(queue.ConsumeNext().get()->GetValue() == 1);
    REQUIRE(queue.ConsumeNext().get()->GetValue() == 2);
    REQUIRE(queue.ConsumeNext().get()->GetValue() == 3);
  }

  SECTION("Specific error case: consuming one element in a 0-capacity queue.") {
    auto queue =
        structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(0);

    bool is_exception = false;

    try {
      queue.ConsumeNext();
    } catch (const structure::EmptyError& error) {
      is_exception = true;
    }

    REQUIRE(is_exception);
  }

  SECTION("Try consuming an element.") {
    std::size_t capacity = 5;
    auto queue = structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(
        capacity);

    for (std::size_t index = 0; index < capacity; index++) {
      queue.Push(std::make_unique<dummy::DummyObject>(0));
    }

    for (std::size_t index = 0; index < capacity + 1; index++) {
      const auto element = queue.TryConsumeNext();
      const bool is_element = element.has_value();
      const bool is_okay = index < capacity ? is_element : !is_element;
      REQUIRE(is_okay);
    }

    REQUIRE(queue.GetSize() == 0);
  }
}

TEST_CASE("Ring queue pushing operations in multiple threads",
          "[structure]") {
  const std::size_t capacity = 15;
  SECTION("Usage in threads: push operations.") {
    std::size_t thread_number = 5;
    auto queue =
        structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(5);

    std::vector<std::thread> threads;

    for (std::size_t index = 0; index < thread_number; index++) {
      threads.push_back(std::thread([&queue]() {
        queue.Push(std::make_unique<dummy::DummyObject>(0));
      }));
    }

    std::for_each(threads.begin(), threads.end(),
                  [](std::thread& thread) { thread.join(); });

    REQUIRE(queue.GetSize() == thread_number);
  }
}

TEST_CASE("Ring queue consume operations in multiple threads",
          "[structure]") {
  SECTION("Usage in threads: consume operations.") {
    std::size_t thread_number = 5;
    auto queue =
        structure::RingQueue<std::shared_ptr<dummy::DummyObject>>(5);

    std::vector<std::thread> threads;

    for (std::size_t index = 0; index < thread_number; index++) {
      queue.Push(std::make_unique<dummy::DummyObject>(0));
    }

    for (std::size_t index = 0; index < thread_number; index++) {
      threads.push_back(std::thread([&queue]() { queue.ConsumeNext(); }));
    }

    std::for_each(threads.begin(), threads.end(),
                  [](std::thread& thread) { thread.join(); });

    REQUIRE(queue.GetSize() == 0);
  }
}
```

Answer 1

ConsumeNext() should not return a reference

The problem with ConsumeNext() is that it returns a reference to an object. However, that element is also marked as consumed. So the caller cannot safely access this element, since it can be overwritten by a call to Push() from another thread.

There are two options to solve this:

1. Have ConsumeNext() return by value, just like TryConsumeNext() does
2. Split ConsumeNext() into a GetNext() that returns a reference but does not consume, and a ConsumeNext() that returns void.

Returning by value has the drawback that T needs to be copied, which can be expensive or impossible, depending on the exact type. So I recommmend going for option 2. This, by the way, is exactly what the STL does for containers such as std::stack.

Consider avoiding the head == tail ambiguity

The problem with having a head and tail index or pointer for a ringbuffer is that when head == tail, this can either be because the ringbuffer is empty or that it is completely full. You solved this by adding an extra variable is_full_, and checking this in many places.

A neater solution in my opinion is to not have a tail_ index, but rather a size_ variable that tracks how many elements are actually in use. This simplifies a lot of code. Of course, you now have to derive the tail index in Push() and TryPush(), like so:

elements_[(head_ + size_++) % capacity_] = std::forward<T>(element);

Useless lock in GetCapacity()

You don't need to take the lock in GetCapacity(), since the capacity can never change after constructing a RingQueue. You can even make capacity_ itself a const variable.

Remove IsEmpty() and IsFull()

The functions IsEmpty() and IsFull() don't return useful information to users of your class. By the time they return true or false, that value might not reflect the actual state of the RingQueue anymore, since another thread might have added or removed elements from it. Keeping them available as public functions invites TOCTTOU bugs.

You might consider keeping them as private functions, but then they don't need to take the lock anymore, and you can change mutex_ from a std::recursive_mutex to the slightly more efficient regular std::mutex.

Naming things

I recommend you give the member functions names that match those of STL containers, in particular those of std::queue, so it is easier use your RingQueues in code that also uses the STL. So:

• Push() -> push() and emplace()
• ConsumeNext() -> front() and pop()
• Clear() -> clear()
• GetCapacity() -> capacity()

Add a function to wait for the possibility to push/pop

It's nice that this function is thread-safe, but that implies you want to use this from multiple threads. In particular, consider a producer-consumer scenario, where one thread wants to add elements to the queue, and another wants to remove them. They might not run at the same speed, so if the queue gets full, you ideally want to let the producer sleep until the queue is no longer full, and conversely, if the queue is empty, you want the consumer to sleep until an element was added. This can be implemented quite easily by using a std::condition_variable.