mpmc_queue_base.cpp

#include "mpmc_queue_base.hpp"
 
#include <algorithm>
#include <new>
 
#include "libxr_mem.hpp"
 
namespace LibXR
{
MPMCQueueBase::MPMCQueueBase(size_t element_size, size_t capacity)
    : element_size_(element_size),
      capacity_(capacity),
      payload_stride_(AlignUpChecked(element_size_, alignof(size_t))),
      sequences_(nullptr),
      payloads_(nullptr),
      head_(0),
      tail_(0)
{
  REQUIRE(element_size_ > 0);
  REQUIRE(capacity_ > 1);
  REQUIRE(capacity_ <= static_cast<size_t>(std::numeric_limits<SequenceDiffType>::max()));
 
  const size_t payload_bytes = MultiplyChecked(payload_stride_, capacity_);
  sequences_ = new (std::align_val_t(alignof(SequenceCell))) SequenceCell[capacity_];
 
  payloads_ = static_cast<std::byte*>(
      ::operator new[](payload_bytes, std::align_val_t(PAYLOAD_ALLOC_ALIGN)));
 
  for (size_t index = 0; index < capacity_; ++index)
  {
    sequences_[index].value.store(static_cast<SequenceType>(index),
                                  std::memory_order_relaxed);
  }
}
 
MPMCQueueBase::~MPMCQueueBase()
{
  ::operator delete[](payloads_, std::align_val_t(PAYLOAD_ALLOC_ALIGN));
  delete[] sequences_;
}
 
ErrorCode MPMCQueueBase::PushBytes(const void* value)
{
  if (value == nullptr)
  {
    return ErrorCode::PTR_NULL;
  }
 
  SequenceType position = tail_.load(std::memory_order_relaxed);
 
  while (true)
  {
    SequenceCell& slot = sequences_[position % capacity_];
    const SequenceType sequence = slot.value.load(std::memory_order_acquire);
    const SequenceDiffType diff = static_cast<SequenceDiffType>(sequence - position);
 
    if (diff == 0)
    {
      if (tail_.compare_exchange_weak(position, position + 1, std::memory_order_relaxed,
                                      std::memory_order_relaxed))
      {
        LibXR::Memory::FastCopy(PayloadPtr(position % capacity_), value, element_size_);
        slot.value.store(position + 1, std::memory_order_release);
        return ErrorCode::OK;
      }
      continue;
    }
 
    if (diff < 0)
    {
      return ErrorCode::FULL;
    }
 
    position = tail_.load(std::memory_order_relaxed);
  }
}
 
ErrorCode MPMCQueueBase::PopBytes(void* value)
{
  SequenceType position = head_.load(std::memory_order_relaxed);
 
  while (true)
  {
    SequenceCell& slot = sequences_[position % capacity_];
    const SequenceType sequence = slot.value.load(std::memory_order_acquire);
    const SequenceType expected_ready = position + 1;
    const SequenceDiffType diff =
        static_cast<SequenceDiffType>(sequence - expected_ready);
 
    if (diff == 0)
    {
      if (head_.compare_exchange_weak(position, position + 1, std::memory_order_relaxed,
                                      std::memory_order_relaxed))
      {
        if (value != nullptr)
        {
          LibXR::Memory::FastCopy(value, PayloadPtr(position % capacity_), element_size_);
        }
        slot.value.store(position + static_cast<SequenceType>(capacity_),
                         std::memory_order_release);
        return ErrorCode::OK;
      }
      continue;
    }
 
    if (diff < 0)
    {
      return ErrorCode::EMPTY;
    }
 
    position = head_.load(std::memory_order_relaxed);
  }
}
 
size_t MPMCQueueBase::Size() const
{
  const SequenceType head_snapshot = head_.load(std::memory_order_acquire);
  const SequenceType tail_snapshot = tail_.load(std::memory_order_acquire);
  const SequenceType used = tail_snapshot - head_snapshot;
  return (used <= capacity_) ? used : capacity_;
}
 
void* MPMCQueueBase::PayloadPtr(size_t index)
{
  return payloads_ + index * payload_stride_;
}
 
const void* MPMCQueueBase::PayloadPtr(size_t index) const
{
  return payloads_ + index * payload_stride_;
}
 
size_t MPMCQueueBase::AlignUpChecked(size_t value, size_t align)
{
  REQUIRE(align > 0);
  REQUIRE(value <= std::numeric_limits<size_t>::max() - (align - 1));
  return ((value + align - 1) / align) * align;
}
 
size_t MPMCQueueBase::MultiplyChecked(size_t lhs, size_t rhs)
{
  if (lhs == 0 || rhs == 0)
  {
    return 0;
  }
 
  REQUIRE(lhs <= std::numeric_limits<size_t>::max() / rhs);
  return lhs * rhs;
}
 
}  // namespace LibXR