Skip to main content

Runtime

byteflow::scheduler

Struct Runtime 

Source 
pub struct Runtime { /* private fields */ }
Expand description

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§

Source§

impl Runtime

Source

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

Source

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.

Source

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

Examples found in repository?
examples/jit_loop.rs (lines 24-35)
22fn run_once(jit: bool) -> Result<(), Box<dyn std::error::Error>> {
23    let chunk = loop_chunk(1_000_000);
24    let rt = Runtime::with_config(
25        chunk,
26        RuntimeConfig {
27            workers: 1,
28            quantum: 50_000_000,
29            jit: JitConfig {
30                enabled: jit,
31                hot_threshold: 1,
32            },
33            ..Default::default()
34        },
35    )?;
36    let start = Instant::now();
37    let outcome = rt.spawn(0, &[])?.join();
38    let snapshot = rt.metrics();
39    rt.shutdown();
40    let elapsed = start.elapsed();
41    match outcome {
42        byteflow::FlowOutcome::Completed(Value::Int(n)) => {
43            println!("jit={jit} result={n} elapsed={elapsed:?} metrics={snapshot}");
44            Ok(())
45        }
46        other => Err(format!("unexpected outcome: {other:?}").into()),
47    }
48}
More examples
Hide additional examples
examples/crash_and_restart.rs (lines 16-24)
15fn main() {
16    let rt = match Runtime::with_config(
17        samples::boom(),
18        RuntimeConfig {
19            workers: 1,
20            quantum: 1_000,
21            mailbox: byteflow::MailboxConfig::DEFAULT,
22            ..Default::default()
23        },
24    ) {
25        Ok(rt) => rt,
26        Err(e) => {
27            eprintln!("runtime: {e}");
28            std::process::exit(1);
29        }
30    };
31    let Some(boom) = rt.function_index("boom") else {
32        eprintln!("missing boom");
33        std::process::exit(1);
34    };
35    let sup = match Supervisor::with_config(
36        rt.spawner(),
37        SupervisorConfig {
38            max_restarts: 2,
39            max_period: Duration::from_secs(5),
40        },
41    ) {
42        Ok(s) => s,
43        Err(e) => {
44            eprintln!("supervisor: {e}");
45            std::process::exit(1);
46        }
47    };
48    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
49    {
50        eprintln!("start_child: {e}");
51        std::process::exit(1);
52    }
53
54    let start = std::time::Instant::now();
55    while !sup.intensity_exceeded() {
56        if start.elapsed() > Duration::from_secs(2) {
57            eprintln!("supervisor did not hit intensity in time");
58            std::process::exit(1);
59        }
60        thread::sleep(Duration::from_millis(5));
61    }
62
63    let metrics = rt.metrics();
64    println!("intensity exceeded after {} failures", metrics.processes_failed);
65    println!("{metrics}");
66    sup.shutdown();
67    rt.shutdown();
68}
examples/throughput.rs (lines 34-42)
20fn main() {
21    let n: u32 = match std::env::args().nth(1) {
22        Some(s) => match s.parse() {
23            Ok(v) => v,
24            Err(_) => 50_000,
25        },
26        None => 50_000,
27    };
28
29    let workers = match std::thread::available_parallelism() {
30        Ok(p) => p.get(),
31        Err(_) => 1,
32    };
33
34    let rt = match Runtime::with_config(
35        trivial_chunk(),
36        RuntimeConfig {
37            workers,
38            quantum: 10_000,
39            mailbox: byteflow::MailboxConfig::DEFAULT,
40            ..Default::default()
41        },
42    ) {
43        Ok(rt) => rt,
44        Err(e) => {
45            eprintln!("runtime: {e}");
46            std::process::exit(1);
47        }
48    };
49    let Some(worker_fn) = rt.function_index("worker") else {
50        eprintln!("missing worker");
51        std::process::exit(1);
52    };
53
54    let start = Instant::now();
55    let mut handles = Vec::with_capacity(n as usize);
56    for _ in 0..n {
57        match rt.spawn(worker_fn, &[]) {
58            Ok(h) => handles.push(h),
59            Err(e) => {
60                eprintln!("spawn: {e}");
61                std::process::exit(1);
62            }
63        }
64    }
65    let mut ok = 0u32;
66    for h in handles {
67        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
68            ok += 1;
69        }
70    }
71    let elapsed = start.elapsed();
72    let metrics = rt.metrics();
73    rt.shutdown();
74
75    let secs = elapsed.as_secs_f64().max(1e-9);
76    println!("processes={ok}/{n}");
77    println!("workers={workers}");
78    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
79    println!("spawns_per_sec={:.0}", ok as f64 / secs);
80    println!("{metrics}");
81}
Source

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)
9fn main() {
10    let chunk = samples::ping_pong();
11    let rt = match Runtime::with_natives_and_config(
12        chunk,
13        std_native_table(),
14        RuntimeConfig {
15            workers: 1,
16            quantum: 10_000,
17            mailbox: byteflow::MailboxConfig::DEFAULT,
18            ..Default::default()
19        },
20    ) {
21        Ok(rt) => rt,
22        Err(e) => {
23            eprintln!("runtime: {e}");
24            std::process::exit(1);
25        }
26    };
27    let Some(main) = rt.function_index("main") else {
28        eprintln!("missing main");
29        std::process::exit(1);
30    };
31    let handle = match rt.spawn(main, &[]) {
32        Ok(h) => h,
33        Err(e) => {
34            eprintln!("spawn: {e}");
35            std::process::exit(1);
36        }
37    };
38    let outcome = handle.join();
39    let metrics = rt.metrics();
40    rt.shutdown();
41
42    match outcome {
43        FlowOutcome::Completed(Value::Int(2)) => {
44            println!("pong replied 2 (Atomic Hop)");
45            println!("{metrics}");
46        }
47        other => {
48            eprintln!("unexpected {other:?}");
49            std::process::exit(1);
50        }
51    }
52}
More examples
Hide additional examples
examples/atomic_actors.rs (lines 23-32)
21fn main() {
22    let chunk = samples::atomic_request_reply();
23    let rt = match Runtime::with_natives_and_config(
24        chunk,
25        std_native_table(),
26        RuntimeConfig {
27            workers: 2,
28            quantum: 10_000,
29            mailbox: byteflow::MailboxConfig::DEFAULT,
30            ..Default::default()
31        },
32    ) {
33        Ok(rt) => rt,
34        Err(e) => {
35            eprintln!("runtime: {e}");
36            std::process::exit(1);
37        }
38    };
39    let Some(main) = rt.function_index("main") else {
40        eprintln!("missing main");
41        std::process::exit(1);
42    };
43    let handle = match rt.spawn(main, &[]) {
44        Ok(h) => h,
45        Err(e) => {
46            eprintln!("spawn: {e}");
47            std::process::exit(1);
48        }
49    };
50    let outcome = handle.join();
51    let metrics = rt.metrics();
52    rt.shutdown();
53
54    match outcome {
55        FlowOutcome::Completed(Value::Int(42)) => {
56            println!("atomic request-reply ok: payload=42");
57            println!("{metrics}");
58        }
59        other => {
60            eprintln!("unexpected {other:?}");
61            std::process::exit(1);
62        }
63    }
64}
Source

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)
22fn run_once(jit: bool) -> Result<(), Box<dyn std::error::Error>> {
23    let chunk = loop_chunk(1_000_000);
24    let rt = Runtime::with_config(
25        chunk,
26        RuntimeConfig {
27            workers: 1,
28            quantum: 50_000_000,
29            jit: JitConfig {
30                enabled: jit,
31                hot_threshold: 1,
32            },
33            ..Default::default()
34        },
35    )?;
36    let start = Instant::now();
37    let outcome = rt.spawn(0, &[])?.join();
38    let snapshot = rt.metrics();
39    rt.shutdown();
40    let elapsed = start.elapsed();
41    match outcome {
42        byteflow::FlowOutcome::Completed(Value::Int(n)) => {
43            println!("jit={jit} result={n} elapsed={elapsed:?} metrics={snapshot}");
44            Ok(())
45        }
46        other => Err(format!("unexpected outcome: {other:?}").into()),
47    }
48}
More examples
Hide additional examples
examples/ping_pong.rs (line 31)
9fn main() {
10    let chunk = samples::ping_pong();
11    let rt = match Runtime::with_natives_and_config(
12        chunk,
13        std_native_table(),
14        RuntimeConfig {
15            workers: 1,
16            quantum: 10_000,
17            mailbox: byteflow::MailboxConfig::DEFAULT,
18            ..Default::default()
19        },
20    ) {
21        Ok(rt) => rt,
22        Err(e) => {
23            eprintln!("runtime: {e}");
24            std::process::exit(1);
25        }
26    };
27    let Some(main) = rt.function_index("main") else {
28        eprintln!("missing main");
29        std::process::exit(1);
30    };
31    let handle = match rt.spawn(main, &[]) {
32        Ok(h) => h,
33        Err(e) => {
34            eprintln!("spawn: {e}");
35            std::process::exit(1);
36        }
37    };
38    let outcome = handle.join();
39    let metrics = rt.metrics();
40    rt.shutdown();
41
42    match outcome {
43        FlowOutcome::Completed(Value::Int(2)) => {
44            println!("pong replied 2 (Atomic Hop)");
45            println!("{metrics}");
46        }
47        other => {
48            eprintln!("unexpected {other:?}");
49            std::process::exit(1);
50        }
51    }
52}
examples/atomic_actors.rs (line 43)
21fn main() {
22    let chunk = samples::atomic_request_reply();
23    let rt = match Runtime::with_natives_and_config(
24        chunk,
25        std_native_table(),
26        RuntimeConfig {
27            workers: 2,
28            quantum: 10_000,
29            mailbox: byteflow::MailboxConfig::DEFAULT,
30            ..Default::default()
31        },
32    ) {
33        Ok(rt) => rt,
34        Err(e) => {
35            eprintln!("runtime: {e}");
36            std::process::exit(1);
37        }
38    };
39    let Some(main) = rt.function_index("main") else {
40        eprintln!("missing main");
41        std::process::exit(1);
42    };
43    let handle = match rt.spawn(main, &[]) {
44        Ok(h) => h,
45        Err(e) => {
46            eprintln!("spawn: {e}");
47            std::process::exit(1);
48        }
49    };
50    let outcome = handle.join();
51    let metrics = rt.metrics();
52    rt.shutdown();
53
54    match outcome {
55        FlowOutcome::Completed(Value::Int(42)) => {
56            println!("atomic request-reply ok: payload=42");
57            println!("{metrics}");
58        }
59        other => {
60            eprintln!("unexpected {other:?}");
61            std::process::exit(1);
62        }
63    }
64}
examples/throughput.rs (line 57)
20fn main() {
21    let n: u32 = match std::env::args().nth(1) {
22        Some(s) => match s.parse() {
23            Ok(v) => v,
24            Err(_) => 50_000,
25        },
26        None => 50_000,
27    };
28
29    let workers = match std::thread::available_parallelism() {
30        Ok(p) => p.get(),
31        Err(_) => 1,
32    };
33
34    let rt = match Runtime::with_config(
35        trivial_chunk(),
36        RuntimeConfig {
37            workers,
38            quantum: 10_000,
39            mailbox: byteflow::MailboxConfig::DEFAULT,
40            ..Default::default()
41        },
42    ) {
43        Ok(rt) => rt,
44        Err(e) => {
45            eprintln!("runtime: {e}");
46            std::process::exit(1);
47        }
48    };
49    let Some(worker_fn) = rt.function_index("worker") else {
50        eprintln!("missing worker");
51        std::process::exit(1);
52    };
53
54    let start = Instant::now();
55    let mut handles = Vec::with_capacity(n as usize);
56    for _ in 0..n {
57        match rt.spawn(worker_fn, &[]) {
58            Ok(h) => handles.push(h),
59            Err(e) => {
60                eprintln!("spawn: {e}");
61                std::process::exit(1);
62            }
63        }
64    }
65    let mut ok = 0u32;
66    for h in handles {
67        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
68            ok += 1;
69        }
70    }
71    let elapsed = start.elapsed();
72    let metrics = rt.metrics();
73    rt.shutdown();
74
75    let secs = elapsed.as_secs_f64().max(1e-9);
76    println!("processes={ok}/{n}");
77    println!("workers={workers}");
78    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
79    println!("spawns_per_sec={:.0}", ok as f64 / secs);
80    println!("{metrics}");
81}
Source

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)
15fn main() {
16    let rt = match Runtime::with_config(
17        samples::boom(),
18        RuntimeConfig {
19            workers: 1,
20            quantum: 1_000,
21            mailbox: byteflow::MailboxConfig::DEFAULT,
22            ..Default::default()
23        },
24    ) {
25        Ok(rt) => rt,
26        Err(e) => {
27            eprintln!("runtime: {e}");
28            std::process::exit(1);
29        }
30    };
31    let Some(boom) = rt.function_index("boom") else {
32        eprintln!("missing boom");
33        std::process::exit(1);
34    };
35    let sup = match Supervisor::with_config(
36        rt.spawner(),
37        SupervisorConfig {
38            max_restarts: 2,
39            max_period: Duration::from_secs(5),
40        },
41    ) {
42        Ok(s) => s,
43        Err(e) => {
44            eprintln!("supervisor: {e}");
45            std::process::exit(1);
46        }
47    };
48    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
49    {
50        eprintln!("start_child: {e}");
51        std::process::exit(1);
52    }
53
54    let start = std::time::Instant::now();
55    while !sup.intensity_exceeded() {
56        if start.elapsed() > Duration::from_secs(2) {
57            eprintln!("supervisor did not hit intensity in time");
58            std::process::exit(1);
59        }
60        thread::sleep(Duration::from_millis(5));
61    }
62
63    let metrics = rt.metrics();
64    println!("intensity exceeded after {} failures", metrics.processes_failed);
65    println!("{metrics}");
66    sup.shutdown();
67    rt.shutdown();
68}
Source

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

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

Source

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)
9fn main() {
10    let chunk = samples::ping_pong();
11    let rt = match Runtime::with_natives_and_config(
12        chunk,
13        std_native_table(),
14        RuntimeConfig {
15            workers: 1,
16            quantum: 10_000,
17            mailbox: byteflow::MailboxConfig::DEFAULT,
18            ..Default::default()
19        },
20    ) {
21        Ok(rt) => rt,
22        Err(e) => {
23            eprintln!("runtime: {e}");
24            std::process::exit(1);
25        }
26    };
27    let Some(main) = rt.function_index("main") else {
28        eprintln!("missing main");
29        std::process::exit(1);
30    };
31    let handle = match rt.spawn(main, &[]) {
32        Ok(h) => h,
33        Err(e) => {
34            eprintln!("spawn: {e}");
35            std::process::exit(1);
36        }
37    };
38    let outcome = handle.join();
39    let metrics = rt.metrics();
40    rt.shutdown();
41
42    match outcome {
43        FlowOutcome::Completed(Value::Int(2)) => {
44            println!("pong replied 2 (Atomic Hop)");
45            println!("{metrics}");
46        }
47        other => {
48            eprintln!("unexpected {other:?}");
49            std::process::exit(1);
50        }
51    }
52}
More examples
Hide additional examples
examples/atomic_actors.rs (line 39)
21fn main() {
22    let chunk = samples::atomic_request_reply();
23    let rt = match Runtime::with_natives_and_config(
24        chunk,
25        std_native_table(),
26        RuntimeConfig {
27            workers: 2,
28            quantum: 10_000,
29            mailbox: byteflow::MailboxConfig::DEFAULT,
30            ..Default::default()
31        },
32    ) {
33        Ok(rt) => rt,
34        Err(e) => {
35            eprintln!("runtime: {e}");
36            std::process::exit(1);
37        }
38    };
39    let Some(main) = rt.function_index("main") else {
40        eprintln!("missing main");
41        std::process::exit(1);
42    };
43    let handle = match rt.spawn(main, &[]) {
44        Ok(h) => h,
45        Err(e) => {
46            eprintln!("spawn: {e}");
47            std::process::exit(1);
48        }
49    };
50    let outcome = handle.join();
51    let metrics = rt.metrics();
52    rt.shutdown();
53
54    match outcome {
55        FlowOutcome::Completed(Value::Int(42)) => {
56            println!("atomic request-reply ok: payload=42");
57            println!("{metrics}");
58        }
59        other => {
60            eprintln!("unexpected {other:?}");
61            std::process::exit(1);
62        }
63    }
64}
examples/crash_and_restart.rs (line 31)
15fn main() {
16    let rt = match Runtime::with_config(
17        samples::boom(),
18        RuntimeConfig {
19            workers: 1,
20            quantum: 1_000,
21            mailbox: byteflow::MailboxConfig::DEFAULT,
22            ..Default::default()
23        },
24    ) {
25        Ok(rt) => rt,
26        Err(e) => {
27            eprintln!("runtime: {e}");
28            std::process::exit(1);
29        }
30    };
31    let Some(boom) = rt.function_index("boom") else {
32        eprintln!("missing boom");
33        std::process::exit(1);
34    };
35    let sup = match Supervisor::with_config(
36        rt.spawner(),
37        SupervisorConfig {
38            max_restarts: 2,
39            max_period: Duration::from_secs(5),
40        },
41    ) {
42        Ok(s) => s,
43        Err(e) => {
44            eprintln!("supervisor: {e}");
45            std::process::exit(1);
46        }
47    };
48    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
49    {
50        eprintln!("start_child: {e}");
51        std::process::exit(1);
52    }
53
54    let start = std::time::Instant::now();
55    while !sup.intensity_exceeded() {
56        if start.elapsed() > Duration::from_secs(2) {
57            eprintln!("supervisor did not hit intensity in time");
58            std::process::exit(1);
59        }
60        thread::sleep(Duration::from_millis(5));
61    }
62
63    let metrics = rt.metrics();
64    println!("intensity exceeded after {} failures", metrics.processes_failed);
65    println!("{metrics}");
66    sup.shutdown();
67    rt.shutdown();
68}
examples/throughput.rs (line 49)
20fn main() {
21    let n: u32 = match std::env::args().nth(1) {
22        Some(s) => match s.parse() {
23            Ok(v) => v,
24            Err(_) => 50_000,
25        },
26        None => 50_000,
27    };
28
29    let workers = match std::thread::available_parallelism() {
30        Ok(p) => p.get(),
31        Err(_) => 1,
32    };
33
34    let rt = match Runtime::with_config(
35        trivial_chunk(),
36        RuntimeConfig {
37            workers,
38            quantum: 10_000,
39            mailbox: byteflow::MailboxConfig::DEFAULT,
40            ..Default::default()
41        },
42    ) {
43        Ok(rt) => rt,
44        Err(e) => {
45            eprintln!("runtime: {e}");
46            std::process::exit(1);
47        }
48    };
49    let Some(worker_fn) = rt.function_index("worker") else {
50        eprintln!("missing worker");
51        std::process::exit(1);
52    };
53
54    let start = Instant::now();
55    let mut handles = Vec::with_capacity(n as usize);
56    for _ in 0..n {
57        match rt.spawn(worker_fn, &[]) {
58            Ok(h) => handles.push(h),
59            Err(e) => {
60                eprintln!("spawn: {e}");
61                std::process::exit(1);
62            }
63        }
64    }
65    let mut ok = 0u32;
66    for h in handles {
67        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
68            ok += 1;
69        }
70    }
71    let elapsed = start.elapsed();
72    let metrics = rt.metrics();
73    rt.shutdown();
74
75    let secs = elapsed.as_secs_f64().max(1e-9);
76    println!("processes={ok}/{n}");
77    println!("workers={workers}");
78    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
79    println!("spawns_per_sec={:.0}", ok as f64 / secs);
80    println!("{metrics}");
81}
Source

pub fn metrics(&self) -> RuntimeMetricsSnapshot

Examples found in repository?
examples/jit_loop.rs (line 38)
22fn run_once(jit: bool) -> Result<(), Box<dyn std::error::Error>> {
23    let chunk = loop_chunk(1_000_000);
24    let rt = Runtime::with_config(
25        chunk,
26        RuntimeConfig {
27            workers: 1,
28            quantum: 50_000_000,
29            jit: JitConfig {
30                enabled: jit,
31                hot_threshold: 1,
32            },
33            ..Default::default()
34        },
35    )?;
36    let start = Instant::now();
37    let outcome = rt.spawn(0, &[])?.join();
38    let snapshot = rt.metrics();
39    rt.shutdown();
40    let elapsed = start.elapsed();
41    match outcome {
42        byteflow::FlowOutcome::Completed(Value::Int(n)) => {
43            println!("jit={jit} result={n} elapsed={elapsed:?} metrics={snapshot}");
44            Ok(())
45        }
46        other => Err(format!("unexpected outcome: {other:?}").into()),
47    }
48}
More examples
Hide additional examples
examples/ping_pong.rs (line 39)
9fn main() {
10    let chunk = samples::ping_pong();
11    let rt = match Runtime::with_natives_and_config(
12        chunk,
13        std_native_table(),
14        RuntimeConfig {
15            workers: 1,
16            quantum: 10_000,
17            mailbox: byteflow::MailboxConfig::DEFAULT,
18            ..Default::default()
19        },
20    ) {
21        Ok(rt) => rt,
22        Err(e) => {
23            eprintln!("runtime: {e}");
24            std::process::exit(1);
25        }
26    };
27    let Some(main) = rt.function_index("main") else {
28        eprintln!("missing main");
29        std::process::exit(1);
30    };
31    let handle = match rt.spawn(main, &[]) {
32        Ok(h) => h,
33        Err(e) => {
34            eprintln!("spawn: {e}");
35            std::process::exit(1);
36        }
37    };
38    let outcome = handle.join();
39    let metrics = rt.metrics();
40    rt.shutdown();
41
42    match outcome {
43        FlowOutcome::Completed(Value::Int(2)) => {
44            println!("pong replied 2 (Atomic Hop)");
45            println!("{metrics}");
46        }
47        other => {
48            eprintln!("unexpected {other:?}");
49            std::process::exit(1);
50        }
51    }
52}
examples/atomic_actors.rs (line 51)
21fn main() {
22    let chunk = samples::atomic_request_reply();
23    let rt = match Runtime::with_natives_and_config(
24        chunk,
25        std_native_table(),
26        RuntimeConfig {
27            workers: 2,
28            quantum: 10_000,
29            mailbox: byteflow::MailboxConfig::DEFAULT,
30            ..Default::default()
31        },
32    ) {
33        Ok(rt) => rt,
34        Err(e) => {
35            eprintln!("runtime: {e}");
36            std::process::exit(1);
37        }
38    };
39    let Some(main) = rt.function_index("main") else {
40        eprintln!("missing main");
41        std::process::exit(1);
42    };
43    let handle = match rt.spawn(main, &[]) {
44        Ok(h) => h,
45        Err(e) => {
46            eprintln!("spawn: {e}");
47            std::process::exit(1);
48        }
49    };
50    let outcome = handle.join();
51    let metrics = rt.metrics();
52    rt.shutdown();
53
54    match outcome {
55        FlowOutcome::Completed(Value::Int(42)) => {
56            println!("atomic request-reply ok: payload=42");
57            println!("{metrics}");
58        }
59        other => {
60            eprintln!("unexpected {other:?}");
61            std::process::exit(1);
62        }
63    }
64}
examples/crash_and_restart.rs (line 63)
15fn main() {
16    let rt = match Runtime::with_config(
17        samples::boom(),
18        RuntimeConfig {
19            workers: 1,
20            quantum: 1_000,
21            mailbox: byteflow::MailboxConfig::DEFAULT,
22            ..Default::default()
23        },
24    ) {
25        Ok(rt) => rt,
26        Err(e) => {
27            eprintln!("runtime: {e}");
28            std::process::exit(1);
29        }
30    };
31    let Some(boom) = rt.function_index("boom") else {
32        eprintln!("missing boom");
33        std::process::exit(1);
34    };
35    let sup = match Supervisor::with_config(
36        rt.spawner(),
37        SupervisorConfig {
38            max_restarts: 2,
39            max_period: Duration::from_secs(5),
40        },
41    ) {
42        Ok(s) => s,
43        Err(e) => {
44            eprintln!("supervisor: {e}");
45            std::process::exit(1);
46        }
47    };
48    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
49    {
50        eprintln!("start_child: {e}");
51        std::process::exit(1);
52    }
53
54    let start = std::time::Instant::now();
55    while !sup.intensity_exceeded() {
56        if start.elapsed() > Duration::from_secs(2) {
57            eprintln!("supervisor did not hit intensity in time");
58            std::process::exit(1);
59        }
60        thread::sleep(Duration::from_millis(5));
61    }
62
63    let metrics = rt.metrics();
64    println!("intensity exceeded after {} failures", metrics.processes_failed);
65    println!("{metrics}");
66    sup.shutdown();
67    rt.shutdown();
68}
examples/throughput.rs (line 72)
20fn main() {
21    let n: u32 = match std::env::args().nth(1) {
22        Some(s) => match s.parse() {
23            Ok(v) => v,
24            Err(_) => 50_000,
25        },
26        None => 50_000,
27    };
28
29    let workers = match std::thread::available_parallelism() {
30        Ok(p) => p.get(),
31        Err(_) => 1,
32    };
33
34    let rt = match Runtime::with_config(
35        trivial_chunk(),
36        RuntimeConfig {
37            workers,
38            quantum: 10_000,
39            mailbox: byteflow::MailboxConfig::DEFAULT,
40            ..Default::default()
41        },
42    ) {
43        Ok(rt) => rt,
44        Err(e) => {
45            eprintln!("runtime: {e}");
46            std::process::exit(1);
47        }
48    };
49    let Some(worker_fn) = rt.function_index("worker") else {
50        eprintln!("missing worker");
51        std::process::exit(1);
52    };
53
54    let start = Instant::now();
55    let mut handles = Vec::with_capacity(n as usize);
56    for _ in 0..n {
57        match rt.spawn(worker_fn, &[]) {
58            Ok(h) => handles.push(h),
59            Err(e) => {
60                eprintln!("spawn: {e}");
61                std::process::exit(1);
62            }
63        }
64    }
65    let mut ok = 0u32;
66    for h in handles {
67        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
68            ok += 1;
69        }
70    }
71    let elapsed = start.elapsed();
72    let metrics = rt.metrics();
73    rt.shutdown();
74
75    let secs = elapsed.as_secs_f64().max(1e-9);
76    println!("processes={ok}/{n}");
77    println!("workers={workers}");
78    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
79    println!("spawns_per_sec={:.0}", ok as f64 / secs);
80    println!("{metrics}");
81}
Source

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.

Source

pub fn worker_count(&self) -> usize

Source

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

Source

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

Replace the runtime bytecode image and invalidate any compiled JIT traces.

Source

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)
22fn run_once(jit: bool) -> Result<(), Box<dyn std::error::Error>> {
23    let chunk = loop_chunk(1_000_000);
24    let rt = Runtime::with_config(
25        chunk,
26        RuntimeConfig {
27            workers: 1,
28            quantum: 50_000_000,
29            jit: JitConfig {
30                enabled: jit,
31                hot_threshold: 1,
32            },
33            ..Default::default()
34        },
35    )?;
36    let start = Instant::now();
37    let outcome = rt.spawn(0, &[])?.join();
38    let snapshot = rt.metrics();
39    rt.shutdown();
40    let elapsed = start.elapsed();
41    match outcome {
42        byteflow::FlowOutcome::Completed(Value::Int(n)) => {
43            println!("jit={jit} result={n} elapsed={elapsed:?} metrics={snapshot}");
44            Ok(())
45        }
46        other => Err(format!("unexpected outcome: {other:?}").into()),
47    }
48}
More examples
Hide additional examples
examples/ping_pong.rs (line 40)
9fn main() {
10    let chunk = samples::ping_pong();
11    let rt = match Runtime::with_natives_and_config(
12        chunk,
13        std_native_table(),
14        RuntimeConfig {
15            workers: 1,
16            quantum: 10_000,
17            mailbox: byteflow::MailboxConfig::DEFAULT,
18            ..Default::default()
19        },
20    ) {
21        Ok(rt) => rt,
22        Err(e) => {
23            eprintln!("runtime: {e}");
24            std::process::exit(1);
25        }
26    };
27    let Some(main) = rt.function_index("main") else {
28        eprintln!("missing main");
29        std::process::exit(1);
30    };
31    let handle = match rt.spawn(main, &[]) {
32        Ok(h) => h,
33        Err(e) => {
34            eprintln!("spawn: {e}");
35            std::process::exit(1);
36        }
37    };
38    let outcome = handle.join();
39    let metrics = rt.metrics();
40    rt.shutdown();
41
42    match outcome {
43        FlowOutcome::Completed(Value::Int(2)) => {
44            println!("pong replied 2 (Atomic Hop)");
45            println!("{metrics}");
46        }
47        other => {
48            eprintln!("unexpected {other:?}");
49            std::process::exit(1);
50        }
51    }
52}
examples/atomic_actors.rs (line 52)
21fn main() {
22    let chunk = samples::atomic_request_reply();
23    let rt = match Runtime::with_natives_and_config(
24        chunk,
25        std_native_table(),
26        RuntimeConfig {
27            workers: 2,
28            quantum: 10_000,
29            mailbox: byteflow::MailboxConfig::DEFAULT,
30            ..Default::default()
31        },
32    ) {
33        Ok(rt) => rt,
34        Err(e) => {
35            eprintln!("runtime: {e}");
36            std::process::exit(1);
37        }
38    };
39    let Some(main) = rt.function_index("main") else {
40        eprintln!("missing main");
41        std::process::exit(1);
42    };
43    let handle = match rt.spawn(main, &[]) {
44        Ok(h) => h,
45        Err(e) => {
46            eprintln!("spawn: {e}");
47            std::process::exit(1);
48        }
49    };
50    let outcome = handle.join();
51    let metrics = rt.metrics();
52    rt.shutdown();
53
54    match outcome {
55        FlowOutcome::Completed(Value::Int(42)) => {
56            println!("atomic request-reply ok: payload=42");
57            println!("{metrics}");
58        }
59        other => {
60            eprintln!("unexpected {other:?}");
61            std::process::exit(1);
62        }
63    }
64}
examples/crash_and_restart.rs (line 67)
15fn main() {
16    let rt = match Runtime::with_config(
17        samples::boom(),
18        RuntimeConfig {
19            workers: 1,
20            quantum: 1_000,
21            mailbox: byteflow::MailboxConfig::DEFAULT,
22            ..Default::default()
23        },
24    ) {
25        Ok(rt) => rt,
26        Err(e) => {
27            eprintln!("runtime: {e}");
28            std::process::exit(1);
29        }
30    };
31    let Some(boom) = rt.function_index("boom") else {
32        eprintln!("missing boom");
33        std::process::exit(1);
34    };
35    let sup = match Supervisor::with_config(
36        rt.spawner(),
37        SupervisorConfig {
38            max_restarts: 2,
39            max_period: Duration::from_secs(5),
40        },
41    ) {
42        Ok(s) => s,
43        Err(e) => {
44            eprintln!("supervisor: {e}");
45            std::process::exit(1);
46        }
47    };
48    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
49    {
50        eprintln!("start_child: {e}");
51        std::process::exit(1);
52    }
53
54    let start = std::time::Instant::now();
55    while !sup.intensity_exceeded() {
56        if start.elapsed() > Duration::from_secs(2) {
57            eprintln!("supervisor did not hit intensity in time");
58            std::process::exit(1);
59        }
60        thread::sleep(Duration::from_millis(5));
61    }
62
63    let metrics = rt.metrics();
64    println!("intensity exceeded after {} failures", metrics.processes_failed);
65    println!("{metrics}");
66    sup.shutdown();
67    rt.shutdown();
68}
examples/throughput.rs (line 73)
20fn main() {
21    let n: u32 = match std::env::args().nth(1) {
22        Some(s) => match s.parse() {
23            Ok(v) => v,
24            Err(_) => 50_000,
25        },
26        None => 50_000,
27    };
28
29    let workers = match std::thread::available_parallelism() {
30        Ok(p) => p.get(),
31        Err(_) => 1,
32    };
33
34    let rt = match Runtime::with_config(
35        trivial_chunk(),
36        RuntimeConfig {
37            workers,
38            quantum: 10_000,
39            mailbox: byteflow::MailboxConfig::DEFAULT,
40            ..Default::default()
41        },
42    ) {
43        Ok(rt) => rt,
44        Err(e) => {
45            eprintln!("runtime: {e}");
46            std::process::exit(1);
47        }
48    };
49    let Some(worker_fn) = rt.function_index("worker") else {
50        eprintln!("missing worker");
51        std::process::exit(1);
52    };
53
54    let start = Instant::now();
55    let mut handles = Vec::with_capacity(n as usize);
56    for _ in 0..n {
57        match rt.spawn(worker_fn, &[]) {
58            Ok(h) => handles.push(h),
59            Err(e) => {
60                eprintln!("spawn: {e}");
61                std::process::exit(1);
62            }
63        }
64    }
65    let mut ok = 0u32;
66    for h in handles {
67        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
68            ok += 1;
69        }
70    }
71    let elapsed = start.elapsed();
72    let metrics = rt.metrics();
73    rt.shutdown();
74
75    let secs = elapsed.as_secs_f64().max(1e-9);
76    println!("processes={ok}/{n}");
77    println!("workers={workers}");
78    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
79    println!("spawns_per_sec={:.0}", ok as f64 / secs);
80    println!("{metrics}");
81}

Auto Trait Implementations§

Blanket Implementations§

Source§

impl<T> Any for T
where T: 'static + ?Sized,

Source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
Source§

impl<T> Borrow<T> for T
where T: ?Sized,

Source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
Source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

Source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
Source§

impl<T> From<T> for T

Source§

fn from(t: T) -> T

Returns the argument unchanged.

Source§

impl<T, U> Into<U> for T
where U: From<T>,

Source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

Source§

impl<T> Pointable for T

Source§

const ALIGN: usize

The alignment of pointer.
Source§

type Init = T

The type for initializers.
Source§

unsafe fn init(init: <T as Pointable>::Init) -> usize

Initializes a with the given initializer. Read more
Source§

unsafe fn deref<'a>(ptr: usize) -> &'a T

Dereferences the given pointer. Read more
Source§

unsafe fn deref_mut<'a>(ptr: usize) -> &'a mut T

Mutably dereferences the given pointer. Read more
Source§

unsafe fn drop(ptr: usize)

Drops the object pointed to by the given pointer. Read more
Source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

Source§

type Error = !

The type returned in the event of a conversion error.
Source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

Performs the conversion.
Source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

Source§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
Source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.