From 1298e3be2c9cd3a9726b9e79f45e5568c0bb1413 Mon Sep 17 00:00:00 2001 From: Quang Nguyen Date: Mon, 13 Jul 2026 16:45:19 +0700 Subject: [PATCH 1/3] Feat: add an example for spmc ring buffer The example follows several other examples such as hello-single-latest-async. It features one producer and two consumers running at different rates and tries to simulate the lagging/dropped message behavior of the ring. --- Cargo.lock | 11 +++ Cargo.toml | 2 +- Makefile | 6 +- README.md | 2 +- examples/hello-spmc-ring-async/Cargo.toml | 14 +++ examples/hello-spmc-ring-async/README.md | 41 +++++++++ examples/hello-spmc-ring-async/src/main.rs | 100 +++++++++++++++++++++ 7 files changed, 172 insertions(+), 4 deletions(-) create mode 100644 examples/hello-spmc-ring-async/Cargo.toml create mode 100644 examples/hello-spmc-ring-async/README.md create mode 100644 examples/hello-spmc-ring-async/src/main.rs diff --git a/Cargo.lock b/Cargo.lock index 6c16ee6..bf67f99 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2193,6 +2193,17 @@ dependencies = [ "tokio", ] +[[package]] +name = "hello-spmc-ring-async" +version = "1.1.0" +dependencies = [ + "aimdb-core", + "aimdb-data-contracts", + "aimdb-sync", + "aimdb-tokio-adapter", + "tokio", +] + [[package]] name = "hermit-abi" version = "0.5.2" diff --git a/Cargo.toml b/Cargo.toml index e0c71f8..ed9a751 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,7 +40,7 @@ members = [ "examples/hello-single-latest-async", "examples/readme-quickstart", # Benchmarking infrastructure — host-only, excluded from default-members - "aimdb-bench", + "aimdb-bench", "examples/hello-spmc-ring-async", ] exclude = ["_external"] resolver = "2" diff --git a/Makefile b/Makefile index 109ff2c..2b76464 100644 --- a/Makefile +++ b/Makefile @@ -182,7 +182,7 @@ test: fmt: @printf "$(GREEN)Formatting code (workspace members only)...$(NC)\n" - @for pkg in aimdb-derive aimdb-data-contracts aimdb-core aimdb-client aimdb-embassy-adapter aimdb-tokio-adapter aimdb-wasm-adapter aimdb-sync aimdb-persistence aimdb-persistence-sqlite aimdb-mqtt-connector aimdb-knx-connector aimdb-ws-protocol aimdb-websocket-connector aimdb-uds-connector aimdb-serial-connector aimdb-tcp-connector aimdb-codegen aimdb-cli aimdb-mcp sync-api-demo tokio-mqtt-connector-demo embassy-mqtt-connector-demo tokio-knx-connector-demo embassy-knx-connector-demo embassy-serial-connector-demo embassy-bench-stm32h5 weather-mesh-common weather-hub weather-station-alpha weather-station-beta hello-mailbox hello-mailbox-async hello-single-latest-async aimdb-bench; do \ + @for pkg in aimdb-derive aimdb-data-contracts aimdb-core aimdb-client aimdb-embassy-adapter aimdb-tokio-adapter aimdb-wasm-adapter aimdb-sync aimdb-persistence aimdb-persistence-sqlite aimdb-mqtt-connector aimdb-knx-connector aimdb-ws-protocol aimdb-websocket-connector aimdb-uds-connector aimdb-serial-connector aimdb-tcp-connector aimdb-codegen aimdb-cli aimdb-mcp sync-api-demo tokio-mqtt-connector-demo embassy-mqtt-connector-demo tokio-knx-connector-demo embassy-knx-connector-demo embassy-serial-connector-demo embassy-bench-stm32h5 weather-mesh-common weather-hub weather-station-alpha weather-station-beta hello-mailbox hello-mailbox-async hello-single-latest-async hello-spmc-ring-async aimdb-bench; do \ printf "$(YELLOW) → Formatting $$pkg$(NC)\n"; \ cargo fmt -p $$pkg 2>/dev/null || true; \ done @@ -191,7 +191,7 @@ fmt: fmt-check: @printf "$(GREEN)Checking code formatting (workspace members only)...$(NC)\n" @FAILED=0; \ - for pkg in aimdb-derive aimdb-data-contracts aimdb-core aimdb-client aimdb-embassy-adapter aimdb-tokio-adapter aimdb-wasm-adapter aimdb-sync aimdb-persistence aimdb-persistence-sqlite aimdb-mqtt-connector aimdb-knx-connector aimdb-ws-protocol aimdb-websocket-connector aimdb-uds-connector aimdb-serial-connector aimdb-tcp-connector aimdb-codegen aimdb-cli aimdb-mcp sync-api-demo tokio-mqtt-connector-demo embassy-mqtt-connector-demo tokio-knx-connector-demo embassy-knx-connector-demo embassy-serial-connector-demo embassy-bench-stm32h5 weather-mesh-common weather-hub weather-station-alpha weather-station-beta hello-mailbox hello-mailbox-async hello-single-latest-async aimdb-bench; do \ + for pkg in aimdb-derive aimdb-data-contracts aimdb-core aimdb-client aimdb-embassy-adapter aimdb-tokio-adapter aimdb-wasm-adapter aimdb-sync aimdb-persistence aimdb-persistence-sqlite aimdb-mqtt-connector aimdb-knx-connector aimdb-ws-protocol aimdb-websocket-connector aimdb-uds-connector aimdb-serial-connector aimdb-tcp-connector aimdb-codegen aimdb-cli aimdb-mcp sync-api-demo tokio-mqtt-connector-demo embassy-mqtt-connector-demo tokio-knx-connector-demo embassy-knx-connector-demo embassy-serial-connector-demo embassy-bench-stm32h5 weather-mesh-common weather-hub weather-station-alpha weather-station-beta hello-mailbox hello-mailbox-async hello-single-latest-async hello-spmc-ring-async aimdb-bench; do \ printf "$(YELLOW) → Checking $$pkg$(NC)\n"; \ if ! cargo fmt -p $$pkg -- --check 2>&1; then \ printf "$(RED)❌ Formatting check failed for $$pkg$(NC)\n"; \ @@ -418,6 +418,8 @@ examples: cargo build --package hello-mailbox-async @printf "$(YELLOW) → Building hello-single-latest-async$(NC)\n" cargo build --package hello-single-latest-async + @printf "$(YELLOW) → Building hello-spmc-ring-async$(NC)\n" + cargo build --package hello-spmc-ring-async @printf "$(YELLOW) → Building readme-quickstart (compiled README example)$(NC)\n" cargo build --package readme-quickstart @printf "$(GREEN)All examples built successfully!$(NC)\n" diff --git a/README.md b/README.md index 1900aab..c5fdfd1 100644 --- a/README.md +++ b/README.md @@ -131,7 +131,7 @@ docker compose up | Buffer | Semantics | Use cases | | --- | --- | --- | -| **SPMC Ring** | Bounded stream with independent consumers | Sensor telemetry, event logs | +| [**SPMC Ring**](examples/hello-spmc-ring-async) | Bounded stream with independent consumers | Sensor telemetry, event logs | | [**SingleLatest**](examples/hello-single-latest-async) | Only the current value matters | Feature flags, config, UI state | | [**Mailbox**](examples/hello-mailbox) / [**async Mailbox**](examples/hello-mailbox-async)| Latest instruction wins | Device commands, actuation, RPC | diff --git a/examples/hello-spmc-ring-async/Cargo.toml b/examples/hello-spmc-ring-async/Cargo.toml new file mode 100644 index 0000000..ecea1ee --- /dev/null +++ b/examples/hello-spmc-ring-async/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "hello-spmc-ring-async" +version.workspace = true +edition.workspace = true +license.workspace = true +description = "AimDB minimal example demonstrating async SPMC buffer" +publish = false + +[dependencies] +aimdb-core = { path = "../../aimdb-core", features = ["std"] } +aimdb-tokio-adapter = { path = "../../aimdb-tokio-adapter", features = ["tokio-runtime"] } +aimdb-sync = { path = "../../aimdb-sync" } +aimdb-data-contracts = { path = "../../aimdb-data-contracts"} +tokio = { workspace = true, features = ["full"] } diff --git a/examples/hello-spmc-ring-async/README.md b/examples/hello-spmc-ring-async/README.md new file mode 100644 index 0000000..671ef9b --- /dev/null +++ b/examples/hello-spmc-ring-async/README.md @@ -0,0 +1,41 @@ +# hello-spmc-ring-async: async Spmc Ring buffer demo +The Spmc Ring buffer supports Single Producer - Multi Consumer pattern. It features a ring buffer with fixed capacity. Each consumer could read the full sequence at its own rate without waiting for consumption states of other consumers. + +## When to use +This Spmc Ring buffer demo is useful in a variety of scenarios where you need to broadcast messages to different types of consumers but you are not sure if one could lag behind others. + +## How it works +The Spmc buffer demo works by simulating: +- A producer generating a message at fixed interval (struct `Temperature`); +- A consumer/observer 01 consuming at a rate faster than producer's; +- A consumer/observer 02 consuming at a rate slower than producer's, and eventually lagging behind (messages dropped), as the ring size is small (10 slots); + +## How to run +From the workspace root, run: +``` +cargo run -p hello-spmc-ring-async +``` +**Expected output** +``` +=== hello-spmc-ring-async: SPMC ring buffer demo === + +Ring size: 10 +Producer at rate 20.0 messages/sec +Observer 01 at rate 25.0 message/sec +Observer 02 at rate 6.7 message/sec + + +Observer 01: At 1783932208: -18 celcius +Observer 02: At 1783932208: -18 celcius +Observer 01: At 1783932208: -16 celcius +Observer 01: At 1783932208: -14 celcius +... +Observer 02: At 1783932208: -2 celcius +Observer 01: At 1783932209: 30 celcius +Observer 01: At 1783932209: 32 celcius +Observer 01: At 1783932209: 34 celcius +Observer 02 lagged, dropped 2 +Observer 02: At 1783932209: 4 celcius +... + +``` diff --git a/examples/hello-spmc-ring-async/src/main.rs b/examples/hello-spmc-ring-async/src/main.rs new file mode 100644 index 0000000..2a475f1 --- /dev/null +++ b/examples/hello-spmc-ring-async/src/main.rs @@ -0,0 +1,100 @@ +use std::time::Duration; +use std::time::{SystemTime, UNIX_EPOCH}; +use std::{fmt::Display, sync::Arc}; + +use aimdb_core::{buffer::BufferCfg, AimDbBuilder, Producer, RuntimeContext}; +use aimdb_core::{Consumer, DbError}; +use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; + +#[derive(Debug, Clone, PartialEq)] +pub struct Temperature { + /// Temperature in degrees Celsius + pub celcius: f32, + + /// Unix timestamp (milliseconds) + pub timestamp: u64, +} + +impl Display for Temperature { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "At {}: {} celcius", self.timestamp, self.celcius) + } +} + +// Main function +#[tokio::main] +async fn main() -> Result<(), DbError> { + println!("=== hello-spmc-ring-async: SPMC ring buffer demo ===\n"); + println!("Ring size: 10"); + println!("Producer at rate 20.0 messages/sec"); + println!("Observer 01 at rate 25.0 message/sec"); + println!("Observer 02 at rate 6.7 message/sec"); + println!("\n"); + + // configuration + let adapter = Arc::new(TokioAdapter); + let mut builder = AimDbBuilder::new().runtime(adapter); + + // Two observers consuming at different rates + builder.configure::("sensor.temp", |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) + .source(rollout_source) + .tap(rollout_observer01) + .tap(rollout_observer02); + }); + + let (_db, runner) = builder.build().await?; + tokio::spawn(runner.run()); + tokio::time::sleep(Duration::from_millis(6000)).await; + + Ok(()) +} + +async fn rollout_source(ctx: RuntimeContext, producer: Producer) { + let time = ctx.time(); + + let t0: f32 = -20.0; + for i in 0..40 { + time.sleep_millis(50).await; + publich_rollout(&producer, t0 + (i as f32 + 1.0) * 2.0).await; + } +} + +async fn publich_rollout(producer: &Producer, t: f32) { + let now = SystemTime::now(); + let d = now.duration_since(UNIX_EPOCH).expect("Error getting time"); + let temperature = Temperature { + celcius: t, + timestamp: d.as_secs(), + }; + producer.produce(temperature); +} + +async fn rollout_observer01(ctx: RuntimeContext, consumer: Consumer) { + let mut reader = consumer.subscribe(); + let time = ctx.time(); + + while let Ok(t) = reader.recv().await { + println!("Observer 01: {}", t); + time.sleep_millis(40).await; + } +} + +async fn rollout_observer02(ctx: RuntimeContext, consumer: Consumer) { + let mut reader = consumer.subscribe(); + let time = ctx.time(); + + // Consume at slower rate, must miss data + loop { + match reader.recv().await { + Ok(t) => { + println!("Observer 02: {}", t); + time.sleep_millis(150).await; + } + Err(DbError::BufferLagged { lag_count, .. }) => { + println!("Observer 02 lagged, dropped {lag_count}"); + } + Err(_) => break, + }; + } +} From a36764473f716988bc88ee73a8263e1647bc91d8 Mon Sep 17 00:00:00 2001 From: Quang Nguyen <61941605+solus161@users.noreply.github.com> Date: Tue, 14 Jul 2026 06:59:55 +0700 Subject: [PATCH 2/3] Update examples/hello-spmc-ring-async/Cargo.toml Remove unneeded package. Using tokio with time feature only. Co-authored-by: sounds.like.lx <147444674+lxsaah@users.noreply.github.com> --- examples/hello-spmc-ring-async/Cargo.toml | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/examples/hello-spmc-ring-async/Cargo.toml b/examples/hello-spmc-ring-async/Cargo.toml index ecea1ee..c610225 100644 --- a/examples/hello-spmc-ring-async/Cargo.toml +++ b/examples/hello-spmc-ring-async/Cargo.toml @@ -9,6 +9,4 @@ publish = false [dependencies] aimdb-core = { path = "../../aimdb-core", features = ["std"] } aimdb-tokio-adapter = { path = "../../aimdb-tokio-adapter", features = ["tokio-runtime"] } -aimdb-sync = { path = "../../aimdb-sync" } -aimdb-data-contracts = { path = "../../aimdb-data-contracts"} -tokio = { workspace = true, features = ["full"] } +tokio = { workspace = true, features = ["time"] } From 1da0ae713587705987c2d8199a314c16bec0b640 Mon Sep 17 00:00:00 2001 From: Quang Nguyen Date: Tue, 14 Jul 2026 07:21:10 +0700 Subject: [PATCH 3/3] Update to adhere to convention - Reorganize workspace order - Switch to getting timestamp from time context instead of system unix time --- Cargo.lock | 2 -- Cargo.toml | 3 ++- examples/hello-spmc-ring-async/src/main.rs | 17 ++++++++--------- 3 files changed, 10 insertions(+), 12 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bf67f99..ae6d85b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2198,8 +2198,6 @@ name = "hello-spmc-ring-async" version = "1.1.0" dependencies = [ "aimdb-core", - "aimdb-data-contracts", - "aimdb-sync", "aimdb-tokio-adapter", "tokio", ] diff --git a/Cargo.toml b/Cargo.toml index ed9a751..34e7894 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -38,9 +38,10 @@ members = [ "examples/hello-mailbox", "examples/hello-mailbox-async", "examples/hello-single-latest-async", + "examples/hello-spmc-ring-async", "examples/readme-quickstart", # Benchmarking infrastructure — host-only, excluded from default-members - "aimdb-bench", "examples/hello-spmc-ring-async", + "aimdb-bench", ] exclude = ["_external"] resolver = "2" diff --git a/examples/hello-spmc-ring-async/src/main.rs b/examples/hello-spmc-ring-async/src/main.rs index 2a475f1..e506390 100644 --- a/examples/hello-spmc-ring-async/src/main.rs +++ b/examples/hello-spmc-ring-async/src/main.rs @@ -1,5 +1,4 @@ use std::time::Duration; -use std::time::{SystemTime, UNIX_EPOCH}; use std::{fmt::Display, sync::Arc}; use aimdb_core::{buffer::BufferCfg, AimDbBuilder, Producer, RuntimeContext}; @@ -27,8 +26,8 @@ async fn main() -> Result<(), DbError> { println!("=== hello-spmc-ring-async: SPMC ring buffer demo ===\n"); println!("Ring size: 10"); println!("Producer at rate 20.0 messages/sec"); - println!("Observer 01 at rate 25.0 message/sec"); - println!("Observer 02 at rate 6.7 message/sec"); + println!("Observer 01 at rate 25.0 messages/sec"); + println!("Observer 02 at rate 6.7 messages/sec"); println!("\n"); // configuration @@ -52,20 +51,20 @@ async fn main() -> Result<(), DbError> { async fn rollout_source(ctx: RuntimeContext, producer: Producer) { let time = ctx.time(); - let t0: f32 = -20.0; for i in 0..40 { time.sleep_millis(50).await; - publich_rollout(&producer, t0 + (i as f32 + 1.0) * 2.0).await; + // Read wall-clock time through the runtime abstraction rather than + // `SystemTime::now()`, so the example stays runtime/sim-friendly. + let timestamp = time.unix_time().map(|(secs, _)| secs).unwrap_or(0); + publich_rollout(&producer, t0 + (i as f32 + 1.0) * 2.0, timestamp).await; } } -async fn publich_rollout(producer: &Producer, t: f32) { - let now = SystemTime::now(); - let d = now.duration_since(UNIX_EPOCH).expect("Error getting time"); +async fn publich_rollout(producer: &Producer, t: f32, timestamp: u64) { let temperature = Temperature { celcius: t, - timestamp: d.as_secs(), + timestamp, }; producer.produce(temperature); }