mpmc_queue_base.hpp

#pragma once
 
#include <algorithm>
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <limits>
#include <new>
#include <type_traits>
 
#include "libxr_def.hpp"
 
namespace LibXR
{
class MPMCQueueBase
{
 public:
  using SequenceType = size_t;  
  using SequenceDiffType =
      std::make_signed_t<SequenceType>;  
 
  MPMCQueueBase(size_t element_size, size_t capacity);
  ~MPMCQueueBase();
 
  ErrorCode PushBytes(const void* value);
  ErrorCode PopBytes(void* value = nullptr);
 
  [[nodiscard]] size_t MaxSize() const { return capacity_; }
  [[nodiscard]] size_t Size() const;
  [[nodiscard]] size_t EmptySize() const { return capacity_ - Size(); }
  [[nodiscard]] size_t ElementSize() const { return element_size_; }
 
 private:
  struct alignas(LibXR::CONCURRENCY_ALIGNMENT) SequenceCell
  {
    std::atomic<SequenceType> value;  
  };
 
  [[nodiscard]] void* PayloadPtr(size_t index);
  [[nodiscard]] const void* PayloadPtr(size_t index) const;
  [[nodiscard]] static size_t AlignUpChecked(size_t value, size_t align);
  [[nodiscard]] static size_t MultiplyChecked(size_t lhs, size_t rhs);
  static constexpr size_t PAYLOAD_ALLOC_ALIGN =
      std::max(alignof(size_t), alignof(std::max_align_t));
 
  MPMCQueueBase(const MPMCQueueBase&);
  MPMCQueueBase& operator=(const MPMCQueueBase&);
  MPMCQueueBase(MPMCQueueBase&&);
  MPMCQueueBase& operator=(MPMCQueueBase&&);
 
  const size_t element_size_;    
  const size_t capacity_;        
  const size_t payload_stride_;  
  SequenceCell* sequences_;      
  std::byte* payloads_;          
 
  alignas(LibXR::CONCURRENCY_ALIGNMENT) std::atomic<SequenceType>
      head_;  
  alignas(LibXR::CONCURRENCY_ALIGNMENT) std::atomic<SequenceType>
      tail_;  
};
}  // namespace LibXR