samples: async-philosophers: Convert to general async

Convert away from the Zephyr-specific async to standard Rust async.
This uses the Embassy crates for time and synchronization.  All tasks
are spawned within a single executor that runs in the main thread.

Signed-off-by: David Brown <[email protected]>

Author: d3zd3z

samples/async-philosophers/Cargo.toml

@@ -13,7 +13,14 @@ license = "Apache-2.0 or MIT"
 crate-type = ["staticlib"]
 
 [dependencies]
-zephyr = "0.1.0"
+zephyr = { version = "0.1.0", features = ["time-driver", "executor-zephyr"] }
+static_cell = "2.1"
+
+embassy-executor = { version = "0.7.0", features = ["log", "task-arena-size-2048"] }
+embassy-sync = "0.6.2"
+
+# TODO: Don't hardcode the tickrate.
+embassy-time = { version = "0.4.0", features = ["tick-hz-10_000"] }
 
 # Dependencies that are used by build.rs.
 [build-dependencies]

samples/async-philosophers/prj.conf

@@ -7,6 +7,7 @@ CONFIG_MAIN_STACK_SIZE=8192
 CONFIG_SYSTEM_WORKQUEUE_STACK_SIZE=4096
 
 CONFIG_POLL=y
+CONFIG_STACK_CANARIES=y
 
 # CONFIG_DEBUG=y
 

samples/async-philosophers/src/async_sem.rs

@@ -9,17 +9,24 @@
 //! tasks run on the same worker do not need Send.  It is just important that write operations on
 //! the RefCell do not `.await` or a panic is likely.
 
-use core::cell::RefCell;
-
-use alloc::{rc::Rc, vec::Vec};
-use zephyr::{
-    kio::{sleep, spawn_local},
-    printkln,
-    sys::sync::Semaphore,
-    time::Forever,
+use embassy_executor::Spawner;
+use embassy_sync::{
+    blocking_mutex::raw::CriticalSectionRawMutex,
+    mutex::Mutex,
+    semaphore::{FairSemaphore, Semaphore},
 };
+use embassy_time::Timer;
+use zephyr::{printkln, sync::Arc};
 
-use crate::{get_random_delay, Stats, NUM_PHIL};
+use crate::{get_random_delay, ResultSignal, Stats, NUM_PHIL};
+
+type ESemaphore = FairSemaphore<CriticalSectionRawMutex, NUM_PHIL>;
+
+/// The semaphores for the forks.
+static FORKS: [ESemaphore; NUM_PHIL] = [const { ESemaphore::new(1) }; NUM_PHIL];
+
+/// The semaphore to wait for them all to finish.
+static DONE_SEM: ESemaphore = ESemaphore::new(0);
 
 /// Number of iterations of each philospher.
 ///
@@ -28,56 +35,27 @@ use crate::{get_random_delay, Stats, NUM_PHIL};
 /// there are two waits, so typically, each "eat" will take about a second.
 const EAT_COUNT: usize = 10;
 
-pub async fn phil() -> Stats {
-    // It is a little tricky to be able to use local workers.  We have to have this nested thread
-    // that waits.  This is because the Future from `local_phil()` does not implement Send, since it
-    // waits for the philosophers, which are not Send.  However, this outer async function does not
-    // hold onto any data that is not send, and therefore will be Send.  Fortunately, this extra
-    // Future is very lightweight.
-    spawn_local(local_phil(), c"phil_wrap").join_async().await
-}
-
-async fn local_phil() -> Stats {
+#[embassy_executor::task]
+pub async fn phil(spawner: Spawner, stats_sig: &'static ResultSignal) {
     // Our overall stats.
-    let stats = Rc::new(RefCell::new(Stats::default()));
-
-    // One fork for each philospher.
-    let forks: Vec<_> = (0..NUM_PHIL)
-        .map(|_| Rc::new(Semaphore::new(1, 1)))
-        .collect();
-
-    // Create all of the philosphers
-    let phils: Vec<_> = (0..NUM_PHIL)
-        .map(|i| {
-            // Determine the two forks.  The forks are paired with each philosopher taking the fork of
-            // their number, and the next on, module the size of the ring.  However, for the last case,
-            // we need to swap the forks used, it is necessary to obey a strict ordering of the locks to
-            // avoid deadlocks.
-            let forks = if i == NUM_PHIL - 1 {
-                [forks[0].clone(), forks[i].clone()]
-            } else {
-                [forks[i].clone(), forks[i + 1].clone()]
-            };
-
-            spawn_local(one_phil(forks, i, stats.clone()), c"phil")
-        })
-        .collect();
+    let stats = Arc::new(Mutex::new(Stats::default()));
 
-    // Wait for them all to finish.
-    for p in phils {
-        p.join_async().await;
+    // Spawn off each philosopher.
+    for i in 0..NUM_PHIL {
+        let forks = if i == NUM_PHIL - 1 {
+            [&FORKS[0], &FORKS[i]]
+        } else {
+            [&FORKS[i], &FORKS[i + 1]]
+        };
+
+        spawner.spawn(one_phil(forks, i, stats.clone())).unwrap();
     }
 
-    // Leak the stats as a test.
-    // Uncomment this to test that the expect below does truly detect a missed drop.
-    // let _ = Rc::into_raw(stats.clone());
+    // Wait for them all to finish.
+    DONE_SEM.acquire(NUM_PHIL).await.unwrap();
 
-    // At this point, all of the philosphers should have dropped their stats ref, and we should be
-    // able to turn stats back into it's value.
-    // This tests that completed work does drop the future.
-    Rc::into_inner(stats)
-        .expect("Failure: a philospher didn't drop it's future")
-        .into_inner()
+    // Send the stats back.
+    stats_sig.signal(stats);
 }
 
 /// Simulate a single philospher.
@@ -86,28 +64,35 @@ async fn local_phil() -> Stats {
 /// likely deadlock.
 ///
 /// This will run for EAT_COUNT times, and then return.
-async fn one_phil(forks: [Rc<Semaphore>; 2], n: usize, stats: Rc<RefCell<Stats>>) {
+#[embassy_executor::task(pool_size = NUM_PHIL)]
+async fn one_phil(
+    forks: [&'static ESemaphore; 2],
+    n: usize,
+    stats: Arc<Mutex<CriticalSectionRawMutex, Stats>>,
+) {
     for i in 0..EAT_COUNT {
         // Acquire the forks.
         // printkln!("Child {n} take left fork");
-        forks[0].take_async(Forever).await.unwrap();
+        forks[0].acquire(1).await.unwrap().disarm();
         // printkln!("Child {n} take right fork");
-        forks[1].take_async(Forever).await.unwrap();
+        forks[1].acquire(1).await.unwrap().disarm();
 
         // printkln!("Child {n} eating");
         let delay = get_random_delay(n, 25);
-        sleep(delay).await;
-        stats.borrow_mut().record_eat(n, delay);
+        Timer::after(delay).await;
+        stats.lock().await.record_eat(n, delay);
 
         // Release the forks.
         // printkln!("Child {n} giving up forks");
-        forks[1].give();
-        forks[0].give();
+        forks[1].release(1);
+        forks[0].release(1);
 
         let delay = get_random_delay(n, 25);
-        sleep(delay).await;
-        stats.borrow_mut().record_think(n, delay);
+        Timer::after(delay).await;
+        stats.lock().await.record_think(n, delay);
 
         printkln!("Philospher {n} finished eating time {i}");
     }
+
+    DONE_SEM.release(1);
 }

samples/async-philosophers/src/lib.rs

@@ -9,76 +9,59 @@
 
 extern crate alloc;
 
-use zephyr::{
-    kio::spawn,
-    kobj_define, printkln,
-    sync::Arc,
-    sys::uptime_get,
-    time::{Duration, Tick},
-    work::WorkQueueBuilder,
-};
+use embassy_executor::Spawner;
+use embassy_sync::{blocking_mutex::raw::CriticalSectionRawMutex, mutex::Mutex, signal::Signal};
+use embassy_time::Duration;
+use static_cell::StaticCell;
+use zephyr::{embassy::Executor, printkln, sync::Arc, sys::uptime_get};
 
 mod async_sem;
 
 /// How many philosophers.  There will be the same number of forks.
 const NUM_PHIL: usize = 6;
 
-/// Size of the stack for the work queue.
-const WORK_STACK_SIZE: usize = 2048;
-
 // The dining philosophers problem is a simple example of cooperation between multiple threads.
 // This implementation demonstrates a few ways that Zephyr's work-queues can be used to simulate
 // this problem.
 
 #[no_mangle]
 extern "C" fn rust_main() {
-    printkln!(
-        "Async/work-queue dining philosophers{}",
-        zephyr::kconfig::CONFIG_BOARD
-    );
+    printkln!("Async dining philosophers{}", zephyr::kconfig::CONFIG_BOARD);
     printkln!("Time tick: {}", zephyr::time::SYS_FREQUENCY);
 
-    // Create the work queue to run this.
-    let worker = Arc::new(
-        WorkQueueBuilder::new()
-            .set_priority(1)
-            .start(WORK_STACK.init_once(()).unwrap()),
-    );
-
-    // In addition, create a lower priority worker.
-    let lower_worker = Arc::new(
-        WorkQueueBuilder::new()
-            .set_priority(5)
-            .start(LOWER_WORK_STACK.init_once(()).unwrap()),
-    );
-
-    // It is important that work queues are not dropped, as they are persistent objects in the
-    // Zephyr world.
-    let _ = Arc::into_raw(lower_worker.clone());
-    let _ = Arc::into_raw(worker.clone());
+    let executor = EXECUTOR.init(Executor::new());
+    executor.run(|spawner| {
+        spawner.spawn(main(spawner)).unwrap();
+    })
+}
+
+static EXECUTOR: StaticCell<Executor> = StaticCell::new();
+
+type ResultSignal = Signal<CriticalSectionRawMutex, Arc<Mutex<CriticalSectionRawMutex, Stats>>>;
+static RESULT_SIGNAL: ResultSignal = Signal::new();
 
+#[embassy_executor::task]
+async fn main(spawner: Spawner) -> () {
     // First run the async semaphore based one.
     printkln!("Running 'async-sem' test");
-    let handle = spawn(async_sem::phil(), &worker, c"async-sem");
-    let stats = handle.join();
+    spawner
+        .spawn(async_sem::phil(spawner, &RESULT_SIGNAL))
+        .unwrap();
+
+    let stats = RESULT_SIGNAL.wait().await;
     printkln!("Done with 'async-sem' test");
-    stats.show();
+    stats.lock().await.show();
 
     printkln!("All threads done");
 }
 
-kobj_define! {
-    static WORK_STACK: ThreadStack<WORK_STACK_SIZE>;
-    static LOWER_WORK_STACK: ThreadStack<WORK_STACK_SIZE>;
-}
-
 /// Get a random delay, based on the ID of this user, and the current uptime.
 fn get_random_delay(id: usize, period: usize) -> Duration {
-    let tick = (uptime_get() & (usize::MAX as i64)) as usize;
-    let delay = (tick / 100 * (id + 1)) & 0x1f;
+    let tick = (uptime_get() & (usize::MAX as i64)) as u64;
+    let delay = (tick / 100 * (id as u64 + 1)) & 0x1f;
 
     // Use one greater to be sure to never get a delay of zero.
-    Duration::millis_at_least(((delay + 1) * period) as Tick)
+    Duration::from_millis((delay + 1) * (period as u64))
 }
 
 /// Instead of just printint out so much information that the data just scolls by, gather
@@ -95,11 +78,11 @@ struct Stats {
 
 impl Stats {
     fn record_eat(&mut self, index: usize, time: Duration) {
-        self.eating[index] += time.to_millis();
+        self.eating[index] += time.as_millis();
     }
 
     fn record_think(&mut self, index: usize, time: Duration) {
-        self.thinking[index] += time.to_millis();
+        self.thinking[index] += time.as_millis();
         self.count[index] += 1;
     }