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 [super::error]).

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

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

More examples

Hide additional examples

examples/throughput.rs (lines 29-35)

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

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

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

More examples

Hide additional examples

examples/atomic_actors.rs (lines 23-30)

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

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

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

More examples

Hide additional examples

examples/atomic_actors.rs (line 41)

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

examples/throughput.rs (line 50)

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

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

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

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

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

More examples

Hide additional examples

examples/atomic_actors.rs (line 37)

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

examples/crash_and_restart.rs (line 29)

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

examples/throughput.rs (line 42)

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

Source

pub fn metrics(&self) -> RuntimeMetricsSnapshot

Examples found in repository ?

examples/ping_pong.rs (line 37)

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

More examples

Hide additional examples

examples/atomic_actors.rs (line 49)

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

examples/crash_and_restart.rs (line 61)

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

examples/throughput.rs (line 65)

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

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

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

examples/crash_and_restart.rs (line 65)

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

examples/throughput.rs (line 66)

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