Struct FlowHandle 

pub struct FlowHandle { /* private fields */ }

A reference to a spawned flow, returned by super::runtime::Runtime::spawn.

Holding a FlowHandle does not keep the flow alive — it is a way to (a) read its FlowId for addressing with Send, and (b) block the calling native thread until it finishes via FlowHandle::join. Dropping without joining is fine (fire-and-forget).

Implementations

impl FlowHandle

pub fn id(&self) -> FlowId

pub fn join(self) -> FlowOutcome

Block the current (native) thread until the flow terminates. Never call from inside a worker / from bytecode.

Returns FlowOutcome::Failed rather than blocking forever if the flow was destroyed without producing an outcome — most commonly Runtime::shutdown while the flow was suspended, since shutdown does not drain flows out of the timer or the worker deques. The message comes from super::error::RuntimeError::Abandoned, so it is distinguishable from a flow that genuinely faulted.

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 38)

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 50)

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 67)

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 try_join(&self) -> Option<FlowOutcome>

The outcome if the flow has already terminated, None if it is still running. Never waits.

For embedders that drive their own loop and cannot surrender a thread to FlowHandle::join. Takes &self, so it can be polled until it answers; note that the outcome is handed out exactly once, and a further poll after that reports super::error::RuntimeError::AlreadyCollected as FlowOutcome::Failed rather than repeating it.

pub fn join_timeout(&self, timeout: Duration) -> Option<FlowOutcome>

Wait up to timeout for the flow to terminate; None if it is still running when the bound elapses.

This is the variant to reach for in anything with a deadline — a control loop, a watchdog, a test harness — since it is the only join whose worst-case duration the caller chooses.

pub fn join_deadline(&self, deadline: Instant) -> Option<FlowOutcome>

FlowHandle::join_timeout against an absolute deadline, for callers that already track one and must not have it drift across repeated waits.