spsc_queue.hpp

#pragma once
 
#include <cstddef>
#include <type_traits>
 
#include "queue_typed_base.hpp"
#include "spsc_queue_base.hpp"
 
namespace LibXR
{
template <typename Data>
class SPSCQueue final : public QueueTypedBase<SPSCQueue<Data>, Data>, public SPSCQueueBase
{
 public:
  static_assert(alignof(Data) <= alignof(std::max_align_t),
                "SPSCQueue does not support over-aligned payload types");
 
  using ValueType = Data;  
  using QueueTypedBase<SPSCQueue<Data>, Data>::Pop;
  using QueueTypedBase<SPSCQueue<Data>, Data>::Push;
 
  explicit SPSCQueue(size_t length) : SPSCQueueBase(sizeof(Data), alignof(Data), length)
  {
  }
 
  ~SPSCQueue() = default;
 
  ErrorCode Peek(Data& item) { return SPSCQueueBase::PeekBytes(&item); }
 
  ErrorCode PushBatch(const Data* data, size_t size)
  {
    return SPSCQueueBase::PushBatchBytes(data, size);
  }
 
  template <typename Writer>
  ErrorCode PushWithWriter(Writer&& writer)
  {
    return PushWithWriter(1, std::forward<Writer>(writer));
  }
 
  template <typename Writer>
  ErrorCode PushWithWriter(size_t size, Writer&& writer)
  {
    static_assert(
        std::is_trivially_copyable_v<Data>,
        "batched SPSCQueue::PushWithWriter requires trivially copyable payloads");
    static_assert(
        std::is_trivially_destructible_v<Data>,
        "batched SPSCQueue::PushWithWriter requires trivially destructible payloads");
    static_assert(std::is_invocable_v<Writer&, Data*, size_t>,
                  "PushWithWriter writer must be callable as "
                  "ErrorCode(Data* buffer, size_t count)");
    using WriterRet = std::invoke_result_t<Writer&, Data*, size_t>;
    static_assert(std::is_convertible_v<WriterRet, ErrorCode>,
                  "PushWithWriter writer return type must be convertible to ErrorCode");
 
    Writer& writer_ref = writer;
    return SPSCQueueBase::PushBytesWithWriter(
        size, [&](void* buffer, size_t count) -> ErrorCode
        { return writer_ref(static_cast<Data*>(buffer), count); });
  }
 
  template <typename Writer>
  size_t ProduceWithWriter(size_t limit, Writer&& writer)
  {
    static_assert(std::is_trivially_copyable_v<Data>);
    static_assert(std::is_trivially_destructible_v<Data>);
    return SPSCQueueBase::ProduceWithWriter(
        limit,
        [&](void* first, size_t first_size, void* second, size_t second_size) -> size_t
        {
          return writer(static_cast<Data*>(first), first_size, static_cast<Data*>(second),
                        second_size);
        });
  }
 
  template <typename Reader>
  ErrorCode PopWithReader(Reader&& reader)
  {
    return PopWithReader(1, std::forward<Reader>(reader));
  }
 
  template <typename Reader>
  ErrorCode PopWithReader(size_t size, Reader&& reader)
  {
    static_assert(
        std::is_trivially_copyable_v<Data>,
        "batched SPSCQueue::PopWithReader requires trivially copyable payloads");
    static_assert(
        std::is_trivially_destructible_v<Data>,
        "batched SPSCQueue::PopWithReader requires trivially destructible payloads");
    static_assert(std::is_invocable_v<Reader&, const Data*, size_t>,
                  "PopWithReader reader must be callable as "
                  "ErrorCode(const Data* buffer, size_t count)");
    using ReaderRet = std::invoke_result_t<Reader&, const Data*, size_t>;
    static_assert(std::is_convertible_v<ReaderRet, ErrorCode>,
                  "PopWithReader reader return type must be convertible to ErrorCode");
 
    Reader& reader_ref = reader;
    return SPSCQueueBase::PopBytesWithReader(
        size, [&](const void* buffer, size_t count) -> ErrorCode
        { return reader_ref(static_cast<const Data*>(buffer), count); });
  }
 
  template <typename Reader>
  size_t ConsumeWithReader(size_t limit, Reader&& reader)
  {
    static_assert(std::is_trivially_copyable_v<Data>);
    static_assert(std::is_trivially_destructible_v<Data>);
    return SPSCQueueBase::ConsumeWithReader(
        limit,
        [&](const void* first, size_t first_size, const void* second,
            size_t second_size) -> size_t
        {
          return reader(static_cast<const Data*>(first), first_size,
                        static_cast<const Data*>(second), second_size);
        });
  }
 
  ErrorCode PopBatch(Data* data, size_t size)
  {
    return SPSCQueueBase::PopBatchBytes(data, size);
  }
 
  ErrorCode PeekBatch(Data* data, size_t size)
  {
    return SPSCQueueBase::PeekBatchBytes(data, size);
  }
 
  void Reset() { SPSCQueueBase::Reset(); }
};
}  // namespace LibXR