5
\$\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\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.