mpsc_shm.hpp

Go to the documentation of this file.

 
#pragma once
 
#include "spsc_shm.hpp"
#include <array>
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <string>
#include <vector>
 
namespace bb::ipc {
 
struct alignas(64) MpscDoorbell {
    // Producer-written (written by producers in publish())
    alignas(64) std::atomic<uint32_t> seq;
    std::array<uint8_t, 60> _pad0;
 
    // Consumer-written (written by consumer in wait_for_data())
    alignas(64) std::atomic<bool> consumer_blocked; // Set RIGHT BEFORE futex_wait, cleared RIGHT AFTER
    std::array<uint8_t, 63> _pad1;
};
 
class MpscConsumer {
  public:
    static MpscConsumer create(const std::string& name, size_t num_producers, size_t ring_capacity);
 
    static bool unlink(const std::string& name, size_t num_producers);
 
    // Move-only
    MpscConsumer(MpscConsumer&& other) noexcept;
    MpscConsumer& operator=(MpscConsumer&& other) noexcept;
    MpscConsumer(const MpscConsumer&) = delete;
    MpscConsumer& operator=(const MpscConsumer&) = delete;
 
    ~MpscConsumer();
 
    int wait_for_data(uint32_t timeout_ns);
 
    void* peek(size_t ring_idx, size_t want, uint32_t timeout_ns);
 
    void release(size_t ring_idx, size_t n);
 
    void wakeup_all();
 
  private:
    MpscConsumer(std::vector<SpscShm>&& rings, int doorbell_fd, size_t doorbell_len, MpscDoorbell* doorbell);
 
    std::vector<SpscShm> rings_;
    int doorbell_fd_ = -1;
    size_t doorbell_len_ = 0;
    MpscDoorbell* doorbell_ = nullptr;
    size_t last_served_ = 0;         // Round-robin fairness
    bool previous_had_data_ = false; // Adaptive spinning: only spin if previous call found data
};
 
class MpscProducer {
  public:
    static MpscProducer connect(const std::string& name, size_t producer_id);
 
    // Move-only
    MpscProducer(MpscProducer&& other) noexcept;
    MpscProducer& operator=(MpscProducer&& other) noexcept;
    MpscProducer(const MpscProducer&) = delete;
    MpscProducer& operator=(const MpscProducer&) = delete;
 
    ~MpscProducer();
 
    void* claim(size_t want, uint32_t timeout_ns);
 
    void publish(size_t n);
 
  private:
    MpscProducer(SpscShm&& ring, int doorbell_fd, size_t doorbell_len, MpscDoorbell* doorbell, size_t producer_id);
 
    SpscShm ring_;
    int doorbell_fd_ = -1;
    size_t doorbell_len_ = 0;
    MpscDoorbell* doorbell_ = nullptr;
    size_t producer_id_ = 0;
};
 
} // namespace bb::ipc