chronon::InPort

More...

#include <InPort.hpp>

Inherits from chronon::sender::PortBase, chronon::sender::IArbitratablePort

Public Types

	Name
using detail::PortEnvelope< T >	StoredMessage

Public Functions

	Name
bool	usesLockFreeQueue() const
void	useSingleThreadQueue()
void	useMultiProducerQueue(size_t min_per_thread_usable_capacity =0)
void	useLockFreeQueue(size_t min_usable_capacity =0)
std::optional< T >	tryReceive(uint64_t current_cycle)
size_t	storageCapacity() const
virtual uint64_t	stagingOverflowEvents() const override
void	setCapacity(size_t capacity)
virtual void	setArbitrationProgressPointers(std::vector< const std::atomic< uint64_t > * > ptrs) override
virtual void	setArbitrationConnProgress(const std::unordered_map< Unit *, const std::atomic< uint64_t > * > & src_progress) override
void	resetSelectiveCancellation()
size_t	registerProducerThread(size_t thread_id)
void	registerMPSCConnection(ConnectionBase * conn)
const std::vector< T > &	receiveAllBuffered(uint64_t current_cycle)
const std::vector< T > &	receiveAllBuffered()
std::vector< T >	receiveAll(uint64_t current_cycle)
std::vector< T >	receiveAll() Receive all messages ready at the owning Unit's current local cycle.
size_t	queuedMessageCount() const
bool	pushToThreadQueueCancelable(size_t queue_id, T data, uint64_t arrive_cycle, const std::atomic< uint64_t > * cancel_epoch, uint64_t epoch_snapshot, uint64_t enqueue_cycle =0, uint32_t sender_id =0) Enqueue a cancelable message to a specific thread queue (MPSC mode).
bool	pushToThreadQueue(size_t queue_id, T data, uint64_t arrive_cycle, uint64_t enqueue_cycle =0, uint32_t sender_id =0)
PortPolicy	policy() const
virtual bool	mpscConnProgressFullyResolved() const override
virtual std::optional< uint64_t >	minArrivalCycle() const override Earliest arrival cycle of pending or staged messages, used for lookahead.
bool	isMultiProducerMode() const True if the port is in multi-producer mode.
bool	hasMessages() const True if messages are ready at the owning Unit's current local cycle.
virtual bool	hasMPSCConnections() const override True if this port owns any MPSC staging queues that need consumer-side drains.
bool	hasData(uint64_t current_cycle) const
size_t	getQueueIdForThread(size_t thread_id) const Get the queue ID for a given producer key (multi-producer mode only).
void	flush() Drop all queued messages (including future arrivals).
bool	enqueueCancelable(T data, uint64_t arrive_cycle, const std::atomic< uint64_t > * cancel_epoch, uint64_t epoch_snapshot, uint32_t sender_id =0)
bool	enqueueCancelable(T data, uint64_t arrive_cycle, const std::atomic< uint64_t > * cancel_epoch, uint64_t epoch_snapshot, uint64_t enqueue_cycle, uint32_t sender_id =0)
size_t	configuredCapacity() const
void	configureForSourceThreadCount(size_t source_thread_count)
size_t	capacity() const
template <auto KeyFn,typename K > void	cancelYoungerThan(K watermark) Selectively cancel in-flight messages where KeyFn(data) > watermark.
template <auto KeyFn,typename MinK ,typename MaxK > void	cancelOutsideInclusive(MinK min_keep, MaxK max_keep) Keep only keys in [min_keep, max_keep] for current in-flight generation.
template <auto KeyFn,typename K > void	cancelOlderThan(K watermark)
bool	canPushToThreadQueue(size_t queue_id) const
bool	canAcceptOnThreadQueue(size_t queue_id, size_t pending =0) const
bool	canAccept(size_t pending =0) const True if this producer can push again in the current simulated cycle.
size_t	available() const
virtual void	arbitrateMPSCConsumerDriven() override
virtual void	arbitrateMPSC() override
virtual void *	arbitratablePortKey() override
size_t	admissionOccupancy(uint64_t send_cycle) const
std::optional< uint64_t >	admissionMinArrivalCycle(uint64_t send_cycle) const
size_t	admissionCapacity() const
	InPort(Unit * owner, std::string name, size_t capacity =UNLIMITED_CAPACITY, PortPolicy policy =PortPolicy::LegacyFastPath)
	InPort(Unit * owner, std::string name, PortPolicy policy) Convenience constructor: specify policy without setting capacity.

Public Attributes

	Name
size_t	UNLIMITED_CAPACITY

Additional inherited members

Public Functions inherited from chronon::sender::PortBase

	Name
virtual	~PortBase() =default
Unit *	owner() const
const std::string &	name() const

Protected Functions inherited from chronon::sender::PortBase

	Name
	PortBase(Unit * owner, std::string name)

Protected Attributes inherited from chronon::sender::PortBase

	Name
Unit *	owner_
std::string	name_

Public Functions inherited from chronon::sender::IArbitratablePort

	Name
virtual	~IArbitratablePort() =default

Detailed Description

template <typename T >
class chronon::InPort;

InPort - Receives data from connected OutPorts.

Features:

• Timestamped message queue for deterministic delivery
• Synchronous tryReceive for tick-based units

Usage: InPort in{this, "in"};

// In tick() method if (auto value = in.tryReceive(localCycle())) { process(*value); }

Public Types Documentation

using StoredMessage

using chronon::sender::InPort< T >::StoredMessage =  detail::PortEnvelope<T>;

Public Functions Documentation

function usesLockFreeQueue

inline bool usesLockFreeQueue() const

True when this port drains a bounded lock-free SPSC ring (cross-thread, single producer). The ring is finite even for an unlimited-capacity port, so the epoch-free gate must bound producer run-ahead against capacity().

function useSingleThreadQueue

inline void useSingleThreadQueue()

Switch to single-thread queue (no mutex overhead).

Call this during initialization when both producer and consumer are determined to be on the same thread.

WARNING: Must be called before simulation starts (queue must be empty).

function useMultiProducerQueue

inline void useMultiProducerQueue(
    size_t min_per_thread_usable_capacity =0
)

Switch to multi-producer queue mode.

Call this during initialization for cross-thread connections where multiple source threads write to this port. Creates independent SPSC queues polled by the consumer. TickSimulation uses stable connection ids as the producer keys.

function useLockFreeQueue

inline void useLockFreeQueue(
    size_t min_usable_capacity =0
)

Switch to lock-free SPSC queue.

Call this during initialization for cross-thread connections where there is only ONE source thread writing to this port. Uses atomic operations instead of mutex.

function tryReceive

inline std::optional< T > tryReceive(
    uint64_t current_cycle
)

Parameters:

• current_cycle The current simulation cycle

Return: The message if available, std::nullopt otherwise

Try to receive a message synchronously.

function storageCapacity

inline size_t storageCapacity() const

function stagingOverflowEvents

inline virtual uint64_t stagingOverflowEvents() const override

Reimplements: chronon::sender::IArbitratablePort::stagingOverflowEvents

Number of staged pushes dropped because the physical MPSC ring was full. Nonzero indicates the lookahead window outran the ring capacity — a correctness failure. Surfaced for the epoch-free A/B watchdog. Default 0 for ports with no multi-producer staging.

function setCapacity

inline void setCapacity(
    size_t capacity
)

function setArbitrationProgressPointers

inline virtual void setArbitrationProgressPointers(
    std::vector< const std::atomic< uint64_t > * > ptrs
) override

Reimplements: chronon::sender::IArbitratablePort::setArbitrationProgressPointers

Install the set of completed_cycle atomics for the predecessor threads feeding this InPort's MPSC connections. Called at TickSimulation::initialize() once the thread_progress_array_ has been allocated. An empty set (e.g. under Sequential or Barrier execution) makes arbitrateMPSCConsumerDriven() degrade to the legacy unbounded arbiter — safe because those modes only call it when producers are known to have finished their cycle.

function setArbitrationConnProgress

inline virtual void setArbitrationConnProgress(
    const std::unordered_map< Unit *, const std::atomic< uint64_t > * > & src_progress
) override

Reimplements: chronon::sender::IArbitratablePort::setArbitrationConnProgress

Resolve one producer completed_cycle atomic per MPSC connection (by the connection's source unit), aligned with mpsc_connections_ (conn_id order). Enables the per-connection drain in arbitrateMPSCConsumerDriven().

function resetSelectiveCancellation

inline void resetSelectiveCancellation()

Clear receiver-side selective cancellation bounds and extractor.

StageSelective: this is a NO-OP. Predicates are scoped per-flush and retired automatically pop-driven by shouldCancel(). Clearing live predicates here would re-open the #7 overlapping-flush zombie bug: a second flush's reset would erase the first flush's strict max_keep, allowing zombies enqueued between the two flushes to bypass the stricter predicate. The caller is expected to continue invoking resetSelectiveCancellation() + cancelYoungerThan() in that order (for backward compat with the LegacyFastPath path) — we simply ignore the reset and let install() accumulate predicates into the slots array.

function registerProducerThread

inline size_t registerProducerThread(
    size_t thread_id
)

Parameters:

• thread_id Stable producer key; TickSimulation passes conn_id + 1.

Return: Queue ID for this producer key (used in pushToThreadQueue)

Register a stable producer key and get its queue ID.

Only valid in multi-producer mode.

function registerMPSCConnection

inline void registerMPSCConnection(
    ConnectionBase * conn
)

Register an MPSC-mode Connection with this InPort so that the cycle-boundary arbiter can drain its staging deque. Connections are kept sorted by conn_id to give the arbiter a topology-stable fixed- priority order independent of num_workers or partition layout.

function receiveAllBuffered

inline const std::vector< T > & receiveAllBuffered(
    uint64_t current_cycle
)

Allocation-free variant of receiveAll() with identical ordering and cancellation semantics. Returns a reference into a reused buffer, valid until the next receive on this port. Single-consumer (only the owning unit drains its port), so the buffers need no synchronization.

function receiveAllBuffered

inline const std::vector< T > & receiveAllBuffered()

function receiveAll

inline std::vector< T > receiveAll(
    uint64_t current_cycle
)

Parameters:

• current_cycle The current simulation cycle

Return: Vector of all ready messages

Receive all available messages.

function receiveAll

inline std::vector< T > receiveAll()

Receive all messages ready at the owning Unit's current local cycle.

function queuedMessageCount

inline size_t queuedMessageCount() const

function pushToThreadQueueCancelable

inline bool pushToThreadQueueCancelable(
    size_t queue_id,
    T data,
    uint64_t arrive_cycle,
    const std::atomic< uint64_t > * cancel_epoch,
    uint64_t epoch_snapshot,
    uint64_t enqueue_cycle =0,
    uint32_t sender_id =0
)

Enqueue a cancelable message to a specific thread queue (MPSC mode).

function pushToThreadQueue

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

Parameters:

• queue_id The queue ID (from registerProducerThread)
• data The message data
• arrive_cycle When the message should be delivered

Return: true if enqueued, false if queue full

Push to a specific thread's queue in multi-producer mode.

function policy

inline PortPolicy policy() const

Return: the cancellation policy for this InPort.

function mpscConnProgressFullyResolved

inline virtual bool mpscConnProgressFullyResolved() const override

Reimplements: chronon::sender::IArbitratablePort::mpscConnProgressFullyResolved

True iff every MPSC connection on this port has a resolved per-connection producer progress atomic, i.e. the heterogeneous-delay-correct consumer-driven drain in arbitrateMPSCConsumerDriven() fully covers this port. When false, at least one connection relies on the central per-epoch arbitrateMPSC() flush to deliver its tail (an unresolved producer is skipped by the consumer-driven path), so the epoch-free scheduler — which has no per-epoch flush — must NOT be used. Conservative default: false, so unknown port types veto epoch-free execution. InPort overrides.

function minArrivalCycle

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

Earliest arrival cycle of pending or staged messages, used for lookahead.

Reimplements: chronon::sender::PortBase::minArrivalCycle

function isMultiProducerMode

inline bool isMultiProducerMode() const

True if the port is in multi-producer mode.

function hasMessages

inline bool hasMessages() const

True if messages are ready at the owning Unit's current local cycle.

function hasMPSCConnections

inline virtual bool hasMPSCConnections() const override

True if this port owns any MPSC staging queues that need consumer-side drains.

Reimplements: chronon::sender::PortBase::hasMPSCConnections

function hasData

inline bool hasData(
    uint64_t current_cycle
) const

function getQueueIdForThread

inline size_t getQueueIdForThread(
    size_t thread_id
) const

Get the queue ID for a given producer key (multi-producer mode only).

function flush

inline void flush()

Drop all queued messages (including future arrivals).

function enqueueCancelable

inline bool enqueueCancelable(
    T data,
    uint64_t arrive_cycle,
    const std::atomic< uint64_t > * cancel_epoch,
    uint64_t epoch_snapshot,
    uint32_t sender_id =0
)

Enqueue a cancelable message for delivery.

Used by Connection to support OutPort::cancelInFlight().

function enqueueCancelable

inline bool enqueueCancelable(
    T data,
    uint64_t arrive_cycle,
    const std::atomic< uint64_t > * cancel_epoch,
    uint64_t epoch_snapshot,
    uint64_t enqueue_cycle,
    uint32_t sender_id =0
)

Cancelable enqueue with explicit enqueue_cycle stamp.

Connection::transfer() uses this overload (computed as send_cycle, which is the producer's localCycle) so that StageSelective predicates can decide whether the message predates the most recent flush.

function configuredCapacity

inline size_t configuredCapacity() const

function configureForSourceThreadCount

inline void configureForSourceThreadCount(
    size_t source_thread_count
)

Select a queue implementation from the number of source threads.

This is a semantic wrapper for manual setup. TickSimulation performs this optimization automatically during initialization.

function capacity

inline size_t capacity() const

function cancelYoungerThan

template <auto KeyFn,
typename K >
inline void cancelYoungerThan(
    K watermark
)

Selectively cancel in-flight messages where KeyFn(data) > watermark.

function cancelOutsideInclusive

template <auto KeyFn,
typename MinK ,
typename MaxK >
inline void cancelOutsideInclusive(
    MinK min_keep,
    MaxK max_keep
)

Keep only keys in [min_keep, max_keep] for current in-flight generation.

function cancelOlderThan

template <auto KeyFn,
typename K >
inline void cancelOlderThan(
    K watermark
)

Selectively cancel in-flight messages where KeyFn(data) < watermark.

Receiver-side selective cancellation defaults to in-flight scope: the first cancellation call captures the current enqueue generation so future messages are unaffected.

function canPushToThreadQueue

inline bool canPushToThreadQueue(
    size_t queue_id
) const

function canAcceptOnThreadQueue

inline bool canAcceptOnThreadQueue(
    size_t queue_id,
    size_t pending =0
) const

Check if a specific producer queue can accept data (MPSC mode).

Falls back to port-level canAccept in non-MPSC mode.

function canAccept

inline bool canAccept(
    size_t pending =0
) const

True if this producer can push again in the current simulated cycle.

function available

inline size_t available() const

function arbitrateMPSCConsumerDriven

inline virtual void arbitrateMPSCConsumerDriven() override

Reimplements: chronon::sender::PortBase::arbitrateMPSCConsumerDriven

Consumer-tick-driven arbitration (Option 1, see docs/mpsc-atomic-publish.md). Called by TickableUnit::executeTick on the consumer thread, immediately before the user's tick() body. Computes the safe drain bound S = min over predecessor threads of completed_cycle (acquire) and drains staging entries with enqueue_cycle <= S in conn_id order.

A no-op if the port has no MPSC connections, if the producer- thread set was never resolved (progress-based sync off), or if S has not advanced since the last arbitration.

function arbitrateMPSC

inline virtual void arbitrateMPSC() override

Reimplements: chronon::sender::IArbitratablePort::arbitrateMPSC

Per-cycle MPSC admission arbitration.

Called by the scheduler at each cycle boundary on InPorts that have at least one MPSC connection registered. Iterates the connections in conn_id-ascending order (stable across num_workers since conn_id is assigned at topology construction) and lets each drain its full staging deque into the shared queue.

The user_cap on the InPort governs per-producer staging depth (a Connection's staging deque is bounded at user_cap, giving the producer back-pressure when full); the arbiter itself does NOT cap aggregate per-cycle admission. A strict aggregate-cap budget here would starve lower-conn_id connections under heavy multi-producer fan-in: with N producers each pushing at the cap and a budget of cap/cycle, only the highest-priority producer makes progress. Bounding admission via per-producer staging plus consumer drain preserves both determinism (conn_id ordering of admission) and fairness (every producer that has unblocked staging makes progress every cycle).

function arbitratablePortKey

inline virtual void * arbitratablePortKey() override

Reimplements: chronon::sender::IArbitratablePort::arbitratablePortKey

Opaque identity for this arbitrable port. Returned by InPort as its this pointer (same value as destPortPtr() on connections). Used by TickSimulation to join MPSC InPorts against per-port producer- thread tables at init time.

function admissionOccupancy

inline size_t admissionOccupancy(
    uint64_t send_cycle
) const

function admissionMinArrivalCycle

inline std::optional< uint64_t > admissionMinArrivalCycle(
    uint64_t send_cycle
) const

function admissionCapacity

inline size_t admissionCapacity() const

Effective per-cycle admission bound visible to producer-side logic.

Connection::transfer consults this to enforce the same rate limit as canAccept() on the push path.

function InPort

inline InPort(
    Unit * owner,
    std::string name,
    size_t capacity =UNLIMITED_CAPACITY,
    PortPolicy policy =PortPolicy::LegacyFastPath
)

Parameters:

• owner The unit that owns this port
• name The port name (for debugging)
• capacity Maximum queue capacity (default unlimited)

Create an input port.

Automatically registers with PortDirectory when owner's TreeNode is set.

function InPort

inline InPort(
    Unit * owner,
    std::string name,
    PortPolicy policy
)

Convenience constructor: specify policy without setting capacity.

Public Attributes Documentation

variable UNLIMITED_CAPACITY

static size_t UNLIMITED_CAPACITY = std::numeric_limits<size_t>::max();

---

Updated on 2026-07-06 at 09:03:34 +0000