chronon::InPort

More...

#include <InPort.hpp>

Inherits from chronon::sender::PortBase, chronon::sender::IMultiProducerPort

Public Types

	Name
using detail::PortEnvelope< T >	StoredMessage

Public Functions

	Name
bool	usesTransparentBroadcast() const
bool	usesLockFreeQueue() const
bool	usesDirectSPSC() const
void	useSingleThreadQueue(bool cycle_strict_admission =false)
void	useMultiProducerQueue(size_t min_per_thread_usable_capacity =0)
void	useLockFreeQueue(size_t min_usable_capacity =0)
template <typename Filter > std::optional< T >	tryReceiveFiltered(uint64_t current_cycle, Filter && filter)
template <typename Filter > std::optional< T >	tryReceiveFiltered(Filter && filter)
std::optional< T >	tryReceive(uint64_t current_cycle)
size_t	transportPendingMessageCount() const
virtual uint64_t	transportOverflowEvents() const override
bool	transparentBroadcastEligible() const
size_t	threadQueueAdmissionOccupancy(size_t queue_id, uint64_t send_cycle) const
std::optional< uint64_t >	threadQueueAdmissionMinArrivalCycle(size_t queue_id, uint64_t send_cycle) const
size_t	storageCapacity() const
size_t	sharedFifoHighWatermark() const
virtual void	setProducerProgress(const std::unordered_map< Unit *, const std::atomic< uint64_t > * > & src_progress) override
void	setCapacity(size_t capacity)
void	resetSelectiveCancellation()
detail::ProducerDestinationTransactionState *	registerTransactionProducer(Unit * producer, ConnectionBase * connection)
void	registerSharedBroadcastConnection(Connection< T > * conn)
size_t	registerProducerThread(size_t thread_id)
size_t	registerProducerThread(size_t thread_id, bool track_admission)
void	registerMPSCConnection(ConnectionBase * conn)
void	registerIncomingDelay(uint32_t delay)
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)
virtual bool	producerProgressFullyResolved() const override
virtual void	prepareConsumerCycle(uint64_t current_cycle) override
PortPolicy	policy() const
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.
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).
template <auto KeyFn> void	flush(FlushRange keep_range)
bool	finalizeTransparentBroadcastReplay(size_t producer_count)
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) Compatibility spelling: cancel keys strictly above watermark.
template <auto KeyFn,typename MinK ,typename MaxK > void	cancelOutsideInclusive(MinK min_keep, MaxK max_keep) Compatibility spelling: keep only [min_keep, max_keep].
template <auto KeyFn,typename K > void	cancelOlderThan(K watermark) Compatibility spelling: cancel keys strictly below 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
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

Friends

	Name
struct	InPortSelectiveFlushTestAccess
class	Connection< T >

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::IMultiProducerPort

	Name
virtual	~IMultiProducerPort() =default
uint64_t	stagingOverflowEvents() const Backward-compatible name retained for diagnostics clients.
void	setArbitrationConnProgress(const std::unordered_map< Unit *, const std::atomic< uint64_t > * > & src_progress)
bool	mpscConnProgressFullyResolved() const

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 usesTransparentBroadcast

inline bool usesTransparentBroadcast() const

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 usesDirectSPSC

inline bool usesDirectSPSC() const

function useSingleThreadQueue

inline void useSingleThreadQueue(
    bool cycle_strict_admission =false
)

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 ingress lanes polled by the consumer. TickSimulation uses stable connection ids as the producer keys. Bounded ports also create one receiver-owned aggregate FIFO; unbounded ports keep the direct in-place lane-consume fast path.

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 tryReceiveFiltered

template <typename Filter >
inline std::optional< T > tryReceiveFiltered(
    uint64_t current_cycle,
    Filter && filter
)

Receive the first ready message accepted by filter.

Ready messages rejected by the receiver are consumed and permanently canceled. The predicate must be noexcept and is invoked directly (no std::function, allocation, or producer-side shared-state read), so capturing lambdas are supported without indirection.

function tryReceiveFiltered

template <typename Filter >
inline std::optional< T > tryReceiveFiltered(
    Filter && filter
)

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 transportPendingMessageCount

inline size_t transportPendingMessageCount() const

Messages waiting in per-Connection ingress lanes, outside the bounded destination FIFO. Diagnostic only; normal model backpressure uses OutPort/Connection and does not scan every producer lane.

function transportOverflowEvents

inline virtual uint64_t transportOverflowEvents() const override

Reimplements: chronon::sender::IMultiProducerPort::transportOverflowEvents

Number of direct-lane pushes rejected by the physical ring. Nonzero means scheduler run-ahead exceeded provisioned transport storage.

function transparentBroadcastEligible

inline bool transparentBroadcastEligible() const

function threadQueueAdmissionOccupancy

inline size_t threadQueueAdmissionOccupancy(
    size_t queue_id,
    uint64_t send_cycle
) const

function threadQueueAdmissionMinArrivalCycle

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

function storageCapacity

inline size_t storageCapacity() const

function sharedFifoHighWatermark

inline size_t sharedFifoHighWatermark() const

Largest aggregate destination FIFO occupancy observed since creation.

function setProducerProgress

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

Reimplements: chronon::sender::IMultiProducerPort::setProducerProgress

Resolve one producer completed_cycle atomic per MPSC connection (by the connection's source unit), aligned with mpsc_connections_ (conn_id order). Used by the epoch-free safety gate and by selective-flush retirement: every direct lane must have a scheduler progress publication before a visible future head can prove that all pre-flush publishes are complete.

function setCapacity

inline void setCapacity(
    size_t capacity
)

function resetSelectiveCancellation

inline void resetSelectiveCancellation()

Compatibility no-op. Selective predicates are immutable, independently scoped, and retire automatically; clearing live state could resurrect a message canceled by an overlapping flush.

function registerTransactionProducer

inline detail::ProducerDestinationTransactionState * registerTransactionProducer(
    Unit * producer,
    ConnectionBase * connection
)

Register one connection in its producer-owned destination transaction group. Topology construction is single-threaded; after initialization a group is touched only by its source Unit's worker.

function registerSharedBroadcastConnection

inline void registerSharedBroadcastConnection(
    Connection< T > * conn
)

Register one source lane selected by transparent broadcast discovery.

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 registerProducerThread

inline size_t registerProducerThread(
    size_t thread_id,
    bool track_admission
)

function registerMPSCConnection

inline void registerMPSCConnection(
    ConnectionBase * conn
)

Register an MPSC-mode Connection with this InPort. Connections are kept sorted by conn_id for stable topology/progress bookkeeping; payload ordering is handled by the direct-lane frontier.

function registerIncomingDelay

inline void registerIncomingDelay(
    uint32_t delay
)

Initialization-only connection metadata for selective-flush retirement.

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 producerProgressFullyResolved

inline virtual bool producerProgressFullyResolved() const override

Reimplements: chronon::sender::IMultiProducerPort::producerProgressFullyResolved

True iff every MPSC lane has a resolved producer-progress atomic. An unresolved lane vetoes epoch-free execution. Conservative default: false.

function prepareConsumerCycle

inline virtual void prepareConsumerCycle(
    uint64_t current_cycle
) override

Reimplements: chronon::sender::PortBase::prepareConsumerCycle

Receiver-cycle hook registered only for bounded MPSC InPorts.

function policy

inline PortPolicy policy() const

Return: the cancellation policy for this InPort.

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 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 flush

template <auto KeyFn>
inline void flush(
    FlushRange keep_range
)

Apply one receiver-owned range predicate to messages enqueued before the current receiver cycle. Messages enqueued in this cycle or later are post-flush and bypass it. The operation is independent of PortPolicy, queue backend, producer count, and edge delay.

function finalizeTransparentBroadcastReplay

inline bool finalizeTransparentBroadcastReplay(
    size_t producer_count
)

Finalize the initialization-time direct replay plan for a complete bus.

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 receiver-owned predicates can decide whether the message predates a 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
)

Compatibility spelling: cancel keys strictly above watermark.

function cancelOutsideInclusive

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

Compatibility spelling: keep only [min_keep, max_keep].

function cancelOlderThan

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

Compatibility spelling: cancel keys strictly below watermark.

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 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();

Friends

friend InPortSelectiveFlushTestAccess

friend struct InPortSelectiveFlushTestAccess(
    InPortSelectiveFlushTestAccess 
);

friend Connection< T >

friend class Connection< T >(
    Connection< T > 
);

---

Updated on 2026-07-23 at 16:24:33 +0000