samples: work-philosphers: Async Semaphore based solution

d3zd3z · d3zd3z · commit 0dba021cb57c · 2025-02-04T14:18:28.000-07:00
Implement the async Semaphore based solution to the philospher problem.
This demonstrates a few things:

- The use of multiple async tasks created with `spawn` and `spawn_local`.
- How to use `spawn_local` to allow a group of tasks all running on the
  same worker to use Rc instead of Arc for sharing.
- A check that completed work (from a spawn, after calling join or
  join_async) drops all of its data properly.

Signed-off-by: David Brown &lt;david.brown@linaro.org&gt;
diff --git a/samples/work-philosophers/src/async_sem.rs b/samples/work-philosophers/src/async_sem.rs
@@ -0,0 +1,113 @@
+//! Async Semaphore based demo
+//!
+//! This implementation on the dining philosopher problem uses Zephyr semaphores to represent the
+//! forks.  Each philosopher dines as per the algorithm a number of times, and when the are all
+//! finished, the test is considered successful.  Deadlock will result in the primary thread not
+//! completing.
+//!
+//! Notably, this uses Rc and RefCell along with spawn_local to demonstrate that multiple async
+//! 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 crate::{get_random_delay, Stats, NUM_PHIL};
+
+/// Number of iterations of each philospher.
+///
+/// Should be long enough to exercise the test, but too
+/// long and the test will timeout.  The delay calculated will randomly be between 25 and 775, and
+/// 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 {
+    // 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).unwrap()))
+        .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();
+
+    // Wait for them all to finish.
+    for p in phils {
+        p.join_async().await;
+    }
+
+    // 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());
+
+    // 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()
+}
+
+/// Simulate a single philospher.
+///
+/// The forks must be ordered with the first fork having th lowest number, otherwise this will
+/// 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>>) {
+    for i in 0..EAT_COUNT {
+        // Acquire the forks.
+        // printkln!("Child {n} take left fork");
+        forks[0].take_async(Forever).await.unwrap();
+        // printkln!("Child {n} take right fork");
+        forks[1].take_async(Forever).await.unwrap();
+
+        // printkln!("Child {n} eating");
+        let delay = get_random_delay(n, 25);
+        sleep(delay).await;
+        stats.borrow_mut().record_eat(n, delay);
+
+        // Release the forks.
+        // printkln!("Child {n} giving up forks");
+        forks[1].give();
+        forks[0].give();
+
+        let delay = get_random_delay(n, 25);
+        sleep(delay).await;
+        stats.borrow_mut().record_think(n, delay);
+
+        printkln!("Philospher {n} finished eating time {i}");
+    }
+}
diff --git a/samples/work-philosophers/src/lib.rs b/samples/work-philosophers/src/lib.rs
@@ -2,213 +2,76 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #![no_std]
-
 // Cargo tries to detect configs that have typos in them.  Unfortunately, the Zephyr Kconfig system
 // uses a large number of Kconfigs and there is no easy way to know which ones might conceivably be
 // valid.  This prevents a warning about each cfg that is used.
 #![allow(unexpected_cfgs)]
 
 extern crate alloc;
 
-use alloc::{boxed::Box, ffi::CString, format};
-use zephyr::{kobj_define, printkln, sync::Arc, sys::sync::Semaphore, time::{Duration, Forever}, work::{futures::sleep, Signal, WorkQueue, WorkQueueBuilder}};
+use zephyr::{
+    kio::spawn,
+    kobj_define, printkln,
+    sync::Arc,
+    sys::uptime_get,
+    time::{Duration, Tick},
+    work::WorkQueueBuilder,
+};
+
+mod async_sem;
 
 /// How many philosophers.  There will be the same number of forks.
-const _NUM_PHIL: usize = 6;
+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 use one of several different underlying mechanism to support this cooperation.
-
-// This example uses dynamic dispatch to allow multiple implementations.  The intent is to be able
-// to periodically shut down all of the philosphers and start them up with a differernt sync
-// mechanism.  This isn't implemented yet.
-
-/// The philosophers use a fork synchronization mechanism.  Essentially, this is 6 locks, and will be
-/// implemented in a few different ways to demonstrate/test different mechanmism in Rust.  All of
-/// them implement The ForkSync trait which provides this mechanism.
-/*
-trait ForkSync: core::fmt::Debug + Sync + Send {
-    /// Take the given fork.  The are indexed the same as the philosopher index number.  This will
-    /// block until the fork is released.
-    fn take(&self, index: usize);
-
-    /// Release the given fork.  Index is the same as take.
-    fn release(&self, index: usize);
-}
-*/
+// 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!("Hello world from Rust on {}",
-              zephyr::kconfig::CONFIG_BOARD);
+    printkln!(
+        "Async/work-queue dining philosophers{}",
+        zephyr::kconfig::CONFIG_BOARD
+    );
     printkln!("Time tick: {}", zephyr::time::SYS_FREQUENCY);
 
     // Create the work queue to run this.
-    let worker = Box::new(WorkQueueBuilder::new()
-                          .set_priority(1)
-                          .start(WORK_STACK.init_once(()).unwrap()));
+    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()));
+    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());
 
-    // Create a semaphore the phil thread can wait for, part way through its work.
-    let sem = Arc::new(Semaphore::new(0, 1).unwrap());
-
-    let th = phil_thread(0, sem.clone(), lower_worker.clone());
-    printkln!("Size of phil thread: {}", core::mem::size_of_val(&th));
-    // TODO: How to do as much on the stack as we can, for now, don't worry too much.
-    // let mut th = pin!(th);
-
-    let th = zephyr::kio::spawn(th, &worker, c"phil-worker");
-
-    // As we no longer use the worker, it is important that it never be dropped.  Leaking the box
-    // ensures that it will not be deallocated.
-    Box::leak(worker);
-
-    // Sleep a bit to allow the worker to run to where it is waiting.
-    zephyr::time::sleep(Duration::millis_at_least(1_000));
-    printkln!("Giving the semaphore");
-    sem.give();
-
-    let result = th.join();
-    printkln!("th result: {:?}", result);
+    // 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();
+    printkln!("Done with 'async-sem' test");
+    stats.show();
 
-    /*
-    // TODO: The allocated Arc doesn't work on all Zephyr platforms, but need to make our own copy
-    // of alloc::task
-    let waker = Arc::new(PWaker).into();
-    let mut cx = Context::from_waker(&waker);
-
-    // Run the future to completion.
-    loop {
-        match th.as_mut().poll(&mut cx) {
-            Poll::Pending => todo!(),
-            Poll::Ready(_) => break,
-        }
-    }
-    */
     printkln!("All threads done");
-
-    /*
-    let stats = Arc::new(Mutex::new_from(Stats::default(), STAT_MUTEX.init_once(()).unwrap()));
-
-    let syncers = get_syncer();
-
-    printkln!("Pre fork");
-
-    for (i, syncer) in (0..NUM_PHIL).zip(syncers.into_iter()) {
-        let child_stat = stats.clone();
-        let thread = PHIL_THREADS[i].init_once(PHIL_STACKS[i].init_once(()).unwrap()).unwrap();
-        thread.spawn(move || {
-            phil_thread(i, syncer, child_stat);
-        });
-    }
-
-    let delay = Duration::secs_at_least(10);
-    loop {
-        // Periodically, printout the stats.
-        zephyr::time::sleep(delay);
-        stats.lock().unwrap().show();
-    }
-    */
-}
-
-async fn phil_thread(n: usize, sem: Arc<Semaphore>, lower_thread: Arc<WorkQueue>) -> usize {
-    printkln!("Child {} started", n);
-    show_it(&sem).await;
-    sleep(Duration::millis_at_least(1000)).await;
-    printkln!("Child {} done sleeping", n);
-
-    // Lastly fire off something on the other worker thread.
-    let sig = Arc::new(Signal::new().unwrap());
-    let sig2 = sig.clone();
-
-    // To help with debugging, we can name these, but it must be a `&'static Cstr`.  We'll build a
-    // CString (on the heap), and then leak it to be owned by the work.  The leaking while keeping
-    // the `&'static CStr` is a bit messy.
-    let name = CString::new(format!("phil-{}", n)).unwrap();
-    let name = Box::leak(name.into_boxed_c_str());
-
-    // It _should_ be safe to drop the child, as it uses an internal clone of the Arc to keep it
-    // around as long as there is a worker.
-    let _child = Box::new(zephyr::kio::spawn(async move {
-        sleep(Duration::millis_at_least(800)).await;
-        sig2.raise(12345).unwrap();
-    }, &lower_thread, name));
-
-    // Wait for the signal.
-    let num = sig.wait_async(Forever).await.unwrap();
-    printkln!("Signaled value: {}", num);
-
-    42
-}
-
-async fn show_it(sem: &Semaphore) {
-    for i in 0..10 {
-        sleep(Duration::millis_at_least(1)).await;
-        printkln!("Tick: {i}");
-
-        // Wait after 5 for the semaphore to be signaled.
-        if i == 4 {
-            printkln!("Waiting for semaphore");
-            sem.take_async(Forever).await.unwrap();
-            printkln!("Done waiting");
-        }
-    }
 }
 
 kobj_define! {
     static WORK_STACK: ThreadStack<WORK_STACK_SIZE>;
     static LOWER_WORK_STACK: ThreadStack<WORK_STACK_SIZE>;
 }
 
-/*
-fn phil_thread(n: usize, syncer: Arc<dyn ForkSync>, stats: Arc<Mutex<Stats>>) {
-    printkln!("Child {} started: {:?}", n, syncer);
-
-    // Determine our two forks.
-    let forks = if n == NUM_PHIL - 1 {
-        // Per Dijkstra, the last phyilosopher needs to reverse forks, or we deadlock.
-        (0, n)
-    } else {
-        (n, n+1)
-    };
-
-    loop {
-        {
-            // printkln!("Child {} hungry", n);
-            // printkln!("Child {} take left fork", n);
-            syncer.take(forks.0);
-            // printkln!("Child {} take right fork", n);
-            syncer.take(forks.1);
-
-            let delay = get_random_delay(n, 25);
-            // printkln!("Child {} eating ({} ms)", n, delay);
-            sleep(delay);
-            stats.lock().unwrap().record_eat(n, delay);
-
-            // Release the forks.
-            // printkln!("Child {} giving up forks", n);
-            syncer.release(forks.1);
-            syncer.release(forks.0);
-
-            let delay = get_random_delay(n, 25);
-            // printkln!("Child {} thinking ({} ms)", n, delay);
-            sleep(delay);
-            stats.lock().unwrap().record_think(n, delay);
-        }
-    }
-}
-
 /// 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;
@@ -241,14 +104,11 @@ impl Stats {
     }
 
     fn show(&self) {
-        printkln!("c:{:?}, e:{:?}, t:{:?}", self.count, self.eating, self.thinking);
+        printkln!(
+            "c:{:?}, e:{:?}, t:{:?}",
+            self.count,
+            self.eating,
+            self.thinking
+        );
     }
 }
-
-kobj_define! {
-    static PHIL_THREADS: [StaticThread; NUM_PHIL];
-    static PHIL_STACKS: [ThreadStack<PHIL_STACK_SIZE>; NUM_PHIL];
-
-    static STAT_MUTEX: StaticMutex;
-}
-*/
