Struct DagExecutor 

pub struct DagExecutor { /* private fields */ }

Ring 0 DAG executor for event processing.

Processes events through a finalized StreamingDag in topological order using the pre-computed RoutingTable for O(1) dispatch.

Construction

let dag = DagBuilder::new()
    .source("src", schema.clone())
    .operator("transform", schema.clone())
    .connect("src", "transform")
    .sink_for("transform", "out", schema.clone())
    .build()?;

let mut executor = DagExecutor::from_dag(&dag);
executor.register_operator(transform_id, Box::new(my_operator));
executor.process_event(src_id, event)?;
let outputs = executor.take_sink_outputs(out_id);

Implementations

impl DagExecutor

pub fn from_dag(dag: &StreamingDag) -> Self

Creates a new executor from a finalized StreamingDag.

Allocates all per-node state (input queues, runtimes, sink buffers) up front in Ring 2. The hot path (process_event) is allocation-free.

Arguments

• dag - A finalized StreamingDag topology

pub fn register_operator(&mut self, node: NodeId, operator: Box<dyn Operator>)

Registers an operator for a node.

Nodes without registered operators act as passthrough: events are forwarded to downstream nodes unchanged. This is the default for source and sink nodes.

Arguments

• node - The node ID to register the operator for
• operator - The operator implementation

pub fn process_event( &mut self, source_node: NodeId, event: Event, ) -> Result<(), DagError>

Processes an event from a source node through the entire DAG.

The event is enqueued at the source node, then all nodes are processed in topological order. Events produced by operators are routed to downstream nodes via the RoutingTable. Sink outputs are collected and can be retrieved via take_sink_outputs().

Arguments

• source_node - The source node to inject the event
• event - The event to process

Errors

Returns DagError::NodeNotFound if the source node is out of bounds.

pub fn take_sink_outputs(&mut self, sink_node: NodeId) -> Vec<Event>

Takes collected sink outputs for a given sink node.

Returns all events that reached this sink during prior process_event calls, draining the internal buffer.

pub fn take_all_sink_outputs(&mut self) -> FxHashMap<NodeId, Vec<Event>>

Takes all sink outputs across all sink nodes.

pub fn metrics(&self) -> &DagExecutorMetrics

Returns a reference to the executor metrics.

pub fn reset_metrics(&mut self)

Resets all executor metrics to zero.

pub fn source_nodes(&self) -> &[NodeId]

Returns the source node IDs.

pub fn sink_nodes(&self) -> &[NodeId]

Returns the sink node IDs.

pub fn node_type(&self, node: NodeId) -> Option<DagNodeType>

Returns the node type for a given node ID.

pub fn checkpoint(&self) -> FxHashMap<NodeId, OperatorState>

Checkpoints all registered operators.

Returns a map of NodeId to OperatorState for all nodes that have registered operators.

pub fn restore( &mut self, states: &FxHashMap<NodeId, OperatorState>, ) -> Result<(), DagError>

Restores operator state from a checkpoint snapshot.

Iterates the provided states and calls operator.restore() on each registered operator.

Errors

Returns DagError::RestoreFailed if any operator fails to restore.

pub fn inject_events(&mut self, node_id: NodeId, events: Vec<Event>)

Injects events into a node’s input queue.

Used during recovery to repopulate queues with buffered events.

pub fn input_count(&self, node_id: NodeId) -> usize

Returns the number of incoming edges for a node.

pub fn process_checkpoint_barrier( &mut self, _barrier: &CheckpointBarrier, ) -> FxHashMap<NodeId, OperatorState>

Snapshots all registered operators in topological order.

Takes the barrier for consistency (future use with epoch tracking). In the synchronous single-threaded executor, topological ordering guarantees upstream-first snapshots.

Trait Implementations

impl Debug for DagExecutor

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.