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..83f12937f 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,135 @@ 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 events = events.into_inner(); + + assert!(success); + assert_eq!(output, input); + 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.contains(&Event::Started(item))); + assert!(completed.contains(&Event::Completed(item))); + } + } + + #[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; + }); + } +}