Struct BspEngine 

pub struct BspEngine<S = ()> { /* private fields */ }

The dynamic scheduler that organizes nodes into supersteps and executes them concurrently (bounded), re-executing nodes as loops deliver new activation tokens.

S is the consumer-owned run state; it defaults to () for pure dataflow graphs.

Implementations

impl<S> BspEngine<S>

pub fn new(graph: BspGraph<S>) -> Self

Instantiates a new BSP engine using a compiled graph layout.

pub fn with_context(self, ctx: BspContext) -> Self

Attaches a shared execution context for dependency injection.

pub const fn context(&self) -> &BspContext

Returns a reference to the active execution context.

pub const fn graph(&self) -> &BspGraph<S>

The graph this engine runs.

pub fn with_interceptor(self, interceptor: Arc<dyn BspInterceptor>) -> Self

Attaches a performance and audit logging interceptor.

pub const fn with_max_concurrency(self, limit: usize) -> Self

Configures the maximum concurrency (throttling limit) for task execution within a superstep.

The throttle is scoped to this engine instance (buffer_unordered per superstep); a nested graph-as-node runs a child engine with its own independent throttle, so nesting can never deadlock or halve capacity per level.

pub const fn with_max_steps(self, steps: u64) -> Self

Configures the global superstep budget. Exceeding it yields EngineResult::StepLimitReached with the live run state — never a silent salvage.

pub const fn with_max_pending_tokens(self, cap: u64) -> Self

Configures the per-edge pending-token cap (token-flood guard).

pub fn with_branch_transform(self, transform: BranchTransform<S>) -> Self

Attaches a transform applied to each branch state clone at multi-node frontier formation, right after the clone is taken from the trunk.

The second argument is the branch’s ordinal within the frontier formation (declaration order, 0-based). This is the hook for fork-fan-out semantics where sibling clones must not all inherit the full trunk — the reference consumer (llm-net) zeroes branch metrics for every sibling after the first so boundary rollups never double-count the parent’s history.

The default (None) is the identity: every branch is an exact clone. The transform never runs for single-node frontiers (they borrow the trunk directly) or on the sequential path (run_sequential does no branch cloning).

pub fn validate(&self) -> Result<(), BspError>

Validates the graph (errors only), looking for unfireable cycles, missing links, invalid port configurations, and unregistered condition/merger references.

pub fn validate_with_report(&self) -> ValidationReport

Validates the graph, additionally returning non-fatal warnings (loop-aware analysis: unbounded cycles, unreachable nodes, dead ends, All-joins on cycles, joins without mergers).

impl<S: Clone> BspEngine<S>

pub async fn run(&self, state: S) -> Result<EngineResult<S>, BspError>

Initiates a clean run of the graph with the given initial consumer state (run(()) for pure dataflow graphs).

Requires S: Clone because concurrent frontiers execute against per-task clones (see the module docs); single-node frontiers borrow the trunk directly. For !Clone states, use BspEngine::run_sequential.

pub async fn run_with_entries( &self, state: S, entries: impl IntoIterator<Item = NodeId>, ) -> Result<EngineResult<S>, BspError>

Like BspEngine::run, but forces the given nodes as the run’s entry frontier instead of the graph’s declared entries.

This is the resumable-control-flow primitive for hosts that checkpoint only the consumer state S (not the engine’s SuspendedExecution snapshot): resuming is “start a fresh run forced at the recorded cursor node(s)”, exactly the model the reference consumer’s resume_at graph cursor uses. Forced nodes join the first frontier unconditionally, even when their incoming edges hold no tokens.

pub async fn resume( &self, suspended: SuspendedExecution<S>, injections: IndexMap<InputPin, BspValue>, ) -> Result<EngineResult<S>, BspError>

Resumes a suspended graph execution, injecting external values into the waiting ports. Iteration and execution counters continue — they never reset.

impl<S> BspEngine<S>

pub async fn run_sequential( &self, state: S, ) -> Result<EngineResult<S>, BspError>

Like BspEngine::run but without a Clone bound on S: frontier tasks execute sequentially, each borrowing the trunk state. Linear chains and loops behave identically; parallel frontiers serialize and share the trunk in declaration order (no isolation, no merging).

pub async fn resume_sequential( &self, suspended: SuspendedExecution<S>, injections: IndexMap<InputPin, BspValue>, ) -> Result<EngineResult<S>, BspError>

Sequential counterpart of BspEngine::resume for !Clone states.