8
\$\begingroup\$

Here's an implementation of a multi-producer single consumer queue that I wanted to use with tasks such as logging from multiple points in a program to a single sink. The implementation is inspired from https://github.com/dbittman/waitfree-mpsc-queue but modified mostly to use C++17 features. Comments? Specifically:

  • Are there race conditions that I am missing?
  • Is this the best I could do if I wanted to avoid copying data as they are pushed and popped?
  • Could it be done in a way that is simpler, cleaner, more maintainable?

Some design choices that I made:

  • API is essentially tryPush() and tryPop(). They are non-blocking - easy to make blocking versions by wrapping them around in a while() statement in user code.

  • Fixed capacity buffer to avoid memory allocation and associated complexities of maintaining a dynamically growing buffer. I want to be able to use this at least in soft real-time applications.

The header-only class implementation of the queue mpscq.h:

#ifndef MPSCQ_H
#define MPSCQ_H

#include <atomic>
#include <cassert>
#include <cinttypes>
#include <cstddef>
#include <optional>
#include <vector>

/// multi-producer, single consumer queue.
/// Inspired from https://github.com/dbittman/waitfree-mpsc-queue
template <typename T, size_t capacity_>
class mpscq
{
public:
  mpscq();
  bool tryPush(T &&obj);
  std::optional<T> tryPop();
  size_t count() const;

private:
  std::atomic<size_t> count_{0};
  std::atomic<size_t> head_{0};
  size_t tail_{0};
  std::vector<std::atomic<bool>> is_readable_;
  std::vector<T> buffer_;
};

template <typename T, size_t capacity_>
mpscq<T, capacity_>::mpscq() : is_readable_(capacity_), buffer_(capacity_)
{
  assert(capacity_ >= 1);
  for (auto &i : is_readable_) {
    i = false;
  }
}

/// Attempt to enqueue without blocking. This is safe to call from multiple threads.
/// \return true on success and false if queue is full.
template <typename T, size_t capacity_>
bool mpscq<T, capacity_>::tryPush(T &&obj)
{
  auto count = count_.fetch_add(1, std::memory_order_acquire);
  if (count >= capacity_) {
    // back off, queue is full
    count_.fetch_sub(1, std::memory_order_release);
    return false;
  }

  // increment the head, which gives us 'exclusive' access to that element until
  // is_reabable_ flag is set
  const auto head = head_.fetch_add(1, std::memory_order_acquire) % capacity_;
  buffer_[head] = std::move(obj);
  assert(is_readable_[head] == false);
  is_readable_[head].store(true, std::memory_order_release);
  return true;
}

/// Attempt to dequeue without blocking
/// \note: This is not safe to call from multiple threads.
/// \return A valid item from queue if the operation won't block, else nothing
template <typename T, size_t capacity_>
std::optional<T> mpscq<T, capacity_>::tryPop()
{
  if (!is_readable_[tail_].load(std::memory_order_acquire)) {
    // A thread could still be writing to this location
    return {};
  }

  assert(is_readable_[tail_]);
  auto ret = std::move(buffer_[tail_]);
  is_readable_[tail_].store(false, std::memory_order_release);

  if (++tail_ >= capacity_) {
    tail_ = 0;
  }

  const auto count = count_.fetch_sub(1, std::memory_order_release);
  assert(count > 0);
  return ret;
}

/// \return The number of items in queue
template <typename T, size_t capacity_>
size_t mpscq<T, capacity_>::count() const
{
  return count_.load(std::memory_order_relaxed);
}

#endif // MPSCQ_H

The example program to test it mpscq_example.cpp:

#include "mpscq.h"
#include <chrono>
#include <csignal>
#include <cstring>
#include <iostream>
#include <thread>

struct HeavyObject
{
  HeavyObject(uint32_t val = 0) { value[0] = val; }
  uint32_t value[128];
};

constexpr size_t Q_LEN = 1000;
static mpscq<HeavyObject, Q_LEN> s_queue;
static std::atomic_bool s_exit = false;
static std::atomic_uint s_value = 0;

void onSignal(int signum)
{
  std::cout << "Received signal " << strsignal(signum) << std::endl;
  s_exit = true;
}

void producer()
{
  while (!s_exit) {
    HeavyObject obj(s_value.load());
    auto pushed = s_queue.tryPush(std::move(obj));
    std::this_thread::sleep_for(std::chrono::microseconds(1));
    std::this_thread::yield();
    if (pushed) {
      std::cout << std::hex << std::this_thread::get_id() << " produced - value: " << s_value << "\n";
      ++s_value;
    }
  }
}

void consumer()
{
  while (!s_exit) {
    std::this_thread::yield();
    std::this_thread::sleep_for(std::chrono::microseconds(1));
    if (s_queue.count() > 0) {
      auto t = s_queue.tryPop();
      if (t.has_value()) {
        auto val_now = t.value().value[0];
        std::cout << "consumer - value: " << val_now << "\n";
      }
    }
  }
}

int main()
{
  signal(SIGINT, onSignal);
  signal(SIGTERM, onSignal);

  std::thread t1(producer);
  std::thread t2(producer);
  consumer();
  t1.join();
  t2.join();
  return 0;
}

And the CMakeLists.txt file to build it

cmake_minimum_required(VERSION 3.10)
project(mpscq LANGUAGES CXX)

set(CMAKE_CXX_STANDARD 17) # Turn on C++17 compile flags
set(CMAKE_CXX_STANDARD_REQUIRED ON) # Yes we really need it
set(CMAKE_CXX_EXTENSIONS OFF) # Turn off non-standard extensions to ISO C++
set(CMAKE_POSITION_INDEPENDENT_CODE ON)

set(THREADS_PREFER_PTHREAD_FLAG ON)
find_package(Threads REQUIRED)

add_executable(mpscq_example mpscq_example.cpp mpscq.h)
target_link_libraries(mpscq_example Threads::Threads)

```
\$\endgroup\$
1
  • \$\begingroup\$ Are you still interested in a review / constructive communication? I need something like this myself, so perhaps we can make something up. So far I just read through the code once and it seems OK to me. I dislike a few things, but I don't see a bug. \$\endgroup\$
    – Carlo Wood
    Commented Mar 18, 2021 at 20:03

1 Answer 1

2
\$\begingroup\$

We have misspelt std::size_t throughout (this is likely masked by using a <cstdint> implementation that also defines names in the global namespace; it's not portable to rely on that).

The public interface could use some blocking push and pop operations (perhaps with a timeout), so the producers and consumers don't have to continuously poll.

I'm not a big fan of parallel vectors - a single container of pairs may be a better choice (e.g. for locality of access). A vector seems overkill, given that the necessary size is constexpr - why not std::array?

The implementation requires that T is default-constructible. We really should be dealing in uninitialised memory and placement new/delete as we are managing a container here.

assert(capacity_ >= 1); can (and should) be a static_assert(). No reason to wait until runtime for a compile-time constraint.

head_ gets incremented but never reset to 0, unlike tail. This means that eventually we'll overflow, and if the overflow value isn't a multiple of capacity_, we'll get incorrect results. I think we need to perform an atomic subtract each time we wrap around, rather than continually incrementing and using the modulo value.


In the test program, we misspell std::uint32_t. Do we need an exact-width type, or would std::uint_fast32_t be more appropriate?

The signal handler onSignal() uses non-signal-safe I/O.

There's no need to explicitly yield() in the producers and consumer, since sleep_for() will yield anyway, I think.

\$\endgroup\$
1
  • \$\begingroup\$ Visiting this page after many year as I am considering using this code again. Thanks for the brilliant feedback. \$\endgroup\$ Commented Jan 4 at 9:32

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.