Skip to main content

MultiProducerQueueAdapter

More...

#include <MultiProducerQueueAdapter.hpp>

Inherits from IMessageQueue< T >

Public Functions​

Name
boolusesActiveFrontier() const
std::optional< T >tryPop(uint64_t current_cycle) override
uint64_ttransportOverflowEvents() const
boolstorageFullForThread(size_t queue_id) const
size_tstorageCapacity() const override
size_tsize() const override
voidsetCapacity(size_t capacity) override
boolpushFromThread(size_t queue_id, T data, uint64_t arrive_cycle, uint32_t sender_id =0)
boolpush(T data, uint64_t arrive_cycle) override
voidpopAllInto(std::vector< T > & out, uint64_t current_cycle) override
std::vector< T >popAll(uint64_t current_cycle) override
std::optional< uint64_t >minArrivalCycle() const override
boolhasReady(uint64_t current_cycle) const override
size_tgetQueueIdForThread(size_t thread_id) const
boolfullForThread(size_t queue_id) const
boolfull() const override
voidensurePerThreadUsableCapacity(size_t min_usable_capacity)
boolempty() const override
template <typename Visitor >
bool		consumeReady(uint64_t current_cycle, Visitor && visitor)
voidclear() override
size_tcapacity() const override
size_tavailable() const override
size_tadmissionOccupancyForThread(size_t queue_id, uint64_t send_cycle) const
std::optional< uint64_t >admissionMinArrivalCycleForThread(size_t queue_id, uint64_t send_cycle) const
size_taddProducerThread(size_t thread_id)
size_taddProducerThread(size_t thread_id, bool track_admission)
MultiProducerQueueAdapter(size_t capacity =std::numeric_limits< size_t >::max(), size_t min_per_thread_usable_capacity =0)

Public Attributes​

Name
size_tkLanesPerSignalWord
size_tkFrontierLaneThreshold

Detailed Description​

template <typename T >
class MultiProducerQueueAdapter;

MultiProducerQueueAdapter - deterministic MPSC built from independent SPSC lanes.

A Chronon Connection has exactly one producer and an InPort has exactly one consumer. Keeping one SPSC lane per Connection avoids the contended tail, slot-sequence CAS loops, and reclamation machinery of a general MPMC queue. The consumer merges lane heads by (arrive_cycle, sender_id, lane_id), which supplies the total order required for cycle-count reproducibility.

Small fan-in uses a branch-friendly linear scan. At larger fan-in, producers publish lane activity into cache-line-separated shards, and the consumer maintains a private min-heap containing at most one head per active lane. The activity bits are notifications, not ownership flags: exchange(0) coalesces repeated pushes, and a lane is reinserted after every pop while it remains non-empty. This avoids the producer-set versus consumer-clear lost-wakeup race of a persistent active-bit protocol.

Public Functions Documentation​

function usesActiveFrontier​

inline bool usesActiveFrontier() const

function tryPop​

inline std::optional< T > tryPop(
    uint64_t current_cycle
) override

function transportOverflowEvents​

inline uint64_t transportOverflowEvents() const

function storageFullForThread​

inline bool storageFullForThread(
    size_t queue_id
) const

function storageCapacity​

inline size_t storageCapacity() const override

function size​

inline size_t size() const override

function setCapacity​

inline void setCapacity(
    size_t capacity
) override

function pushFromThread​

inline bool pushFromThread(
    size_t queue_id,
    T data,
    uint64_t arrive_cycle,
    uint32_t sender_id =0
)

function push​

inline bool push(
    T data,
    uint64_t arrive_cycle
) override

function popAllInto​

inline void popAllInto(
    std::vector< T > & out,
    uint64_t current_cycle
) override

function popAll​

inline std::vector< T > popAll(
    uint64_t current_cycle
) override

function minArrivalCycle​

inline std::optional< uint64_t > minArrivalCycle() const override

function hasReady​

inline bool hasReady(
    uint64_t current_cycle
) const override

function getQueueIdForThread​

inline size_t getQueueIdForThread(
    size_t thread_id
) const

function fullForThread​

inline bool fullForThread(
    size_t queue_id
) const

function full​

inline bool full() const override

function ensurePerThreadUsableCapacity​

inline void ensurePerThreadUsableCapacity(
    size_t min_usable_capacity
)

function empty​

inline bool empty() const override

function consumeReady​

template <typename Visitor >
inline bool consumeReady(
    uint64_t current_cycle,
    Visitor && visitor
)

Consumer-only direct path. The visitor sees the selected lane head while its slot is still consumer-owned, then the lane publishes its new head. This lets InPort apply receiver filtering without type erasure or moving a rejected envelope.

function clear​

inline void clear() override

function capacity​

inline size_t capacity() const override

function available​

inline size_t available() const override

function admissionOccupancyForThread​

inline size_t admissionOccupancyForThread(
    size_t queue_id,
    uint64_t send_cycle
) const

function admissionMinArrivalCycleForThread​

inline std::optional< uint64_t > admissionMinArrivalCycleForThread(
    size_t queue_id,
    uint64_t send_cycle
) const

function addProducerThread​

inline size_t addProducerThread(
    size_t thread_id
)

Register one stable producer key and return its lane id.

function addProducerThread​

inline size_t addProducerThread(
    size_t thread_id,
    bool track_admission
)

function MultiProducerQueueAdapter​

inline explicit MultiProducerQueueAdapter(
    size_t capacity =std::numeric_limits< size_t >::max(),
    size_t min_per_thread_usable_capacity =0
)

Public Attributes Documentation​

variable kLanesPerSignalWord​

static size_t kLanesPerSignalWord = 64;

variable kFrontierLaneThreshold​

static size_t kFrontierLaneThreshold = 32;

Updated on 2026-07-17 at 17:43:31 +0000