Struct Runtime 

pub struct Runtime { /* private fields */ }

A running Byteflow runtime: worker pool + timer thread over one shared Chunk.

Owns M:N scheduling for flows (spawn, yield, sleep, mailboxes, FlowCap resolution, supervised restarts). Optional trace JIT when built with feature = "jit" and enabled in RuntimeConfig::jit.

Construct with Runtime::new (no natives) or Runtime::with_natives when the chunk uses CallNative / crate::std_native_table.

Implementations

impl Runtime

pub fn new(chunk: Chunk) -> Result<Self, SpawnError>

Convenience constructor for chunks that never call out through Opcode::CallNative. Equivalent to Runtime::with_natives(chunk, NativeTable::empty()).

Returns SpawnError instead of panicking: verify failures and OS thread-spawn refusals are category-A errors (see docs::error_model).

pub fn with_natives( chunk: Chunk, natives: Arc<NativeTable>, ) -> Result<Self, SpawnError>

Construct a runtime whose flows can call into natives via Opcode::CallNative — the host FFI boundary.

pub fn with_config( chunk: Chunk, config: RuntimeConfig, ) -> Result<Self, SpawnError>

Examples found in repository

examples/jit_loop.rs (lines 24-35)

fn run_once(jit: bool) -> Result<(), Box<dyn std::error::Error>> {
    let chunk = loop_chunk(1_000_000);
    let rt = Runtime::with_config(
        chunk,
        RuntimeConfig {
            workers: 1,
            quantum: 50_000_000,
            jit: JitConfig {
                enabled: jit,
                hot_threshold: 1,
            },
            ..Default::default()
        },
    )?;
    let start = Instant::now();
    let outcome = rt.spawn(0, &[])?.join();
    let snapshot = rt.metrics();
    rt.shutdown();
    let elapsed = start.elapsed();
    match outcome {
        byteflow::FlowOutcome::Completed(Value::Int(n)) => {
            println!("jit={jit} result={n} elapsed={elapsed:?} metrics={snapshot}");
            Ok(())
        }
        other => Err(format!("unexpected outcome: {other:?}").into()),
    }
}

examples/crash_and_restart.rs (lines 16-24)

fn main() {
    let rt = match Runtime::with_config(
        samples::boom(),
        RuntimeConfig {
            workers: 1,
            quantum: 1_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(boom) = rt.function_index("boom") else {
        eprintln!("missing boom");
        std::process::exit(1);
    };
    let sup = match Supervisor::with_config(
        rt.spawner(),
        SupervisorConfig {
            max_restarts: 2,
            max_period: Duration::from_secs(5),
            strategy: byteflow::RestartStrategy::OneForOne,
        },
    ) {
        Ok(s) => s,
        Err(e) => {
            eprintln!("supervisor: {e}");
            std::process::exit(1);
        }
    };
    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
    {
        eprintln!("start_child: {e}");
        std::process::exit(1);
    }

    let start = std::time::Instant::now();
    while !sup.intensity_exceeded() {
        if start.elapsed() > Duration::from_secs(2) {
            eprintln!("supervisor did not hit intensity in time");
            std::process::exit(1);
        }
        thread::sleep(Duration::from_millis(5));
    }

    let metrics = rt.metrics();
    println!("intensity exceeded after {} failures", metrics.processes_failed);
    println!("{metrics}");
    sup.shutdown();
    rt.shutdown();
}

examples/throughput.rs (lines 34-42)

fn main() {
    let n: u32 = match std::env::args().nth(1) {
        Some(s) => match s.parse() {
            Ok(v) => v,
            Err(_) => 50_000,
        },
        None => 50_000,
    };

    let workers = match std::thread::available_parallelism() {
        Ok(p) => p.get(),
        Err(_) => 1,
    };

    let rt = match Runtime::with_config(
        trivial_chunk(),
        RuntimeConfig {
            workers,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(worker_fn) = rt.function_index("worker") else {
        eprintln!("missing worker");
        std::process::exit(1);
    };

    let start = Instant::now();
    let mut handles = Vec::with_capacity(n as usize);
    for _ in 0..n {
        match rt.spawn(worker_fn, &[]) {
            Ok(h) => handles.push(h),
            Err(e) => {
                eprintln!("spawn: {e}");
                std::process::exit(1);
            }
        }
    }
    let mut ok = 0u32;
    for h in handles {
        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
            ok += 1;
        }
    }
    let elapsed = start.elapsed();
    let metrics = rt.metrics();
    rt.shutdown();

    let secs = elapsed.as_secs_f64().max(1e-9);
    println!("processes={ok}/{n}");
    println!("workers={workers}");
    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
    println!("spawns_per_sec={:.0}", ok as f64 / secs);
    println!("{metrics}");
}

pub fn with_std_natives_and_config( chunk: Chunk, config: RuntimeConfig, ) -> Result<Self, SpawnError>

Like Runtime::with_config but wires crate::std_native_table_with using RuntimeConfig::output for the print native.

pub fn with_natives_and_config( chunk: Chunk, natives: Arc<NativeTable>, config: RuntimeConfig, ) -> Result<Self, SpawnError>

Verify chunk, spawn the worker pool + timer thread, and return a live Runtime.

Failures here mean the runtime was never started (no orphan threads): either the bytecode is invalid (SpawnError::VerifyFailed) or the OS refused a thread (SpawnError::ThreadSpawnFailed).

Examples found in repository

examples/ping_pong.rs (lines 11-20)

fn main() {
    let chunk = samples::ping_pong();
    let rt = match Runtime::with_natives_and_config(
        chunk,
        std_native_table(),
        RuntimeConfig {
            workers: 1,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(main) = rt.function_index("main") else {
        eprintln!("missing main");
        std::process::exit(1);
    };
    let handle = match rt.spawn(main, &[]) {
        Ok(h) => h,
        Err(e) => {
            eprintln!("spawn: {e}");
            std::process::exit(1);
        }
    };
    let outcome = handle.join();
    let metrics = rt.metrics();
    rt.shutdown();

    match outcome {
        FlowOutcome::Completed(Value::Int(2)) => {
            println!("pong replied 2 (Atomic Hop)");
            println!("{metrics}");
        }
        other => {
            eprintln!("unexpected {other:?}");
            std::process::exit(1);
        }
    }
}

examples/atomic_actors.rs (lines 23-32)

fn main() {
    let chunk = samples::atomic_request_reply();
    let rt = match Runtime::with_natives_and_config(
        chunk,
        std_native_table(),
        RuntimeConfig {
            workers: 2,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(main) = rt.function_index("main") else {
        eprintln!("missing main");
        std::process::exit(1);
    };
    let handle = match rt.spawn(main, &[]) {
        Ok(h) => h,
        Err(e) => {
            eprintln!("spawn: {e}");
            std::process::exit(1);
        }
    };
    let outcome = handle.join();
    let metrics = rt.metrics();
    rt.shutdown();

    match outcome {
        FlowOutcome::Completed(Value::Int(42)) => {
            println!("atomic request-reply ok: payload=42");
            println!("{metrics}");
        }
        other => {
            eprintln!("unexpected {other:?}");
            std::process::exit(1);
        }
    }
}

pub fn spawn( &self, function: u32, args: &[Value], ) -> Result<FlowHandle, SpawnError>

Spawn a top-level flow starting at function in this runtime’s chunk, returning a FlowHandle the caller can .join().

Returns SpawnError::BadFunction if function is out of range.

Examples found in repository

examples/jit_loop.rs (line 37)

fn run_once(jit: bool) -> Result<(), Box<dyn std::error::Error>> {
    let chunk = loop_chunk(1_000_000);
    let rt = Runtime::with_config(
        chunk,
        RuntimeConfig {
            workers: 1,
            quantum: 50_000_000,
            jit: JitConfig {
                enabled: jit,
                hot_threshold: 1,
            },
            ..Default::default()
        },
    )?;
    let start = Instant::now();
    let outcome = rt.spawn(0, &[])?.join();
    let snapshot = rt.metrics();
    rt.shutdown();
    let elapsed = start.elapsed();
    match outcome {
        byteflow::FlowOutcome::Completed(Value::Int(n)) => {
            println!("jit={jit} result={n} elapsed={elapsed:?} metrics={snapshot}");
            Ok(())
        }
        other => Err(format!("unexpected outcome: {other:?}").into()),
    }
}

examples/ping_pong.rs (line 31)

fn main() {
    let chunk = samples::ping_pong();
    let rt = match Runtime::with_natives_and_config(
        chunk,
        std_native_table(),
        RuntimeConfig {
            workers: 1,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(main) = rt.function_index("main") else {
        eprintln!("missing main");
        std::process::exit(1);
    };
    let handle = match rt.spawn(main, &[]) {
        Ok(h) => h,
        Err(e) => {
            eprintln!("spawn: {e}");
            std::process::exit(1);
        }
    };
    let outcome = handle.join();
    let metrics = rt.metrics();
    rt.shutdown();

    match outcome {
        FlowOutcome::Completed(Value::Int(2)) => {
            println!("pong replied 2 (Atomic Hop)");
            println!("{metrics}");
        }
        other => {
            eprintln!("unexpected {other:?}");
            std::process::exit(1);
        }
    }
}

examples/atomic_actors.rs (line 43)

fn main() {
    let chunk = samples::atomic_request_reply();
    let rt = match Runtime::with_natives_and_config(
        chunk,
        std_native_table(),
        RuntimeConfig {
            workers: 2,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(main) = rt.function_index("main") else {
        eprintln!("missing main");
        std::process::exit(1);
    };
    let handle = match rt.spawn(main, &[]) {
        Ok(h) => h,
        Err(e) => {
            eprintln!("spawn: {e}");
            std::process::exit(1);
        }
    };
    let outcome = handle.join();
    let metrics = rt.metrics();
    rt.shutdown();

    match outcome {
        FlowOutcome::Completed(Value::Int(42)) => {
            println!("atomic request-reply ok: payload=42");
            println!("{metrics}");
        }
        other => {
            eprintln!("unexpected {other:?}");
            std::process::exit(1);
        }
    }
}

examples/throughput.rs (line 57)

fn main() {
    let n: u32 = match std::env::args().nth(1) {
        Some(s) => match s.parse() {
            Ok(v) => v,
            Err(_) => 50_000,
        },
        None => 50_000,
    };

    let workers = match std::thread::available_parallelism() {
        Ok(p) => p.get(),
        Err(_) => 1,
    };

    let rt = match Runtime::with_config(
        trivial_chunk(),
        RuntimeConfig {
            workers,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(worker_fn) = rt.function_index("worker") else {
        eprintln!("missing worker");
        std::process::exit(1);
    };

    let start = Instant::now();
    let mut handles = Vec::with_capacity(n as usize);
    for _ in 0..n {
        match rt.spawn(worker_fn, &[]) {
            Ok(h) => handles.push(h),
            Err(e) => {
                eprintln!("spawn: {e}");
                std::process::exit(1);
            }
        }
    }
    let mut ok = 0u32;
    for h in handles {
        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
            ok += 1;
        }
    }
    let elapsed = start.elapsed();
    let metrics = rt.metrics();
    rt.shutdown();

    let secs = elapsed.as_secs_f64().max(1e-9);
    println!("processes={ok}/{n}");
    println!("workers={workers}");
    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
    println!("spawns_per_sec={:.0}", ok as f64 / secs);
    println!("{metrics}");
}

pub fn spawner(&self) -> RuntimeSpawner

A cheap, Send + Sync handle that can spawn processes into this runtime from any thread, independent of Runtime’s own lifetime bookkeeping (worker JoinHandles). Used by super::supervisor::Supervisor.

Examples found in repository

examples/crash_and_restart.rs (line 36)

fn main() {
    let rt = match Runtime::with_config(
        samples::boom(),
        RuntimeConfig {
            workers: 1,
            quantum: 1_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(boom) = rt.function_index("boom") else {
        eprintln!("missing boom");
        std::process::exit(1);
    };
    let sup = match Supervisor::with_config(
        rt.spawner(),
        SupervisorConfig {
            max_restarts: 2,
            max_period: Duration::from_secs(5),
            strategy: byteflow::RestartStrategy::OneForOne,
        },
    ) {
        Ok(s) => s,
        Err(e) => {
            eprintln!("supervisor: {e}");
            std::process::exit(1);
        }
    };
    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
    {
        eprintln!("start_child: {e}");
        std::process::exit(1);
    }

    let start = std::time::Instant::now();
    while !sup.intensity_exceeded() {
        if start.elapsed() > Duration::from_secs(2) {
            eprintln!("supervisor did not hit intensity in time");
            std::process::exit(1);
        }
        thread::sleep(Duration::from_millis(5));
    }

    let metrics = rt.metrics();
    println!("intensity exceeded after {} failures", metrics.processes_failed);
    println!("{metrics}");
    sup.shutdown();
    rt.shutdown();
}

pub fn supervisor(&self) -> Result<Supervisor, SpawnError>

A super::supervisor::Supervisor bound to this runtime, ready to take supervised children (design notes §15).

pub fn function_index(&self, name: &str) -> Option<u32>

Look up a function by name in the runtime’s chunk — convenience for callers that built their chunk with crate::Program and don’t want to thread raw indices through their own code.

Examples found in repository

examples/ping_pong.rs (line 27)

fn main() {
    let chunk = samples::ping_pong();
    let rt = match Runtime::with_natives_and_config(
        chunk,
        std_native_table(),
        RuntimeConfig {
            workers: 1,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(main) = rt.function_index("main") else {
        eprintln!("missing main");
        std::process::exit(1);
    };
    let handle = match rt.spawn(main, &[]) {
        Ok(h) => h,
        Err(e) => {
            eprintln!("spawn: {e}");
            std::process::exit(1);
        }
    };
    let outcome = handle.join();
    let metrics = rt.metrics();
    rt.shutdown();

    match outcome {
        FlowOutcome::Completed(Value::Int(2)) => {
            println!("pong replied 2 (Atomic Hop)");
            println!("{metrics}");
        }
        other => {
            eprintln!("unexpected {other:?}");
            std::process::exit(1);
        }
    }
}

examples/atomic_actors.rs (line 39)

fn main() {
    let chunk = samples::atomic_request_reply();
    let rt = match Runtime::with_natives_and_config(
        chunk,
        std_native_table(),
        RuntimeConfig {
            workers: 2,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(main) = rt.function_index("main") else {
        eprintln!("missing main");
        std::process::exit(1);
    };
    let handle = match rt.spawn(main, &[]) {
        Ok(h) => h,
        Err(e) => {
            eprintln!("spawn: {e}");
            std::process::exit(1);
        }
    };
    let outcome = handle.join();
    let metrics = rt.metrics();
    rt.shutdown();

    match outcome {
        FlowOutcome::Completed(Value::Int(42)) => {
            println!("atomic request-reply ok: payload=42");
            println!("{metrics}");
        }
        other => {
            eprintln!("unexpected {other:?}");
            std::process::exit(1);
        }
    }
}

examples/crash_and_restart.rs (line 31)

fn main() {
    let rt = match Runtime::with_config(
        samples::boom(),
        RuntimeConfig {
            workers: 1,
            quantum: 1_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(boom) = rt.function_index("boom") else {
        eprintln!("missing boom");
        std::process::exit(1);
    };
    let sup = match Supervisor::with_config(
        rt.spawner(),
        SupervisorConfig {
            max_restarts: 2,
            max_period: Duration::from_secs(5),
            strategy: byteflow::RestartStrategy::OneForOne,
        },
    ) {
        Ok(s) => s,
        Err(e) => {
            eprintln!("supervisor: {e}");
            std::process::exit(1);
        }
    };
    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
    {
        eprintln!("start_child: {e}");
        std::process::exit(1);
    }

    let start = std::time::Instant::now();
    while !sup.intensity_exceeded() {
        if start.elapsed() > Duration::from_secs(2) {
            eprintln!("supervisor did not hit intensity in time");
            std::process::exit(1);
        }
        thread::sleep(Duration::from_millis(5));
    }

    let metrics = rt.metrics();
    println!("intensity exceeded after {} failures", metrics.processes_failed);
    println!("{metrics}");
    sup.shutdown();
    rt.shutdown();
}

examples/throughput.rs (line 49)

fn main() {
    let n: u32 = match std::env::args().nth(1) {
        Some(s) => match s.parse() {
            Ok(v) => v,
            Err(_) => 50_000,
        },
        None => 50_000,
    };

    let workers = match std::thread::available_parallelism() {
        Ok(p) => p.get(),
        Err(_) => 1,
    };

    let rt = match Runtime::with_config(
        trivial_chunk(),
        RuntimeConfig {
            workers,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(worker_fn) = rt.function_index("worker") else {
        eprintln!("missing worker");
        std::process::exit(1);
    };

    let start = Instant::now();
    let mut handles = Vec::with_capacity(n as usize);
    for _ in 0..n {
        match rt.spawn(worker_fn, &[]) {
            Ok(h) => handles.push(h),
            Err(e) => {
                eprintln!("spawn: {e}");
                std::process::exit(1);
            }
        }
    }
    let mut ok = 0u32;
    for h in handles {
        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
            ok += 1;
        }
    }
    let elapsed = start.elapsed();
    let metrics = rt.metrics();
    rt.shutdown();

    let secs = elapsed.as_secs_f64().max(1e-9);
    println!("processes={ok}/{n}");
    println!("workers={workers}");
    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
    println!("spawns_per_sec={:.0}", ok as f64 / secs);
    println!("{metrics}");
}

pub fn metrics(&self) -> RuntimeMetricsSnapshot

Examples found in repository

examples/jit_loop.rs (line 38)

fn run_once(jit: bool) -> Result<(), Box<dyn std::error::Error>> {
    let chunk = loop_chunk(1_000_000);
    let rt = Runtime::with_config(
        chunk,
        RuntimeConfig {
            workers: 1,
            quantum: 50_000_000,
            jit: JitConfig {
                enabled: jit,
                hot_threshold: 1,
            },
            ..Default::default()
        },
    )?;
    let start = Instant::now();
    let outcome = rt.spawn(0, &[])?.join();
    let snapshot = rt.metrics();
    rt.shutdown();
    let elapsed = start.elapsed();
    match outcome {
        byteflow::FlowOutcome::Completed(Value::Int(n)) => {
            println!("jit={jit} result={n} elapsed={elapsed:?} metrics={snapshot}");
            Ok(())
        }
        other => Err(format!("unexpected outcome: {other:?}").into()),
    }
}

examples/ping_pong.rs (line 39)

fn main() {
    let chunk = samples::ping_pong();
    let rt = match Runtime::with_natives_and_config(
        chunk,
        std_native_table(),
        RuntimeConfig {
            workers: 1,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(main) = rt.function_index("main") else {
        eprintln!("missing main");
        std::process::exit(1);
    };
    let handle = match rt.spawn(main, &[]) {
        Ok(h) => h,
        Err(e) => {
            eprintln!("spawn: {e}");
            std::process::exit(1);
        }
    };
    let outcome = handle.join();
    let metrics = rt.metrics();
    rt.shutdown();

    match outcome {
        FlowOutcome::Completed(Value::Int(2)) => {
            println!("pong replied 2 (Atomic Hop)");
            println!("{metrics}");
        }
        other => {
            eprintln!("unexpected {other:?}");
            std::process::exit(1);
        }
    }
}

examples/atomic_actors.rs (line 51)

fn main() {
    let chunk = samples::atomic_request_reply();
    let rt = match Runtime::with_natives_and_config(
        chunk,
        std_native_table(),
        RuntimeConfig {
            workers: 2,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(main) = rt.function_index("main") else {
        eprintln!("missing main");
        std::process::exit(1);
    };
    let handle = match rt.spawn(main, &[]) {
        Ok(h) => h,
        Err(e) => {
            eprintln!("spawn: {e}");
            std::process::exit(1);
        }
    };
    let outcome = handle.join();
    let metrics = rt.metrics();
    rt.shutdown();

    match outcome {
        FlowOutcome::Completed(Value::Int(42)) => {
            println!("atomic request-reply ok: payload=42");
            println!("{metrics}");
        }
        other => {
            eprintln!("unexpected {other:?}");
            std::process::exit(1);
        }
    }
}

examples/crash_and_restart.rs (line 64)

fn main() {
    let rt = match Runtime::with_config(
        samples::boom(),
        RuntimeConfig {
            workers: 1,
            quantum: 1_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(boom) = rt.function_index("boom") else {
        eprintln!("missing boom");
        std::process::exit(1);
    };
    let sup = match Supervisor::with_config(
        rt.spawner(),
        SupervisorConfig {
            max_restarts: 2,
            max_period: Duration::from_secs(5),
            strategy: byteflow::RestartStrategy::OneForOne,
        },
    ) {
        Ok(s) => s,
        Err(e) => {
            eprintln!("supervisor: {e}");
            std::process::exit(1);
        }
    };
    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
    {
        eprintln!("start_child: {e}");
        std::process::exit(1);
    }

    let start = std::time::Instant::now();
    while !sup.intensity_exceeded() {
        if start.elapsed() > Duration::from_secs(2) {
            eprintln!("supervisor did not hit intensity in time");
            std::process::exit(1);
        }
        thread::sleep(Duration::from_millis(5));
    }

    let metrics = rt.metrics();
    println!("intensity exceeded after {} failures", metrics.processes_failed);
    println!("{metrics}");
    sup.shutdown();
    rt.shutdown();
}

examples/throughput.rs (line 72)

fn main() {
    let n: u32 = match std::env::args().nth(1) {
        Some(s) => match s.parse() {
            Ok(v) => v,
            Err(_) => 50_000,
        },
        None => 50_000,
    };

    let workers = match std::thread::available_parallelism() {
        Ok(p) => p.get(),
        Err(_) => 1,
    };

    let rt = match Runtime::with_config(
        trivial_chunk(),
        RuntimeConfig {
            workers,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(worker_fn) = rt.function_index("worker") else {
        eprintln!("missing worker");
        std::process::exit(1);
    };

    let start = Instant::now();
    let mut handles = Vec::with_capacity(n as usize);
    for _ in 0..n {
        match rt.spawn(worker_fn, &[]) {
            Ok(h) => handles.push(h),
            Err(e) => {
                eprintln!("spawn: {e}");
                std::process::exit(1);
            }
        }
    }
    let mut ok = 0u32;
    for h in handles {
        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
            ok += 1;
        }
    }
    let elapsed = start.elapsed();
    let metrics = rt.metrics();
    rt.shutdown();

    let secs = elapsed.as_secs_f64().max(1e-9);
    println!("processes={ok}/{n}");
    println!("workers={workers}");
    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
    println!("spawns_per_sec={:.0}", ok as f64 / secs);
    println!("{metrics}");
}

pub fn live_flows(&self) -> usize

Number of flows currently registered in the directory — i.e. alive (running, ready, sleeping, or waiting), not counting ones that have already completed or failed.

pub fn worker_count(&self) -> usize

pub fn send(&self, target: FlowId, message: Value) -> Result<(), SendError>

Deliver an Atomic Hop (Value::Message) to target from the embedder (not from bytecode).

Host trust boundary

This path takes a FlowId directly — no Cap required. The host is trusted; bytecode must use Value::Cap via Opcode::Send / Ask. Host-injected messages are not re-stamped (sender / reply_cap stay as built). Bare scalars are rejected (SendError::NotAHop).

pub fn mint_cap(&self, flow: FlowId) -> Result<CapId, LifecycleError>

Mint a SEND|ASK Cap for a live flow (host equivalent of SelfPid).

pub fn grant_cap( &self, holder: FlowId, target: FlowId, ) -> Result<CapId, LifecycleError>

Mint a Cap held by holder targeting target (host introduction).

pub fn monitor( &self, owner: FlowId, target: FlowId, ) -> Result<MonitorRef, LifecycleError>

Watch target; when it exits, owner receives a crate::TAG_SYS_DOWN hop.

Both flows must be live. owner == target is [LifecycleError::SelfRelation].

pub fn demonitor( &self, owner: FlowId, monitor: MonitorRef, ) -> Result<(), LifecycleError>

Drop monitor if owner still owns it.

pub fn link(&self, a: FlowId, b: FlowId) -> Result<LinkId, LifecycleError>

Bidirectional link. Abnormal exit of either side kills the peer.

pub fn unlink(&self, owner: FlowId, link: LinkId) -> Result<(), LifecycleError>

Drop link if owner is one of the endpoints.

pub fn register_name( &self, name: &str, cap: CapId, ) -> Result<(), LifecycleError>

Bind name to a live Cap (address). Names are swept when that flow exits.

pub fn whereis(&self, name: &str) -> Result<Option<CapId>, LifecycleError>

Look up a registered Cap, or None if the name is free / was swept.

pub fn kill(&self, id: FlowId) -> Result<(), LifecycleError>

Cooperative abort. Parked flows finalize immediately; a running flow dies at the next quantum with super::monitor::FlowExitReason::Killed.

pub fn mint_admin_cap(&self, holder: FlowId) -> Result<CapId, LifecycleError>

Mint a scheduler ADMIN cap for a live flow (host introduction).

pub fn admin_kill( &self, holder: FlowId, cap: CapId, target: FlowId, ) -> Result<(), LifecycleError>

Kill target only if holder presents a live ADMIN scheduler cap.

pub fn admin_top_up_cpu( &self, holder: FlowId, cap: CapId, target: FlowId, extra: i64, ) -> Result<(), LifecycleError>

Top up target’s CPU budget. Requires a live ADMIN scheduler cap.

pub fn unregister_name(&self, name: &str) -> Result<bool, LifecycleError>

Remove a name without waiting for the flow to exit. false if unknown.

pub fn reload_chunk(&mut self, chunk: Chunk) -> Result<(), SpawnError>

Replace the image used by new host Self::spawn calls and drop JIT traces. Live flows and bytecode Spawn keep the parent’s existing Vm chunk.

pub fn shutdown(self)

Stop accepting new scheduling work and join every worker + the timer thread. Processes that are mid-quantum are allowed to reach their next natural suspension point; this does not forcibly abort running bytecode (there is no safe way to do that to an OS thread mid-instruction — see design notes §11 on why preemption here is cooperative/budgeted rather than signal-based).

Examples found in repository

examples/jit_loop.rs (line 39)

fn run_once(jit: bool) -> Result<(), Box<dyn std::error::Error>> {
    let chunk = loop_chunk(1_000_000);
    let rt = Runtime::with_config(
        chunk,
        RuntimeConfig {
            workers: 1,
            quantum: 50_000_000,
            jit: JitConfig {
                enabled: jit,
                hot_threshold: 1,
            },
            ..Default::default()
        },
    )?;
    let start = Instant::now();
    let outcome = rt.spawn(0, &[])?.join();
    let snapshot = rt.metrics();
    rt.shutdown();
    let elapsed = start.elapsed();
    match outcome {
        byteflow::FlowOutcome::Completed(Value::Int(n)) => {
            println!("jit={jit} result={n} elapsed={elapsed:?} metrics={snapshot}");
            Ok(())
        }
        other => Err(format!("unexpected outcome: {other:?}").into()),
    }
}

examples/ping_pong.rs (line 40)

fn main() {
    let chunk = samples::ping_pong();
    let rt = match Runtime::with_natives_and_config(
        chunk,
        std_native_table(),
        RuntimeConfig {
            workers: 1,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(main) = rt.function_index("main") else {
        eprintln!("missing main");
        std::process::exit(1);
    };
    let handle = match rt.spawn(main, &[]) {
        Ok(h) => h,
        Err(e) => {
            eprintln!("spawn: {e}");
            std::process::exit(1);
        }
    };
    let outcome = handle.join();
    let metrics = rt.metrics();
    rt.shutdown();

    match outcome {
        FlowOutcome::Completed(Value::Int(2)) => {
            println!("pong replied 2 (Atomic Hop)");
            println!("{metrics}");
        }
        other => {
            eprintln!("unexpected {other:?}");
            std::process::exit(1);
        }
    }
}

examples/atomic_actors.rs (line 52)

fn main() {
    let chunk = samples::atomic_request_reply();
    let rt = match Runtime::with_natives_and_config(
        chunk,
        std_native_table(),
        RuntimeConfig {
            workers: 2,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(main) = rt.function_index("main") else {
        eprintln!("missing main");
        std::process::exit(1);
    };
    let handle = match rt.spawn(main, &[]) {
        Ok(h) => h,
        Err(e) => {
            eprintln!("spawn: {e}");
            std::process::exit(1);
        }
    };
    let outcome = handle.join();
    let metrics = rt.metrics();
    rt.shutdown();

    match outcome {
        FlowOutcome::Completed(Value::Int(42)) => {
            println!("atomic request-reply ok: payload=42");
            println!("{metrics}");
        }
        other => {
            eprintln!("unexpected {other:?}");
            std::process::exit(1);
        }
    }
}

examples/crash_and_restart.rs (line 68)

fn main() {
    let rt = match Runtime::with_config(
        samples::boom(),
        RuntimeConfig {
            workers: 1,
            quantum: 1_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(boom) = rt.function_index("boom") else {
        eprintln!("missing boom");
        std::process::exit(1);
    };
    let sup = match Supervisor::with_config(
        rt.spawner(),
        SupervisorConfig {
            max_restarts: 2,
            max_period: Duration::from_secs(5),
            strategy: byteflow::RestartStrategy::OneForOne,
        },
    ) {
        Ok(s) => s,
        Err(e) => {
            eprintln!("supervisor: {e}");
            std::process::exit(1);
        }
    };
    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
    {
        eprintln!("start_child: {e}");
        std::process::exit(1);
    }

    let start = std::time::Instant::now();
    while !sup.intensity_exceeded() {
        if start.elapsed() > Duration::from_secs(2) {
            eprintln!("supervisor did not hit intensity in time");
            std::process::exit(1);
        }
        thread::sleep(Duration::from_millis(5));
    }

    let metrics = rt.metrics();
    println!("intensity exceeded after {} failures", metrics.processes_failed);
    println!("{metrics}");
    sup.shutdown();
    rt.shutdown();
}

examples/throughput.rs (line 73)

fn main() {
    let n: u32 = match std::env::args().nth(1) {
        Some(s) => match s.parse() {
            Ok(v) => v,
            Err(_) => 50_000,
        },
        None => 50_000,
    };

    let workers = match std::thread::available_parallelism() {
        Ok(p) => p.get(),
        Err(_) => 1,
    };

    let rt = match Runtime::with_config(
        trivial_chunk(),
        RuntimeConfig {
            workers,
            quantum: 10_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
            ..Default::default()
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(worker_fn) = rt.function_index("worker") else {
        eprintln!("missing worker");
        std::process::exit(1);
    };

    let start = Instant::now();
    let mut handles = Vec::with_capacity(n as usize);
    for _ in 0..n {
        match rt.spawn(worker_fn, &[]) {
            Ok(h) => handles.push(h),
            Err(e) => {
                eprintln!("spawn: {e}");
                std::process::exit(1);
            }
        }
    }
    let mut ok = 0u32;
    for h in handles {
        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
            ok += 1;
        }
    }
    let elapsed = start.elapsed();
    let metrics = rt.metrics();
    rt.shutdown();

    let secs = elapsed.as_secs_f64().max(1e-9);
    println!("processes={ok}/{n}");
    println!("workers={workers}");
    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
    println!("spawns_per_sec={:.0}", ok as f64 / secs);
    println!("{metrics}");
}

Trait Implementations

impl Drop for Runtime

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.