ring_buffer

Go to the documentation of this file.

// vim:set ft=cpp: -*- Mode: C++ -*-
/*
 * Copyright (C) 2026 Kernkonzept GmbH.
 * Author(s): Martin Decky <martin.decky@kernkonzept.com>
 *
 * License: see LICENSE.spdx (in this directory or the directories above)
 */

 
#pragma once
 
#include <cstddef>
#include <stdexcept>
#include <atomic>
#include <bit>
#include <mutex>
#include <functional>
 
namespace utrace {

#if defined(ARCH_arm64)
  static constexpr size_t stable_cache_alignment = 256U;
#elif defined(ARCH_ppc32)
  static constexpr size_t stable_cache_alignment = 128U;
#else
  static constexpr size_t stable_cache_alignment = 64U;
#endif
 
#ifdef __GCC_DESTRUCTIVE_SIZE
static_assert(stable_cache_alignment >= __GCC_DESTRUCTIVE_SIZE,
              "Stable cache alignment not sufficient");
#endif // __GCC_DESTRUCTIVE_SIZE

template<typename SEQUENCE_TYPE>
class Ring_buffer
{
public:
  using Sequence = SEQUENCE_TYPE;
  using Generation = std::atomic<Sequence>;

  static constexpr Sequence const nil = 0U;

  static void check_items(size_t const items)
  {
    if (items == 0)
      throw std::invalid_argument("Zero items not supported.");
 
    if (items != std::bit_ceil(items))
      throw std::invalid_argument("Number of items must be power of two.");
  }

  static void check_mask(size_t const mask)
  {
    auto items = mask + 1;
    if (items != std::bit_ceil(items))
      throw std::length_error("Invalid mask.");
  }
};

template<typename SEQUENCE_TYPE>
class Ring_status
{
public:
  using Sequence = SEQUENCE_TYPE;
  using Generation = typename Ring_buffer<Sequence>::Generation;
 
  template<typename S, typename T, auto G> friend class Ring_buffer_producer;
  template<typename S, typename T, auto G>
  friend class Ring_buffer_consumer_raw;

  unsigned version() const
  { return _version; }

  size_t alignment() const
  { return _alignment; }

  size_t mask() const
  { return _mask; }

  size_t items() const
  { return _mask + 1; }
 
private:
  unsigned _version;

  size_t _alignment;

  alignas(stable_cache_alignment) size_t _mask;

  alignas(stable_cache_alignment) Generation _tail;
};

template<typename ITEM_TYPE>
static constexpr size_t ring_slot_alignment = std::bit_ceil(sizeof(ITEM_TYPE));

template<typename ITEM_TYPE, auto GENERATION_PTR>
class alignas(ring_slot_alignment<ITEM_TYPE>) Ring_slot
{
public:
  using This = Ring_slot;
  using Item = ITEM_TYPE;
 
  template<typename S, typename T, auto G> friend class Ring_buffer_producer;
  template<typename S, typename T, auto G>
  friend class Ring_buffer_consumer_raw;

  static size_t size(size_t const items)
  { return items * sizeof(This); }
 
private:
  auto generation() const noexcept
  { return std::atomic_ref(_data.*GENERATION_PTR); }

  Item _data;
};

template<typename SEQUENCE_TYPE, typename ITEM_TYPE, auto GENERATION_PTR>
class Ring_buffer_producer : public Ring_buffer<SEQUENCE_TYPE>
{
public:
  using This = Ring_buffer_producer;
  using Super = Ring_buffer<SEQUENCE_TYPE>;
 
  using Item = ITEM_TYPE;
 
  using Status = Ring_status<SEQUENCE_TYPE>;
  using Slot = Ring_slot<Item, GENERATION_PTR>;

  Ring_buffer_producer(Status &status, Slot &slots, unsigned const version,
                       size_t const items) : _status(status), _slots(slots)
  {
    This::check_items(items);
 
    _status._version = version;
    _status._alignment = stable_cache_alignment;
    _status._mask = items - 1;
    _status._tail = Super::nil;
 
    for (size_t i = 0; i < items; ++i)
      _slots[i].generation() = Super::nil;
  }

  size_t items() const
  { return _status.items(); }

  void enqueue(Item const &item) const
  {
    auto next = ++_status._tail;
    auto index = next & _status._mask;
 
    auto &generation = _slots[index].generation();
    auto &slot = _slots[index]._data;
 
    generation.store(Super::nil, std::memory_order_release);
    slot = item;
    generation.store(next, std::memory_order_release);
  }
 
private:
  Ring_buffer_producer() = delete;
 
  Status &_status;
  Slot &_slots;
};

template<typename SEQUENCE_TYPE, typename ITEM_TYPE, auto GENERATION_PTR>
class Ring_buffer_consumer_raw : public Ring_buffer<SEQUENCE_TYPE>
{
public:
  using This = Ring_buffer_consumer_raw;
  using Super = Ring_buffer<SEQUENCE_TYPE>;
 
  using Sequence = SEQUENCE_TYPE;
  using Item = ITEM_TYPE;
 
  using Status = Ring_status<Sequence>;
  using Slot = Ring_slot<Item, GENERATION_PTR>;

  enum class Drop_policy
  {
    Conservative = 0,

    Minimal
  };

  enum class State
  {
    Idle = 0,  //< No item dequeued.
    Ready,     //< An item has been dequeued.
    Dropped    //< Some items have been dropped.
  };

  Ring_buffer_consumer_raw(Status const &status, Slot const *slots)
    : _status(status), _slots(slots)
  { This::check_mask(_status._mask); }

  size_t items() const
  { return _status.items(); }

  State peek(Item &item, Drop_policy policy, Sequence &current,
             Sequence *drops = nullptr) const
  {
    if (drops)
      *drops = 0;
 
    // First access to the ring buffer.
    if (current == Super::nil)
      current = 1;
 
    auto index = current & _status._mask;
 
    auto const &generation = _slots[index].generation();
    auto const &slot = _slots[index]._data;
 
    auto seen = generation.load();
    if (seen < current)
      {
        // Expected item not produced yet.
        return State::Idle;
      }
 
    if (seen > current)
      goto drop;
 
    item = slot;
    seen = generation.load();
 
    // Make sure the item has not been overwritten while we have been copying
    // the data.
    if (seen == current)
      {
        // Consume item.
        ++current;
        return State::Ready;
      }
 
  drop:
    // Seen > current -> drop detected.
    Sequence head = _status._tail;
 
    // Conservative drop policy continues at the tail.
    // Minimal drop policy continues at the head.
    if (policy == Drop_policy::Minimal)
      head -= _status._mask;
 
    // How many items have been dropped.
    if (drops)
      *drops = head - current;
 
    current = head;
    return State::Dropped;
  }
 
private:
  Ring_buffer_consumer_raw() = delete;
 
  Status const &_status;
  Slot const *_slots;
};

template<typename SEQUENCE_TYPE, typename ITEM_TYPE, auto GENERATION_PTR>
class Ring_buffer_consumer
  : public Ring_buffer_consumer_raw<SEQUENCE_TYPE, ITEM_TYPE, GENERATION_PTR>
{
public:
  using Sequence = SEQUENCE_TYPE;
  using Item = ITEM_TYPE;
 
  using Super = Ring_buffer_consumer_raw<Sequence, Item, GENERATION_PTR>;
 
  using Drop_policy = Super::Drop_policy;
  using State = Super::State;
  using Status = Ring_status<Sequence>;
  using Slot = Ring_slot<Item, GENERATION_PTR>;
 
  using Yield = std::function<bool (Sequence)>;

  Ring_buffer_consumer(Status const &status, Slot const *slots)
    : Super(status, slots)
  {}

  State peek(Item &item, Drop_policy policy, Sequence *drops = nullptr)
  {
    const std::lock_guard guard(_lock);
    return Super::peek(item, policy, _current, drops);
  }

  size_t dequeue(Item items[], size_t capacity, size_t burst,
                 Drop_policy policy, Yield const &yield,
                 Sequence *drops = nullptr)
  {
    if (drops)
      *drops = 0;
 
    size_t count = 0;
    size_t idle_cycles = 0;
 
    for (;;)
      {
        Sequence current_drops;
        auto status = peek(items[count], policy, &current_drops);
 
        if (status == State::Dropped)
          {
            // Some items have been dropped. We need to notify the caller
            // about it. Thus return with the items already dequeued.
 
            if (drops)
              *drops = current_drops;
 
            break;
          }
 
        if (status == State::Ready)
          {
            // Next item dequeued. Return if there is no space for more items
            // in the array.
 
            ++count;
            if (count == capacity)
              break;
 
            continue;
         }
 
       if (status == State::Idle)
         {
           // No next item has been produced yet. Return if we have already
           // dequeued enough items. Otherwise just spin, but allow the yield
           // function to implement passive waiting.
 
           if (count >= burst)
             break;
 
           ++idle_cycles;
           if (yield(idle_cycles))
             idle_cycles = 0;
         }
      }
 
    return count;
  }
 
private:
  Ring_buffer_consumer() = delete;
  State peek(Item &, Drop_policy, Sequence &, Sequence *) const = delete;

  std::mutex _lock;

  Sequence _current = 0;
};
 
} // namespace utrace