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). Intentionally not here yet: JIT, native quotas, distribution across machines.

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/crash_and_restart.rs (lines 16-23)

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

More examples

Hide additional examples

examples/throughput.rs (lines 33-40)

19fn main() {
20    let n: u32 = match std::env::args().nth(1) {
21        Some(s) => match s.parse() {
22            Ok(v) => v,
23            Err(_) => 50_000,
24        },
25        None => 50_000,
26    };
27
28    let workers = match std::thread::available_parallelism() {
29        Ok(p) => p.get(),
30        Err(_) => 1,
31    };
32
33    let rt = match Runtime::with_config(
34        trivial_chunk(),
35        RuntimeConfig {
36            workers,
37            quantum: 10_000,
38            mailbox: byteflow::MailboxConfig::DEFAULT,
39        },
40    ) {
41        Ok(rt) => rt,
42        Err(e) => {
43            eprintln!("runtime: {e}");
44            std::process::exit(1);
45        }
46    };
47    let Some(worker_fn) = rt.function_index("worker") else {
48        eprintln!("missing worker");
49        std::process::exit(1);
50    };
51
52    let start = Instant::now();
53    let mut handles = Vec::with_capacity(n as usize);
54    for _ in 0..n {
55        match rt.spawn(worker_fn, &[]) {
56            Ok(h) => handles.push(h),
57            Err(e) => {
58                eprintln!("spawn: {e}");
59                std::process::exit(1);
60            }
61        }
62    }
63    let mut ok = 0u32;
64    for h in handles {
65        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
66            ok += 1;
67        }
68    }
69    let elapsed = start.elapsed();
70    let metrics = rt.metrics();
71    rt.shutdown();
72
73    let secs = elapsed.as_secs_f64().max(1e-9);
74    println!("processes={ok}/{n}");
75    println!("workers={workers}");
76    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
77    println!("spawns_per_sec={:.0}", ok as f64 / secs);
78    println!("{metrics}");
79}

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

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        },
19    ) {
20        Ok(rt) => rt,
21        Err(e) => {
22            eprintln!("runtime: {e}");
23            std::process::exit(1);
24        }
25    };
26    let Some(main) = rt.function_index("main") else {
27        eprintln!("missing main");
28        std::process::exit(1);
29    };
30    let handle = match rt.spawn(main, &[]) {
31        Ok(h) => h,
32        Err(e) => {
33            eprintln!("spawn: {e}");
34            std::process::exit(1);
35        }
36    };
37    let outcome = handle.join();
38    let metrics = rt.metrics();
39    rt.shutdown();
40
41    match outcome {
42        FlowOutcome::Completed(Value::Int(2)) => {
43            println!("pong replied 2 (Atomic Hop)");
44            println!("{metrics}");
45        }
46        other => {
47            eprintln!("unexpected {other:?}");
48            std::process::exit(1);
49        }
50    }
51}

More examples

Hide additional examples

examples/atomic_actors.rs (lines 23-31)

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        },
31    ) {
32        Ok(rt) => rt,
33        Err(e) => {
34            eprintln!("runtime: {e}");
35            std::process::exit(1);
36        }
37    };
38    let Some(main) = rt.function_index("main") else {
39        eprintln!("missing main");
40        std::process::exit(1);
41    };
42    let handle = match rt.spawn(main, &[]) {
43        Ok(h) => h,
44        Err(e) => {
45            eprintln!("spawn: {e}");
46            std::process::exit(1);
47        }
48    };
49    let outcome = handle.join();
50    let metrics = rt.metrics();
51    rt.shutdown();
52
53    match outcome {
54        FlowOutcome::Completed(Value::Int(42)) => {
55            println!("atomic request-reply ok: payload=42");
56            println!("{metrics}");
57        }
58        other => {
59            eprintln!("unexpected {other:?}");
60            std::process::exit(1);
61        }
62    }
63}

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/ping_pong.rs (line 30)

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        },
19    ) {
20        Ok(rt) => rt,
21        Err(e) => {
22            eprintln!("runtime: {e}");
23            std::process::exit(1);
24        }
25    };
26    let Some(main) = rt.function_index("main") else {
27        eprintln!("missing main");
28        std::process::exit(1);
29    };
30    let handle = match rt.spawn(main, &[]) {
31        Ok(h) => h,
32        Err(e) => {
33            eprintln!("spawn: {e}");
34            std::process::exit(1);
35        }
36    };
37    let outcome = handle.join();
38    let metrics = rt.metrics();
39    rt.shutdown();
40
41    match outcome {
42        FlowOutcome::Completed(Value::Int(2)) => {
43            println!("pong replied 2 (Atomic Hop)");
44            println!("{metrics}");
45        }
46        other => {
47            eprintln!("unexpected {other:?}");
48            std::process::exit(1);
49        }
50    }
51}

More examples

Hide additional examples

examples/atomic_actors.rs (line 42)

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        },
31    ) {
32        Ok(rt) => rt,
33        Err(e) => {
34            eprintln!("runtime: {e}");
35            std::process::exit(1);
36        }
37    };
38    let Some(main) = rt.function_index("main") else {
39        eprintln!("missing main");
40        std::process::exit(1);
41    };
42    let handle = match rt.spawn(main, &[]) {
43        Ok(h) => h,
44        Err(e) => {
45            eprintln!("spawn: {e}");
46            std::process::exit(1);
47        }
48    };
49    let outcome = handle.join();
50    let metrics = rt.metrics();
51    rt.shutdown();
52
53    match outcome {
54        FlowOutcome::Completed(Value::Int(42)) => {
55            println!("atomic request-reply ok: payload=42");
56            println!("{metrics}");
57        }
58        other => {
59            eprintln!("unexpected {other:?}");
60            std::process::exit(1);
61        }
62    }
63}

examples/throughput.rs (line 55)

19fn main() {
20    let n: u32 = match std::env::args().nth(1) {
21        Some(s) => match s.parse() {
22            Ok(v) => v,
23            Err(_) => 50_000,
24        },
25        None => 50_000,
26    };
27
28    let workers = match std::thread::available_parallelism() {
29        Ok(p) => p.get(),
30        Err(_) => 1,
31    };
32
33    let rt = match Runtime::with_config(
34        trivial_chunk(),
35        RuntimeConfig {
36            workers,
37            quantum: 10_000,
38            mailbox: byteflow::MailboxConfig::DEFAULT,
39        },
40    ) {
41        Ok(rt) => rt,
42        Err(e) => {
43            eprintln!("runtime: {e}");
44            std::process::exit(1);
45        }
46    };
47    let Some(worker_fn) = rt.function_index("worker") else {
48        eprintln!("missing worker");
49        std::process::exit(1);
50    };
51
52    let start = Instant::now();
53    let mut handles = Vec::with_capacity(n as usize);
54    for _ in 0..n {
55        match rt.spawn(worker_fn, &[]) {
56            Ok(h) => handles.push(h),
57            Err(e) => {
58                eprintln!("spawn: {e}");
59                std::process::exit(1);
60            }
61        }
62    }
63    let mut ok = 0u32;
64    for h in handles {
65        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
66            ok += 1;
67        }
68    }
69    let elapsed = start.elapsed();
70    let metrics = rt.metrics();
71    rt.shutdown();
72
73    let secs = elapsed.as_secs_f64().max(1e-9);
74    println!("processes={ok}/{n}");
75    println!("workers={workers}");
76    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
77    println!("spawns_per_sec={:.0}", ok as f64 / secs);
78    println!("{metrics}");
79}

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

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

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::bytecode::ChunkBuilder and don’t want to thread raw indices through their own code.

Examples found in repository ?

examples/ping_pong.rs (line 26)

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        },
19    ) {
20        Ok(rt) => rt,
21        Err(e) => {
22            eprintln!("runtime: {e}");
23            std::process::exit(1);
24        }
25    };
26    let Some(main) = rt.function_index("main") else {
27        eprintln!("missing main");
28        std::process::exit(1);
29    };
30    let handle = match rt.spawn(main, &[]) {
31        Ok(h) => h,
32        Err(e) => {
33            eprintln!("spawn: {e}");
34            std::process::exit(1);
35        }
36    };
37    let outcome = handle.join();
38    let metrics = rt.metrics();
39    rt.shutdown();
40
41    match outcome {
42        FlowOutcome::Completed(Value::Int(2)) => {
43            println!("pong replied 2 (Atomic Hop)");
44            println!("{metrics}");
45        }
46        other => {
47            eprintln!("unexpected {other:?}");
48            std::process::exit(1);
49        }
50    }
51}

More examples

Hide additional examples

examples/atomic_actors.rs (line 38)

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        },
31    ) {
32        Ok(rt) => rt,
33        Err(e) => {
34            eprintln!("runtime: {e}");
35            std::process::exit(1);
36        }
37    };
38    let Some(main) = rt.function_index("main") else {
39        eprintln!("missing main");
40        std::process::exit(1);
41    };
42    let handle = match rt.spawn(main, &[]) {
43        Ok(h) => h,
44        Err(e) => {
45            eprintln!("spawn: {e}");
46            std::process::exit(1);
47        }
48    };
49    let outcome = handle.join();
50    let metrics = rt.metrics();
51    rt.shutdown();
52
53    match outcome {
54        FlowOutcome::Completed(Value::Int(42)) => {
55            println!("atomic request-reply ok: payload=42");
56            println!("{metrics}");
57        }
58        other => {
59            eprintln!("unexpected {other:?}");
60            std::process::exit(1);
61        }
62    }
63}

examples/crash_and_restart.rs (line 30)

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

examples/throughput.rs (line 47)

19fn main() {
20    let n: u32 = match std::env::args().nth(1) {
21        Some(s) => match s.parse() {
22            Ok(v) => v,
23            Err(_) => 50_000,
24        },
25        None => 50_000,
26    };
27
28    let workers = match std::thread::available_parallelism() {
29        Ok(p) => p.get(),
30        Err(_) => 1,
31    };
32
33    let rt = match Runtime::with_config(
34        trivial_chunk(),
35        RuntimeConfig {
36            workers,
37            quantum: 10_000,
38            mailbox: byteflow::MailboxConfig::DEFAULT,
39        },
40    ) {
41        Ok(rt) => rt,
42        Err(e) => {
43            eprintln!("runtime: {e}");
44            std::process::exit(1);
45        }
46    };
47    let Some(worker_fn) = rt.function_index("worker") else {
48        eprintln!("missing worker");
49        std::process::exit(1);
50    };
51
52    let start = Instant::now();
53    let mut handles = Vec::with_capacity(n as usize);
54    for _ in 0..n {
55        match rt.spawn(worker_fn, &[]) {
56            Ok(h) => handles.push(h),
57            Err(e) => {
58                eprintln!("spawn: {e}");
59                std::process::exit(1);
60            }
61        }
62    }
63    let mut ok = 0u32;
64    for h in handles {
65        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
66            ok += 1;
67        }
68    }
69    let elapsed = start.elapsed();
70    let metrics = rt.metrics();
71    rt.shutdown();
72
73    let secs = elapsed.as_secs_f64().max(1e-9);
74    println!("processes={ok}/{n}");
75    println!("workers={workers}");
76    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
77    println!("spawns_per_sec={:.0}", ok as f64 / secs);
78    println!("{metrics}");
79}

Source

pub fn metrics(&self) -> RuntimeMetricsSnapshot

Examples found in repository ?

examples/ping_pong.rs (line 38)

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        },
19    ) {
20        Ok(rt) => rt,
21        Err(e) => {
22            eprintln!("runtime: {e}");
23            std::process::exit(1);
24        }
25    };
26    let Some(main) = rt.function_index("main") else {
27        eprintln!("missing main");
28        std::process::exit(1);
29    };
30    let handle = match rt.spawn(main, &[]) {
31        Ok(h) => h,
32        Err(e) => {
33            eprintln!("spawn: {e}");
34            std::process::exit(1);
35        }
36    };
37    let outcome = handle.join();
38    let metrics = rt.metrics();
39    rt.shutdown();
40
41    match outcome {
42        FlowOutcome::Completed(Value::Int(2)) => {
43            println!("pong replied 2 (Atomic Hop)");
44            println!("{metrics}");
45        }
46        other => {
47            eprintln!("unexpected {other:?}");
48            std::process::exit(1);
49        }
50    }
51}

More examples

Hide additional examples

examples/atomic_actors.rs (line 50)

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        },
31    ) {
32        Ok(rt) => rt,
33        Err(e) => {
34            eprintln!("runtime: {e}");
35            std::process::exit(1);
36        }
37    };
38    let Some(main) = rt.function_index("main") else {
39        eprintln!("missing main");
40        std::process::exit(1);
41    };
42    let handle = match rt.spawn(main, &[]) {
43        Ok(h) => h,
44        Err(e) => {
45            eprintln!("spawn: {e}");
46            std::process::exit(1);
47        }
48    };
49    let outcome = handle.join();
50    let metrics = rt.metrics();
51    rt.shutdown();
52
53    match outcome {
54        FlowOutcome::Completed(Value::Int(42)) => {
55            println!("atomic request-reply ok: payload=42");
56            println!("{metrics}");
57        }
58        other => {
59            eprintln!("unexpected {other:?}");
60            std::process::exit(1);
61        }
62    }
63}

examples/crash_and_restart.rs (line 62)

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

examples/throughput.rs (line 70)

19fn main() {
20    let n: u32 = match std::env::args().nth(1) {
21        Some(s) => match s.parse() {
22            Ok(v) => v,
23            Err(_) => 50_000,
24        },
25        None => 50_000,
26    };
27
28    let workers = match std::thread::available_parallelism() {
29        Ok(p) => p.get(),
30        Err(_) => 1,
31    };
32
33    let rt = match Runtime::with_config(
34        trivial_chunk(),
35        RuntimeConfig {
36            workers,
37            quantum: 10_000,
38            mailbox: byteflow::MailboxConfig::DEFAULT,
39        },
40    ) {
41        Ok(rt) => rt,
42        Err(e) => {
43            eprintln!("runtime: {e}");
44            std::process::exit(1);
45        }
46    };
47    let Some(worker_fn) = rt.function_index("worker") else {
48        eprintln!("missing worker");
49        std::process::exit(1);
50    };
51
52    let start = Instant::now();
53    let mut handles = Vec::with_capacity(n as usize);
54    for _ in 0..n {
55        match rt.spawn(worker_fn, &[]) {
56            Ok(h) => handles.push(h),
57            Err(e) => {
58                eprintln!("spawn: {e}");
59                std::process::exit(1);
60            }
61        }
62    }
63    let mut ok = 0u32;
64    for h in handles {
65        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
66            ok += 1;
67        }
68    }
69    let elapsed = start.elapsed();
70    let metrics = rt.metrics();
71    rt.shutdown();
72
73    let secs = elapsed.as_secs_f64().max(1e-9);
74    println!("processes={ok}/{n}");
75    println!("workers={workers}");
76    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
77    println!("spawns_per_sec={:.0}", ok as f64 / secs);
78    println!("{metrics}");
79}

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 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/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        },
19    ) {
20        Ok(rt) => rt,
21        Err(e) => {
22            eprintln!("runtime: {e}");
23            std::process::exit(1);
24        }
25    };
26    let Some(main) = rt.function_index("main") else {
27        eprintln!("missing main");
28        std::process::exit(1);
29    };
30    let handle = match rt.spawn(main, &[]) {
31        Ok(h) => h,
32        Err(e) => {
33            eprintln!("spawn: {e}");
34            std::process::exit(1);
35        }
36    };
37    let outcome = handle.join();
38    let metrics = rt.metrics();
39    rt.shutdown();
40
41    match outcome {
42        FlowOutcome::Completed(Value::Int(2)) => {
43            println!("pong replied 2 (Atomic Hop)");
44            println!("{metrics}");
45        }
46        other => {
47            eprintln!("unexpected {other:?}");
48            std::process::exit(1);
49        }
50    }
51}

More examples

Hide additional examples

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        },
31    ) {
32        Ok(rt) => rt,
33        Err(e) => {
34            eprintln!("runtime: {e}");
35            std::process::exit(1);
36        }
37    };
38    let Some(main) = rt.function_index("main") else {
39        eprintln!("missing main");
40        std::process::exit(1);
41    };
42    let handle = match rt.spawn(main, &[]) {
43        Ok(h) => h,
44        Err(e) => {
45            eprintln!("spawn: {e}");
46            std::process::exit(1);
47        }
48    };
49    let outcome = handle.join();
50    let metrics = rt.metrics();
51    rt.shutdown();
52
53    match outcome {
54        FlowOutcome::Completed(Value::Int(42)) => {
55            println!("atomic request-reply ok: payload=42");
56            println!("{metrics}");
57        }
58        other => {
59            eprintln!("unexpected {other:?}");
60            std::process::exit(1);
61        }
62    }
63}

examples/crash_and_restart.rs (line 66)

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

examples/throughput.rs (line 71)

19fn main() {
20    let n: u32 = match std::env::args().nth(1) {
21        Some(s) => match s.parse() {
22            Ok(v) => v,
23            Err(_) => 50_000,
24        },
25        None => 50_000,
26    };
27
28    let workers = match std::thread::available_parallelism() {
29        Ok(p) => p.get(),
30        Err(_) => 1,
31    };
32
33    let rt = match Runtime::with_config(
34        trivial_chunk(),
35        RuntimeConfig {
36            workers,
37            quantum: 10_000,
38            mailbox: byteflow::MailboxConfig::DEFAULT,
39        },
40    ) {
41        Ok(rt) => rt,
42        Err(e) => {
43            eprintln!("runtime: {e}");
44            std::process::exit(1);
45        }
46    };
47    let Some(worker_fn) = rt.function_index("worker") else {
48        eprintln!("missing worker");
49        std::process::exit(1);
50    };
51
52    let start = Instant::now();
53    let mut handles = Vec::with_capacity(n as usize);
54    for _ in 0..n {
55        match rt.spawn(worker_fn, &[]) {
56            Ok(h) => handles.push(h),
57            Err(e) => {
58                eprintln!("spawn: {e}");
59                std::process::exit(1);
60            }
61        }
62    }
63    let mut ok = 0u32;
64    for h in handles {
65        if matches!(h.join(), FlowOutcome::Completed(Value::Int(1))) {
66            ok += 1;
67        }
68    }
69    let elapsed = start.elapsed();
70    let metrics = rt.metrics();
71    rt.shutdown();
72
73    let secs = elapsed.as_secs_f64().max(1e-9);
74    println!("processes={ok}/{n}");
75    println!("workers={workers}");
76    println!("elapsed_ms={:.2}", elapsed.as_secs_f64() * 1000.0);
77    println!("spawns_per_sec={:.0}", ok as f64 / secs);
78    println!("{metrics}");
79}