diff --git a/FUNDING.md b/FUNDING.md new file mode 100644 index 0000000..70a9395 --- /dev/null +++ b/FUNDING.md @@ -0,0 +1,100 @@ +# Funding + +Limen is an independent open-source project. This document describes what +funding supports, the licensing commitments for funded work, and how the +project plans to sustain itself long-term. + +--- + +## What Funding Supports + +Limen is working toward a production-ready v1.0 release. Funding +accelerates the path from the current alpha to a stable, battle-tested +framework that downstream projects can depend on. + +The v1.0 roadmap includes: + +- **Contract stabilisation** — freezing the public trait surface (`Node`, + `Edge`, `GraphApi`, `LimenRuntime`, `MemoryManager`) with semver + guarantees. +- **Robotics primitives** — freshness, liveness, criticality, urgency, and + mailbox semantics for real-time control system graphs. +- **Production runtimes** — `NoAllocRuntime` with full policy enforcement + and `ThreadedRuntime` with EDF and throughput scheduling. +- **Inference backends** — TFLite Micro (Cortex-M4) and Tract + (desktop/server) behind the `ComputeBackend` trait. +- **Platform implementations** — Raspberry Pi and Cortex-M4 reference + platforms with clock, GPIO, and I/O support. +- **Conformance and testing** — comprehensive test suite covering tensor + payloads, N-to-M graphs, mailbox semantics, staleness detection, liveness + violations, and telemetry events. +- **Documentation and release hygiene** — full API documentation, feature + flag audit, CI gates, changelog, and cross-platform reference examples. + +See the [Roadmap](docs/roadmap.md) for the full phased plan. + +--- + +## Licensing Commitment for Funded Work + +All deliverables produced under accepted public grants — including code, +tests, documentation, and example artifacts delivered under funded +milestones — will be released publicly under the Apache License, Version 2.0. + +These deliverables will remain available under Apache-2.0. The Apache +License is irrevocable: once a version is released under Apache-2.0, that +version stays under Apache-2.0 permanently. This is a property of the +licence itself, not a policy that can be changed. + +The goal is to deliver a production-ready v1.0 under Apache-2.0 as a +public good. + +--- + +## Future Versions and Sustainability + +Future major versions of Limen may be distributed under different or +additional licence terms (for example, dual-licensing for commercial +sustainability). This is a standard approach used by many open-source +projects and does not affect the Apache-2.0 availability of earlier +releases. + +In concrete terms: if v1.0 is released under Apache-2.0 with grant +funding, v1.0 remains Apache-2.0 forever. Any future versions developed +independently may ship under different terms — but v1.0 and its source +code remain freely available under the original licence. + +The long-term aspiration is to keep Limen free and open-source. If the +project achieves significant adoption, moving it to a non-profit +foundation for community stewardship is a possibility the maintainer is +open to. + +--- + +## Contributor Licence Agreement + +Contributions to Limen are accepted under a lightweight +[Contributor Licence Agreement](CLA.md) (CLA). The CLA grants the project +maintainer the right to sublicence and relicence contributions. This +flexibility supports long-term sustainability while keeping the open-source +core accessible. + +The CLA does not affect the licensing commitments for funded work described +in this document. + +--- + +## How to Support Limen + +If you are interested in funding, sponsoring, or collaborating on Limen, +please open an issue or contact the maintainer directly. + +- **Public grants and innovation funds** — Limen is actively seeking + support from open-source and public innovation programmes. +- **Corporate sponsorship** — partnerships with companies in robotics, + embedded systems, and edge AI. +- **Community support** — GitHub Sponsors (coming soon). + +--- + +Copyright © 2025–present Arlo Louis Byrne (idky137) diff --git a/README.md b/README.md index 0c2b391..8cd83c7 100644 --- a/README.md +++ b/README.md @@ -60,23 +60,22 @@ Limen closes this gap. ## Quickstart -**Prerequisites:** Rust stable toolchain. No external dependencies. +**Prerequisites:** Rust stable toolchain (1.81+). No external dependencies. ```bash # Clone and enter the repository git clone https://github.com/idky137/limen.git cd limen -# Run the no_std single-threaded pipeline -cargo test -p limen-examples -- codegen_core_pipeline_runs_with_nostd_runtime --nocapture - -# Run the std multi-threaded pipeline (concurrent queues, scoped threads) -cargo test -p limen-examples --features std -- codegen_std_pipeline_runs_with_std_runtime --nocapture +# Run the pipeline demo (requires std feature for concurrent execution) +cargo run -p limen-examples --features std --example pipeline_demo ``` -Both tests build a three-node **Source → Model → Sink** pipeline from a -codegen-generated graph, wire it to a runtime, and step it through several -iterations. `--nocapture` prints edge occupancies and telemetry to stdout. +The demo builds a **Sensor → Model → Actuator** pipeline, runs it +step-by-step showing backpressure and batching, then runs the same graph +concurrently with one thread per node. See the +[Quickstart Guide](docs/quickstart.md) for a full walkthrough of the output +and what each section demonstrates. --- @@ -208,9 +207,11 @@ beyond. | Document | Description | |---|---| | [API Reference](https://idky137.github.io/limen/) | Rustdoc API documentation (GitHub Pages) | +| [Quickstart Guide](docs/quickstart.md) | Pipeline demo walkthrough and expected output | | [Architecture Guide](docs/architecture/index.md) | System design, memory model, execution flow | | [Decision Records](docs/ADRs/) | Rationale behind key design decisions | | [Roadmap](docs/roadmap.md) | Phased plan to v0.1.0 and stretch goals | +| [Funding & Licensing](FUNDING.md) | Funded work commitments and how to support the project | | [Development Guide](docs/dev_guide.md) | Building, testing, and contributing | --- @@ -221,12 +222,14 @@ Limen is released under the Apache License, Version 2.0 ([LICENSE-APACHE](LICENSE-APACHE)). The project is developed and owned by its original author. Contributions -are accepted under a Contributor Licence Agreement (CLA), which ensures -the project can be maintained, evolved, and (if necessary) dual-licensed -for commercial use. - -The goal is to keep Limen broadly accessible while retaining the ability -to support long-term sustainability and real-world deployment. +are accepted under a Contributor Licence Agreement ([CLA](CLA.md)), which +ensures the project can be maintained, evolved, and (if necessary) +dual-licensed for commercial use. + +All deliverables produced under accepted public grants will be released +under Apache-2.0 and remain available under Apache-2.0. Future major +versions may be distributed under different or additional licence terms. +See [FUNDING.md](FUNDING.md) for details. --- diff --git a/docs/quickstart.md b/docs/quickstart.md new file mode 100644 index 0000000..d219ae1 --- /dev/null +++ b/docs/quickstart.md @@ -0,0 +1,198 @@ +# Quickstart Guide + +This guide walks through the **pipeline demo** — a runnable example that +demonstrates Limen's core capabilities in under a minute. + +--- + +## Prerequisites + +- **Rust stable toolchain** (1.81+). Install via [rustup](https://rustup.rs/). +- No external dependencies — everything builds from source. + +```bash +git clone https://github.com/idky137/limen.git +cd limen +``` + +--- + +## Running the Demo + +```bash +cargo run -p limen-examples --features std --example pipeline_demo +``` + +The `std` feature enables concurrent queues and threaded runtimes. Without +it, only the single-threaded `no_std` path is available. + +--- + +## What the Demo Does + +The demo builds a three-node **Sensor → Model → Actuator** inference +pipeline — the canonical pattern for edge AI and robotics systems: + +``` +[Sensor] --edge 0--> [Model] --edge 1--> [Actuator] +``` + +| Node | Role | Real-world analogue | +|---|---|---| +| **Sensor** | Produces tensor data from an upstream backlog | IMU, camera, LiDAR | +| **Model** | Identity inference pass-through (batched) | TFLite Micro, Tract | +| **Actuator** | Consumes results | Motor controller, UART, MQTT | + +**Configuration:** +- **Edges:** SPSC (single-producer single-consumer) queues, capacity 16 +- **Backpressure:** DropOldest — when the queue is full, the oldest item is + evicted. This is the correct policy for real-time systems where stale + sensor data should be discarded in favour of fresh readings. +- **Batching:** Fixed size 4 on sensor and model nodes — items are grouped + into batches of 4 before processing. +- **Memory:** Static (no-alloc) managers with 20 slots per edge — no heap + allocation required. + +The same graph definition, node logic, and policies work identically on +bare-metal `no_std` microcontrollers and multi-threaded Linux servers. + +--- + +## Expected Output Walkthrough + +### Header + +``` +==================================================================== + Limen Pipeline Demo -- Edge Inference Runtime +==================================================================== + + Graph topology: + + [Sensor] --edge 0--> [Model] --edge 1--> [Actuator] + + Edges: SPSC queues, capacity 16, DropOldest backpressure + Batching: fixed size 4 on sensor and model nodes + Memory: static (no-alloc) managers, 20 slots per edge +``` + +### Part 1: Step-by-step execution + +This section runs the pipeline one step at a time using a single-threaded +`no_std`-compatible runtime. The source is pre-loaded with 20 items (more +than the edge capacity of 16), so backpressure will engage. + +``` + Step | sensor->model | model->actuator + -------+------------------+--------------------- + 1 | 4 items | 0 items + 2 | 0 items | 4 items + 3 | 0 items | 3 items + 4 | 4 items | 3 items + 5 | 0 items | 7 items + ... + 15 | 0 items | 15 items + + Pipeline complete: + Messages consumed by actuator: 5 + Backpressure policy: DropOldest (stale data evicted) +``` + +**What to notice:** + +- **Batching in action:** The sensor pushes 4 items per batch (visible in + steps 1, 4, 7, 10, 13 where `sensor→model` jumps by 4). +- **Pipeline flow:** The model consumes from edge 0 and produces to edge 1. + The actuator consumes from edge 1 one item at a time. +- **Round-robin scheduling:** The single-threaded runtime visits each node + in turn. One step = one pass through all three nodes. +- **Accumulation:** Items accumulate in `model→actuator` because the + actuator consumes one item per step while the model produces batches. +- **Backpressure:** The source generates more items than can fit (the + source also adds 1–2 random items to its backlog each step to simulate + continuous sensor input). The DropOldest policy ensures that stale data + is evicted rather than blocking the pipeline. + +### Part 2: Concurrent execution + +This section rebuilds the graph with thread-safe concurrent edges and runs +it with one worker thread per node for 500ms. + +``` + Launching 3 worker threads (one per node)... + Running for 500ms with continuous sensor input. + + Concurrent pipeline stopped gracefully via RuntimeStopHandle. + Messages processed by actuator: ~400 +``` + +**What to notice:** + +- **Thread-per-node:** Each node runs on its own OS thread with a backoff + scheduler (sleeps briefly when idle or backpressured). +- **Graceful shutdown:** A `RuntimeStopHandle` is cloned before `run()` and + passed to a timer thread that requests cooperative stop after 500ms. All + threads drain cleanly. +- **Throughput:** The concurrent pipeline processes hundreds of messages in + 500ms despite simulated processing delays in each node. + +### Telemetry Summary + +The final section prints per-node and per-edge metrics collected during the +concurrent run: + +``` +graph id: 0 + node id: 0 processed=~800 dropped=~360 ingress=0 egress=~800 ... + node id: 1 processed=~400 dropped=0 ingress=~400 egress=~400 ... + node id: 2 processed=~400 dropped=0 ingress=~400 egress=0 ... + edge id: 0 queue_depth=0 + edge id: 1 queue_depth=16 + edge id: 2 queue_depth=2 +``` + +**What to notice:** + +- **Sensor (node 0):** `dropped=~360` — nearly half the produced messages + were dropped by the DropOldest admission policy. This is correct + behaviour: the model processes batches slower than the sensor produces, + so stale readings are evicted. +- **Model (node 1):** `dropped=0` — the model never drops; it processes + everything it receives. `processed` is roughly half of the sensor's + output because it processes in batches of 4. +- **Actuator (node 2):** Consumes everything the model produces. +- **Edge depths:** Edge 1 (`sensor→model`) is drained (0). Edge 2 + (`model→actuator`) is at capacity (16) — the actuator is the bottleneck. +- **Latency metrics:** `lat_sum`, `lat_cnt`, `lat_max` track per-node + processing time in nanoseconds. + +> Exact numbers vary between runs due to thread scheduling and simulated +> processing delays. + +--- + +## Next Steps + +| Topic | Resource | +|---|---| +| Define your own graph | See `limen-examples/examples/pipeline_demo.rs` for the `define_graph!` DSL | +| Architecture deep dive | [Architecture Guide](architecture/index.md) | +| API reference | `cargo doc --workspace --open` or [GitHub Pages](https://idky137.github.io/limen/) | +| Design decisions | [ADRs](ADRs/) | +| Full roadmap | [Roadmap](roadmap.md) | + +### Running the Integration Tests + +The full test suite exercises all graph generation methods (proc-macro, +build-script codegen, hand-written) across all feature profiles: + +```bash +# no_std (default) — single-threaded, static queues +cargo test -p limen-examples + +# std — concurrent queues, threaded runtimes +cargo test -p limen-examples --features std + +# All features including unsafe lock-free ring buffer +cargo test -p limen-examples --features spsc_raw +``` diff --git a/limen-examples/Cargo.toml b/limen-examples/Cargo.toml index b845dcf..5dc8a9c 100644 --- a/limen-examples/Cargo.toml +++ b/limen-examples/Cargo.toml @@ -14,10 +14,18 @@ alloc = ["limen-core/alloc"] std = ["alloc", "limen-core/std"] spsc_raw = ["std", "limen-core/spsc_raw"] +[dependencies] +limen-core = { workspace = true, features = ["bench"] } +limen-build = { workspace = true } + [dev-dependencies] limen-core = { workspace = true, features = ["bench"] } limen-build = { workspace = true } +[[example]] +name = "pipeline_demo" +required-features = ["std"] + [build-dependencies] limen-core = { workspace = true, features = ["bench"] } limen-codegen = { workspace = true } diff --git a/limen-examples/examples/pipeline_demo.rs b/limen-examples/examples/pipeline_demo.rs new file mode 100644 index 0000000..b390e22 --- /dev/null +++ b/limen-examples/examples/pipeline_demo.rs @@ -0,0 +1,487 @@ +//! # Limen Pipeline Demo +//! +//! Demonstrates a **Sensor -> Model -> Actuator** inference pipeline built with +//! Limen's graph runtime. The same node logic and policies work identically on +//! bare-metal `no_std` targets and multi-threaded Linux servers. +//! +//! **What this demo shows:** +//! +//! 1. **Graph definition** via the `define_graph!` proc-macro DSL +//! 2. **Step-by-step execution** with edge occupancy visibility +//! 3. **Backpressure handling** — DropOldest policy evicting stale data +//! 4. **Batched inference** — fixed-size batch processing through the model node +//! 5. **Concurrent execution** — one thread per node with graceful shutdown +//! 6. **Structured telemetry** — per-node and per-edge metric collection +//! +//! Run with: +//! ```sh +//! cargo run -p limen-examples --features std --example pipeline_demo +//! ``` + +// --------------------------------------------------------------------------- +// Graph definition via proc-macro DSL +// --------------------------------------------------------------------------- +// +// This defines a three-node pipeline: +// +// [Sensor] --edge 0--> [Model] --edge 1--> [Actuator] +// +// - Sensor: produces incrementing tensor data (simulates a hardware sensor) +// - Model: identity inference pass-through (simulates an ML model) +// - Actuator: consumes results (simulates an actuator or output device) +// +// Edges use SPSC (single-producer single-consumer) queues with DropOldest +// backpressure — when the queue is full, the oldest item is evicted to make +// room for fresh data. This is the correct policy for real-time systems where +// stale sensor readings should be discarded. + +use limen_build::define_graph; +use limen_core::edge::Edge; +use limen_core::graph::{GraphApi, GraphNodeAccess}; +use limen_core::memory::PlacementAcceptance; +use limen_core::message::MessageFlags; +use limen_core::node::bench::{ + TestCounterSourceTensor, TestIdentityModelNodeTensor, TestSinkNodeTensor, TestTensorBackend, +}; +use limen_core::node::source::Source; +use limen_core::node::{Node, NodeCapabilities}; +use limen_core::policy::{ + AdmissionPolicy, BatchingPolicy, BudgetPolicy, DeadlinePolicy, EdgePolicy, NodePolicy, + OverBudgetAction, QueueCaps, +}; +use limen_core::prelude::linux::NoStdLinuxMonotonicClock; +use limen_core::prelude::TestTensor; +use limen_core::runtime::LimenRuntime; +use limen_core::types::{QoSClass, SequenceNumber, TraceId}; + +// Concurrent imports (std only) +use limen_core::edge::spsc_concurrent::ConcurrentEdge; +use limen_core::memory::concurrent_manager::ConcurrentMemoryManager; +use limen_core::prelude::concurrent::{spawn_telemetry_core, TelemetrySender}; +use limen_core::prelude::graph_telemetry::GraphTelemetry; +use limen_core::prelude::sink::IoLineWriter; +use limen_core::runtime::bench::concurrent_runtime::TestScopedRuntime; +use limen_core::telemetry::Telemetry; + +// --------------------------------------------------------------------------- +// Edge policy: capacity 16, DropOldest backpressure, drop over-budget items. +// --------------------------------------------------------------------------- +const EDGE_POLICY: EdgePolicy = EdgePolicy::new( + QueueCaps::new(16, 16, None, None), + AdmissionPolicy::DropOldest, + OverBudgetAction::Drop, +); + +// --------------------------------------------------------------------------- +// No-std graph (step-by-step demo) +// --------------------------------------------------------------------------- +define_graph! { + pub struct DemoPipelineNoStd; + + nodes { + 0: { + ty: limen_core::node::bench::TestCounterSourceTensor< + limen_core::prelude::linux::NoStdLinuxMonotonicClock, 32>, + in_ports: 0, + out_ports: 1, + in_payload: (), + out_payload: TestTensor, + name: Some("sensor"), + ingress_policy: EDGE_POLICY + }, + 1: { + ty: limen_core::node::bench::TestIdentityModelNodeTensor<16>, + in_ports: 1, + out_ports: 1, + in_payload: TestTensor, + out_payload: TestTensor, + name: Some("model") + }, + 2: { + ty: limen_core::node::bench::TestSinkNodeTensor, + in_ports: 1, + out_ports: 0, + in_payload: TestTensor, + out_payload: (), + name: Some("actuator") + }, + } + + edges { + 0: { + ty: limen_core::edge::bench::TestSpscRingBuf<16>, + payload: TestTensor, + manager: limen_core::memory::static_manager::StaticMemoryManager, + from: (0, 0), + to: (1, 0), + policy: EDGE_POLICY, + name: Some("sensor->model") + }, + 1: { + ty: limen_core::edge::bench::TestSpscRingBuf<16>, + payload: TestTensor, + manager: limen_core::memory::static_manager::StaticMemoryManager, + from: (1, 0), + to: (2, 0), + policy: EDGE_POLICY, + name: Some("model->actuator") + }, + } +} + +// --------------------------------------------------------------------------- +// Concurrent graph (thread-per-node demo) +// --------------------------------------------------------------------------- +define_graph! { + pub struct DemoPipelineConcurrent; + + nodes { + 0: { + ty: limen_core::node::bench::TestCounterSourceTensor< + limen_core::prelude::linux::NoStdLinuxMonotonicClock, 32>, + in_ports: 0, + out_ports: 1, + in_payload: (), + out_payload: TestTensor, + name: Some("sensor"), + ingress_policy: EDGE_POLICY + }, + 1: { + ty: limen_core::node::bench::TestIdentityModelNodeTensor<16>, + in_ports: 1, + out_ports: 1, + in_payload: TestTensor, + out_payload: TestTensor, + name: Some("model") + }, + 2: { + ty: limen_core::node::bench::TestSinkNodeTensor, + in_ports: 1, + out_ports: 0, + in_payload: TestTensor, + out_payload: (), + name: Some("actuator") + }, + } + + edges { + 0: { + ty: limen_core::edge::spsc_concurrent::ConcurrentEdge, + payload: TestTensor, + manager: limen_core::memory::concurrent_manager::ConcurrentMemoryManager, + from: (0, 0), + to: (1, 0), + policy: EDGE_POLICY, + name: Some("sensor->model") + }, + 1: { + ty: limen_core::edge::spsc_concurrent::ConcurrentEdge, + payload: TestTensor, + manager: limen_core::memory::concurrent_manager::ConcurrentMemoryManager, + from: (1, 0), + to: (2, 0), + policy: EDGE_POLICY, + name: Some("model->actuator") + }, + } + + concurrent; +} + +// --------------------------------------------------------------------------- +// Type aliases +// --------------------------------------------------------------------------- +type Clock = NoStdLinuxMonotonicClock; +type Sensor = TestCounterSourceTensor; +type Model = TestIdentityModelNodeTensor<16>; +type Actuator = TestSinkNodeTensor; + +type StdTelemetryInner = GraphTelemetry<3, 3, IoLineWriter>; +type StdTelemetry = TelemetrySender; +type StdRuntime = TestScopedRuntime; + +fn main() { + println!( + "\n\ + ====================================================================\n\ + \x20 Limen Pipeline Demo -- Edge Inference Runtime\n\ + ====================================================================\n\ + \n\ + \x20 Graph topology:\n\ + \n\ + \x20 [Sensor] --edge 0--> [Model] --edge 1--> [Actuator]\n\ + \n\ + \x20 Edges: SPSC queues, capacity 16, DropOldest backpressure\n\ + \x20 Batching: fixed size 4 on sensor and model nodes\n\ + \x20 Memory: static (no-alloc) managers, 20 slots per edge\n" + ); + + part1_step_by_step(); + part2_concurrent(); +} + +// ========================================================================= +// Part 1: Step-by-step execution (no_std compatible) +// ========================================================================= +fn part1_step_by_step() { + println!( + "--------------------------------------------------------------------\n\ + \x20 Part 1: Step-by-step execution (no_std compatible)\n\ + --------------------------------------------------------------------\n" + ); + + let clock = Clock::new(); + + // Node policies: fixed batch size of 4. + let sensor_policy = NodePolicy::new( + BatchingPolicy::fixed(4), + BudgetPolicy::new(None, None), + DeadlinePolicy::new(false, None, None), + ); + let model_policy = NodePolicy::new( + BatchingPolicy::fixed(4), + BudgetPolicy::new(None, None), + DeadlinePolicy::new(false, None, None), + ); + let actuator_policy = NodePolicy::new( + BatchingPolicy::none(), + BudgetPolicy::new(None, None), + DeadlinePolicy::new(false, None, None), + ); + + // Build nodes. + let mut sensor: Sensor = Sensor::new( + clock, + 0, + TraceId::new(0u64), + SequenceNumber::new(0u64), + None, + QoSClass::BestEffort, + MessageFlags::empty(), + NodeCapabilities::default(), + sensor_policy, + [PlacementAcceptance::default()], + EDGE_POLICY, + ); + + // Populate the sensor's upstream backlog with 20 items. + // This exceeds the edge capacity (16), so backpressure will engage. + sensor.produce_n_items_in_backlog(20); + println!(" Source backlog: 20 items loaded (exceeds edge capacity of 16)\n"); + + let model: Model = Model::new( + TestTensorBackend, + (), + model_policy, + NodeCapabilities::default(), + [PlacementAcceptance::default()], + [PlacementAcceptance::default()], + ) + .unwrap(); + + let actuator: Actuator = Actuator::new( + NodeCapabilities::default(), + actuator_policy, + [PlacementAcceptance::default()], + |_s: &str| { + // Suppress per-message output in step mode for cleaner demo output. + }, + ); + + // Build queues and memory managers (all static, no heap allocation). + let q0 = limen_core::edge::bench::TestSpscRingBuf::<16>::default(); + let q1 = limen_core::edge::bench::TestSpscRingBuf::<16>::default(); + let mgr0 = limen_core::memory::static_manager::StaticMemoryManager::::new(); + let mgr1 = limen_core::memory::static_manager::StaticMemoryManager::::new(); + + // Assemble the graph. All wiring is type-checked at compile time. + let mut graph = DemoPipelineNoStd::new(sensor, model, actuator, q0, q1, mgr0, mgr1); + + // Validate graph integrity (descriptor consistency, edge wiring). + graph.validate_graph().unwrap(); + + // Create the no_std runtime (single-threaded, round-robin scheduler). + use limen_core::prelude::graph_telemetry::GraphTelemetry; + use limen_core::prelude::sink::FmtLineWriter; + use limen_core::telemetry::sink::FixedBuffer; + + type NoStdTelemetry = GraphTelemetry<3, 3, FmtLineWriter>>; + + let telemetry_sink = FmtLineWriter::new(FixedBuffer::<4096>::new()); + let telemetry: NoStdTelemetry = NoStdTelemetry::new(0, true, telemetry_sink); + + use limen_core::runtime::bench::TestNoStdRuntime; + + type NoStdRuntime = TestNoStdRuntime; + type NoStdGraph = DemoPipelineNoStd; + + let mut runtime: NoStdRuntime = NoStdRuntime::new(); + runtime.init(&mut graph, clock, telemetry).unwrap(); + + // Step through the pipeline, printing edge occupancies after each step. + let total_steps = 15; + println!( + " {:>6} | {:>16} | {:>20}", + "Step", "sensor->model", "model->actuator" + ); + println!(" {:-<6}-+-{:-<16}-+-{:-<20}", "", "", ""); + + for step in 1..=total_steps { + let _ = runtime.step(&mut graph).unwrap(); + + let occ = LimenRuntime::::occupancies(&runtime); + // occ[0] is the virtual ingress edge, occ[1] is sensor->model, occ[2] is model->actuator + println!( + " {:>6} | {:>10} items | {:>14} items", + step, + occ[1].items(), + occ[2].items(), + ); + } + + // Access the actuator to check processed count. + let actuator_node = >::node_mut(&mut graph); + let processed = *actuator_node.node_mut().sink_mut().processed(); + + println!("\n Pipeline complete:"); + println!(" Messages consumed by actuator: {}", processed); + println!(" Backpressure policy: DropOldest (stale data evicted)\n"); +} + +// ========================================================================= +// Part 2: Concurrent execution (std, thread-per-node) +// ========================================================================= +fn part2_concurrent() { + println!( + "--------------------------------------------------------------------\n\ + \x20 Part 2: Concurrent execution (std, thread-per-node)\n\ + --------------------------------------------------------------------\n" + ); + + let clock = Clock::new(); + + let sensor_policy = NodePolicy::new( + BatchingPolicy::fixed(4), + BudgetPolicy::new(None, None), + DeadlinePolicy::new(false, None, None), + ); + let model_policy = NodePolicy::new( + BatchingPolicy::fixed(4), + BudgetPolicy::new(None, None), + DeadlinePolicy::new(false, None, None), + ); + let actuator_policy = NodePolicy::new( + BatchingPolicy::none(), + BudgetPolicy::new(None, None), + DeadlinePolicy::new(false, None, None), + ); + + let mut sensor: Sensor = Sensor::new( + clock, + 0, + TraceId::new(0u64), + SequenceNumber::new(0u64), + None, + QoSClass::BestEffort, + MessageFlags::empty(), + NodeCapabilities::default(), + sensor_policy, + [PlacementAcceptance::default()], + EDGE_POLICY, + ); + sensor.produce_n_items_in_backlog(20); + + let model: Model = Model::new( + TestTensorBackend, + (), + model_policy, + NodeCapabilities::default(), + [PlacementAcceptance::default()], + [PlacementAcceptance::default()], + ) + .unwrap(); + + let actuator: Actuator = Actuator::new( + NodeCapabilities::default(), + actuator_policy, + [PlacementAcceptance::default()], + |_s: &str| { + // Suppress per-message output for clean demo. + }, + ); + + // Concurrent edges and heap-backed memory managers. + let q0 = ConcurrentEdge::new(16); + let q1 = ConcurrentEdge::new(16); + let mgr0 = ConcurrentMemoryManager::::new(20); + let mgr1 = ConcurrentMemoryManager::::new(20); + + let mut graph = DemoPipelineConcurrent::new(sensor, model, actuator, q0, q1, mgr0, mgr1); + graph.validate_graph().unwrap(); + + // Telemetry: GraphTelemetry writing to stdout, wrapped in a concurrent sender. + // Events are disabled during the run to keep output clean; only the final + // metrics summary is printed. + let sink = IoLineWriter::::stdout_writer(); + let inner_telemetry: StdTelemetryInner = StdTelemetryInner::new(0, false, sink); + let telemetry_core = spawn_telemetry_core(inner_telemetry); + let telemetry: StdTelemetry = telemetry_core.sender(); + + // Runtime: one worker thread per node, simple backoff scheduler. + let mut runtime: StdRuntime = StdRuntime::new(); + runtime.init(&mut graph, clock, telemetry).unwrap(); + + // Get a stop handle (thread-safe), then launch concurrent execution. + type ConcurrentGraph = DemoPipelineConcurrent; + + let run_duration = std::time::Duration::from_millis(500); + let handle = LimenRuntime::::stop_handle(&runtime).unwrap(); + + println!(" Launching 3 worker threads (one per node)..."); + println!( + " Running for {}ms with continuous sensor input.\n", + run_duration.as_millis() + ); + + std::thread::spawn(move || { + std::thread::sleep(run_duration); + handle.request_stop(); + }); + + LimenRuntime::::run(&mut runtime, &mut graph).unwrap(); + graph.validate_graph().unwrap(); + + // Read processed count from the actuator. + let actuator_node = >::node_mut(&mut graph); + let processed = *actuator_node.node_mut().sink_mut().processed(); + + println!(" Concurrent pipeline stopped gracefully via RuntimeStopHandle."); + println!(" Messages processed by actuator: {}\n", processed); + + // Push final telemetry snapshot and flush. + println!( + "--------------------------------------------------------------------\n\ + \x20 Telemetry Summary\n\ + --------------------------------------------------------------------\n" + ); + + runtime + .with_telemetry(|telemetry| { + telemetry.push_metrics(); + telemetry.flush(); + }) + .expect("telemetry flush failed"); + + telemetry_core.shutdown_and_join(); + + println!( + "\n\ + ====================================================================\n\ + \x20 Demo complete. For more information:\n\ + \n\ + \x20 Quickstart guide: docs/quickstart.md\n\ + \x20 Architecture guide: docs/architecture/index.md\n\ + \x20 API reference: cargo doc --workspace --open\n\ + ====================================================================\n" + ); +}