From 20cf519b86476725bf5064c20976f03831291080 Mon Sep 17 00:00:00 2001 From: Adam Sasine Date: Wed, 9 Sep 2026 10:52:37 -0700 Subject: [PATCH 1/3] Serialize two-device CFU updates Await the first of two splitter targets before starting the second so shared-bus CFU state machines cannot interleave. Cover ordering and failure behavior while retaining three- and four-target concurrency. Assisted-by: GitHub Copilot:gpt-5.6-sol Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- Cargo.lock | 1 + cfu-service/Cargo.toml | 5 ++ cfu-service/src/splitter.rs | 142 ++++++++++++++++++++++++++++++++++-- 3 files changed, 142 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2e3a35f4b..7052cd8b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -320,6 +320,7 @@ checksum = "2fd1289c04a9ea8cb22300a459a72a385d7c73d3259e2ed7dcb2af674838cfa9" name = "cfu-service" version = "0.1.0" dependencies = [ + "critical-section", "defmt 0.3.100", "embassy-futures", "embassy-sync", diff --git a/cfu-service/Cargo.toml b/cfu-service/Cargo.toml index 811c82ea7..9dc6ad491 100644 --- a/cfu-service/Cargo.toml +++ b/cfu-service/Cargo.toml @@ -36,3 +36,8 @@ log = [ "embassy-sync/log", "embedded-cfu-protocol/log", ] + +[dev-dependencies] +critical-section = { workspace = true, features = ["std"] } +embassy-sync = { workspace = true, features = ["std"] } +embassy-time = { workspace = true, features = ["std", "generic-queue-8"] } diff --git a/cfu-service/src/splitter.rs b/cfu-service/src/splitter.rs index bbbed3d50..69e9764d1 100644 --- a/cfu-service/src/splitter.rs +++ b/cfu-service/src/splitter.rs @@ -3,7 +3,7 @@ use core::{future::Future, iter::zip}; -use embassy_futures::join::{join, join3, join4}; +use embassy_futures::join::{join3, join4}; use embedded_cfu_protocol::protocol_definitions::*; use embedded_services::{ cfu::{ @@ -197,9 +197,8 @@ impl<'a, C: Customization> Splitter<'a, C> { /// Map items in an input slice to an output slice using an async closure. /// -/// This function will execute the closure concurrently in groups up to four items at a time. -/// Four is an arbitrary but is a balance between two (easy to implement, but not very concurrent) and eight (more implementation work). -/// This will exit early and return false if any item results in `None`. +/// This function executes one item directly, two items sequentially, and three or four items concurrently. +/// It returns false if any item results in `None`. async fn map_slice_join<'i, 'o, I, O, F: Future>>( input: &'i [I], output: &'o mut [O], @@ -222,8 +221,9 @@ async fn map_slice_join<'i, 'o, I, O, F: Future>>( } } (Some((i0, o0)), Some((i1, o1)), None, None) => { - let results = join(f(i0), f(i1)).await; - if let (Some(r0), Some(r1)) = results { + let result_0 = f(i0).await; + let result_1 = f(i1).await; + if let (Some(r0), Some(r1)) = (result_0, result_1) { *o0 = r0; *o1 = r1; } else { @@ -257,3 +257,133 @@ async fn map_slice_join<'i, 'o, I, O, F: Future>>( } } } + +#[cfg(test)] +#[allow(clippy::panic)] +#[allow(clippy::unwrap_used)] +mod tests { + use core::{cell::RefCell, future::poll_fn, task::Poll}; + + use super::map_slice_join; + + #[derive(Clone, Copy, Debug, PartialEq, Eq)] + enum Event { + Started(u8), + Completed(u8), + } + + async fn yield_once() { + let mut yielded = false; + poll_fn(|cx| { + if yielded { + Poll::Ready(()) + } else { + yielded = true; + cx.waker().wake_by_ref(); + Poll::Pending + } + }) + .await; + } + + async fn assert_items_remain_concurrent() { + let input = core::array::from_fn(|item| item as u8); + let mut output = [u8::MAX; N]; + let events = RefCell::new(heapless::Vec::::new()); + + let success = map_slice_join(&input, &mut output, |item| { + let events = &events; + async move { + events.borrow_mut().push(Event::Started(*item)).unwrap(); + yield_once().await; + events.borrow_mut().push(Event::Completed(*item)).unwrap(); + Some(*item) + } + }) + .await; + + let mut expected = heapless::Vec::::new(); + for item in input { + expected.push(Event::Started(item)).unwrap(); + } + for item in input { + expected.push(Event::Completed(item)).unwrap(); + } + + assert!(success); + assert_eq!(output, input); + assert_eq!(events.into_inner(), expected); + } + + #[test] + fn two_items_run_sequentially() { + embassy_futures::block_on(async { + let input = [0, 1]; + let mut output = [0; 2]; + let events = RefCell::new(heapless::Vec::::new()); + + let success = map_slice_join(&input, &mut output, |item| { + let events = &events; + async move { + events.borrow_mut().push(Event::Started(*item)).unwrap(); + yield_once().await; + events.borrow_mut().push(Event::Completed(*item)).unwrap(); + Some(*item) + } + }) + .await; + + assert!(success); + assert_eq!(output, input); + assert_eq!( + events.into_inner().as_slice(), + [ + Event::Started(0), + Event::Completed(0), + Event::Started(1), + Event::Completed(1), + ] + ); + }); + } + + #[test] + fn second_item_runs_after_first_returns_none() { + embassy_futures::block_on(async { + let input = [0, 1]; + let mut output = [2; 2]; + let events = RefCell::new(heapless::Vec::::new()); + + let success = map_slice_join(&input, &mut output, |item| { + let events = &events; + async move { + events.borrow_mut().push(Event::Started(*item)).unwrap(); + yield_once().await; + events.borrow_mut().push(Event::Completed(*item)).unwrap(); + (*item != 0).then_some(*item) + } + }) + .await; + + assert!(!success); + assert_eq!(output, [2; 2]); + assert_eq!( + events.into_inner().as_slice(), + [ + Event::Started(0), + Event::Completed(0), + Event::Started(1), + Event::Completed(1), + ] + ); + }); + } + + #[test] + fn three_and_four_items_remain_concurrent() { + embassy_futures::block_on(async { + assert_items_remain_concurrent::<3>().await; + assert_items_remain_concurrent::<4>().await; + }); + } +} From 55b26474c416b2219576529ab34207c211450a18 Mon Sep 17 00:00:00 2001 From: Adam Sasine Date: Wed, 9 Sep 2026 11:02:54 -0700 Subject: [PATCH 2/3] Relax test assertion Only assert that all Started occurs before any Completed Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- cfu-service/src/splitter.rs | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/cfu-service/src/splitter.rs b/cfu-service/src/splitter.rs index 69e9764d1..aa7a8a85a 100644 --- a/cfu-service/src/splitter.rs +++ b/cfu-service/src/splitter.rs @@ -302,18 +302,19 @@ mod tests { }) .await; - let mut expected = heapless::Vec::::new(); - for item in input { - expected.push(Event::Started(item)).unwrap(); - } - for item in input { - expected.push(Event::Completed(item)).unwrap(); - } + let events = events.into_inner(); assert!(success); assert_eq!(output, input); - assert_eq!(events.into_inner(), expected); - } + assert_eq!(events.len(), N * 2); + + let (started, completed) = events.as_slice().split_at(N); + assert!(started.iter().all(|e| matches!(e, Event::Started(_)))); + assert!(completed.iter().all(|e| matches!(e, Event::Completed(_)))); + for item in input { + assert!(started.iter().any(|e| *e == Event::Started(item))); + assert!(completed.iter().any(|e| *e == Event::Completed(item))); + } #[test] fn two_items_run_sequentially() { From f95b6b08bb0241ec7879f3a1a89c22839e7399bf Mon Sep 17 00:00:00 2001 From: Adam Sasine Date: Wed, 9 Sep 2026 11:51:27 -0700 Subject: [PATCH 3/3] Fix concurrent splitter test syntax Close the helper after the relaxed ordering assertions and use slice contains checks required by the workspace clippy configuration. Assisted-by: GitHub Copilot:gpt-5.6-sol Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- cfu-service/src/splitter.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/cfu-service/src/splitter.rs b/cfu-service/src/splitter.rs index aa7a8a85a..83f12937f 100644 --- a/cfu-service/src/splitter.rs +++ b/cfu-service/src/splitter.rs @@ -312,9 +312,10 @@ mod tests { assert!(started.iter().all(|e| matches!(e, Event::Started(_)))); assert!(completed.iter().all(|e| matches!(e, Event::Completed(_)))); for item in input { - assert!(started.iter().any(|e| *e == Event::Started(item))); - assert!(completed.iter().any(|e| *e == Event::Completed(item))); + assert!(started.contains(&Event::Started(item))); + assert!(completed.contains(&Event::Completed(item))); } + } #[test] fn two_items_run_sequentially() {