spsc_shm.hpp

Go to the documentation of this file.

 
#pragma once
 
#include <array>
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <string>
 
namespace bb::ipc {
 
constexpr size_t SPSC_CACHELINE = 64;
 
struct alignas(SPSC_CACHELINE) SpscCtrl {
    // Producer-owned (written by producer, read by consumer)
    alignas(SPSC_CACHELINE) std::atomic<uint64_t> head; // bytes written
    uint64_t wrap_head; // Head value when last message wrapped (UINT64_MAX = no wrap), synchronized by head
    std::atomic<bool> producer_blocked; // Written by producer in wait_for_space()
    std::array<char, SPSC_CACHELINE - sizeof(head) - sizeof(wrap_head) - sizeof(producer_blocked)> _pad0;
 
    // Consumer-owned (written by consumer, read by producer)
    alignas(SPSC_CACHELINE) std::atomic<uint64_t> tail; // bytes consumed
    std::atomic<bool> consumer_blocked;                 // Written by consumer in wait_for_data()
    std::array<char, SPSC_CACHELINE - sizeof(tail) - sizeof(consumer_blocked)> _pad1;
 
    // Immutable capacity information
    alignas(SPSC_CACHELINE) uint64_t capacity; // power of two
    alignas(SPSC_CACHELINE) uint64_t mask;     // capacity - 1
 
    // uint8_t buffer[capacity] follows in memory...
};
 
static_assert(alignof(SpscCtrl) == SPSC_CACHELINE, "SpscCtrl alignment");
static_assert(sizeof(SpscCtrl) % SPSC_CACHELINE == 0, "SpscCtrl size multiple of cache line");
 
class SpscShm {
  public:
    static SpscShm create(const std::string& name, size_t min_capacity);
 
    static SpscShm connect(const std::string& name);
 
    static bool unlink(const std::string& name);
 
    // Move-only (no copy)
    SpscShm(SpscShm&& other) noexcept;
    SpscShm& operator=(SpscShm&& other) noexcept;
    SpscShm(const SpscShm&) = delete;
    SpscShm& operator=(const SpscShm&) = delete;
 
    ~SpscShm();
 
    // Introspection
    uint64_t available() const; // bytes ready to read
 
    uint64_t capacity() const { return ctrl_->capacity; }
 
    void* claim(size_t want, uint32_t timeout_ns);
 
    void publish(size_t n);
 
    void* peek(size_t want, uint32_t timeout_ns);
 
    void release(size_t n);
 
    void wakeup_all();
 
    bool wait_for_data(size_t need, uint32_t spin_ns);
    bool wait_for_space(size_t need, uint32_t spin_ns);
 
    void debug_dump(const char* prefix) const;
 
  private:
    // Private constructor for create/connect factories
    SpscShm(int fd, size_t map_len, SpscCtrl* ctrl, uint8_t* buf);
 
    int fd_ = -1;
    size_t map_len_ = 0;
    SpscCtrl* ctrl_ = nullptr;
    uint8_t* buf_ = nullptr;
    bool previous_had_data_ = false;  // Adaptive spinning: consumer only spins if previous call found data
    bool previous_had_space_ = false; // Adaptive spinning: producer only spins if previous call found space
};
 
} // namespace bb::ipc