zepyr: sys: sync: Mutex/Condvar

Update the sys Mutex/Condvar to use the new object initialization, and
update the uses.  This requires a few fixes:

- The SimpleTls no longer needs the StaticTls helper, and we can just
  use a plain global static Mutex.
- Bounded channels add items to a free queue during init.  To avoid
  moving this queue after items have been inserted, we place it in a
  box.  Given that the data of bounded queues is also boxed, this isn't
  really a concern.

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

Author: d3zd3z

zephyr/src/simpletls.rs

@@ -5,14 +5,9 @@
 
 extern crate alloc;
 
-use core::{ptr, sync::atomic::Ordering};
-
-use alloc::boxed::Box;
 use alloc::vec::Vec;
 use zephyr_sys::{k_current_get, k_thread};
 
-use crate::sync::{atomic::AtomicPtr, Mutex};
-
 /// A container for simple thread local storage.
 ///
 /// This will maintain a mapping between Zephyr threads and a value of type T.  Entries will have to
@@ -55,70 +50,3 @@ impl<T: Copy + Send> SimpleTls<T> {
             .map(|pos| self.map[pos].1)
     }
 }
-
-/// A helper to safely use these with static.
-///
-/// The StaticTls type has a constant constructor, and the same insert and get methods as the
-/// underlying SimpleTls, with support for initializing the Mutex as needed.
-// TODO: This should eventually make it into a more general lazy mechanism.
-pub struct StaticTls<T: Copy + Send> {
-    /// The container for the data.
-    ///
-    /// The AtomicPtr is either Null, or contains a raw pointer to the underlying Mutex holding the
-    /// data.
-    data: AtomicPtr<Mutex<SimpleTls<T>>>,
-}
-
-impl<T: Copy + Send> StaticTls<T> {
-    /// Create a new StaticTls that is empty.
-    pub const fn new() -> Self {
-        Self {
-            data: AtomicPtr::new(ptr::null_mut()),
-        }
-    }
-
-    /// Get the underlying Mutex out of the data, initializing it with an empty type if necessary.
-    fn get_inner(&self) -> &Mutex<SimpleTls<T>> {
-        let data = self.data.fetch_update(
-            // TODO: These orderings are likely stronger than necessary.
-            Ordering::SeqCst,
-            Ordering::SeqCst,
-            |ptr| {
-                if ptr.is_null() {
-                    // For null, we need to allocate a new one.
-                    let data = Box::new(Mutex::new(SimpleTls::new()));
-                    Some(Box::into_raw(data))
-                } else {
-                    // If there was already a value, just use it.
-                    None
-                }
-            },
-        );
-        let data = match data {
-            Ok(_) => {
-                // If the update stored something, it unhelpfully returns the old value, which was
-                // the null pointer.  Since the pointer will only ever be updated once, it is safe
-                // to use a relaxed load here.
-                self.data.load(Ordering::Relaxed)
-            }
-            // If there was already a pointer, that is what we want.
-            Err(ptr) => ptr,
-        };
-
-        // SAFETY: The stored data was updated at most once, by the above code, and we now have a
-        // pointer to a valid leaked box holding the data.
-        unsafe { &*data }
-    }
-
-    /// Insert a new association into the StaticTls.
-    pub fn insert(&self, thread: *const k_thread, data: T) {
-        let inner = self.get_inner();
-        inner.lock().unwrap().insert(thread, data);
-    }
-
-    /// Lookup the data associated with a given thread.
-    pub fn get(&self) -> Option<T> {
-        let inner = self.get_inner();
-        inner.lock().unwrap().get()
-    }
-}

zephyr/src/sync/channel.rs

@@ -569,8 +569,11 @@ struct Bounded<T> {
     /// The UnsafeCell is needed to indicate that this data is handled outside of what Rust is aware
     /// of.  MaybeUninit allows us to create this without allocation.
     _slots: Pin<Box<[Slot<T>]>>,
-    /// The free queue, holds messages that aren't be used.
-    free: Queue,
+    /// The free queue, holds messages that aren't be used.  The free queue has to be in a box,
+    /// because it cannot move after the constructor runs.  The chan is fine to wait until first
+    /// use, when the object has settled.  As we are also boxing the messages, this isn't really
+    /// that costly.
+    free: Box<Queue>,
     /// The channel queue.  These are messages that have been sent and are waiting to be received.
     chan: Queue,
 }
@@ -582,7 +585,7 @@ impl<T> Bounded<T> {
             .collect();
         let slots = Box::into_pin(slots);
 
-        let free = Queue::new();
+        let free = Box::new(Queue::new());
         let chan = Queue::new();
 
         // Add each of the boxes to the free list.

zephyr/src/sync/mutex.rs

@@ -98,9 +98,8 @@ impl<T> Mutex<T> {
     }
 
     /// Construct a new Mutex, dynamically allocating the underlying sys Mutex.
-    #[cfg(CONFIG_RUST_ALLOC)]
-    pub fn new(t: T) -> Mutex<T> {
-        Mutex::new_from(t, sys::Mutex::new().unwrap())
+    pub const fn new(t: T) -> Mutex<T> {
+        Mutex::new_from(t, sys::Mutex::new())
     }
 }
 
@@ -164,7 +163,9 @@ impl<T: ?Sized> DerefMut for MutexGuard<'_, T> {
 impl<T: ?Sized> Drop for MutexGuard<'_, T> {
     #[inline]
     fn drop(&mut self) {
-        self.lock.inner.unlock().unwrap();
+        if let Err(e) = self.lock.inner.unlock() {
+            panic!("Problem unlocking MutexGuard in drop: {:?}", e);
+        }
     }
 }
 
@@ -195,7 +196,7 @@ impl Condvar {
     /// Construct a new Condvar, dynamically allocating the underlying Zephyr `k_condvar`.
     #[cfg(CONFIG_RUST_ALLOC)]
     pub fn new() -> Condvar {
-        Condvar::new_from(sys::Condvar::new().unwrap())
+        Condvar::new_from(sys::Condvar::new())
     }
 
     /// Blocks the current thread until this conditional variable receives a notification.

zephyr/src/sys/sync/mutex.rs

@@ -8,7 +8,7 @@
 //!
 //! [`object`]: crate::object
 
-use crate::object::{Fixed, StaticKernelObject, Wrapped};
+use crate::object::{ObjectInit, StaticKernelObject, ZephyrObject};
 use crate::sys::K_FOREVER;
 use crate::{
     error::{to_result_void, Result},
@@ -19,8 +19,6 @@ use crate::{
     time::Timeout,
 };
 use core::fmt;
-#[cfg(CONFIG_RUST_ALLOC)]
-use core::mem;
 
 /// A Zephyr `k_mutux` usable from safe Rust code.
 ///
@@ -46,20 +44,15 @@ use core::mem;
 /// [`sync::Mutex`]: http://example.com/TODO
 pub struct Mutex {
     /// The raw Zephyr mutex.
-    item: Fixed<k_mutex>,
+    item: ZephyrObject<k_mutex>,
 }
 
 impl Mutex {
     /// Create a new Mutex in an unlocked state.
     ///
     /// Create a new dynamically allocated Mutex.  The Mutex can only be used from system threads.
-    #[cfg(CONFIG_RUST_ALLOC)]
-    pub fn new() -> Result<Mutex> {
-        let item: Fixed<k_mutex> = Fixed::new(unsafe { mem::zeroed() });
-        unsafe {
-            to_result_void(k_mutex_init(item.get()))?;
-        }
-        Ok(Mutex { item })
+    pub const fn new() -> Mutex {
+        Mutex { item: <ZephyrObject<k_mutex>>::new_raw() }
     }
 
     /// Lock a Zephyr Mutex.
@@ -84,6 +77,15 @@ impl Mutex {
     }
 }
 
+impl ObjectInit<k_mutex> for ZephyrObject<k_mutex> {
+    fn init(item: *mut k_mutex) {
+        // SAFETY: ZephyrObject handles initialization and move prevention.
+        unsafe {
+            k_mutex_init(item);
+        }
+    }
+}
+
 /// A static Zephyr `k_mutex`
 ///
 /// This is intended to be used from within the `kobj_define!` macro.  It declares a static
@@ -101,35 +103,18 @@ unsafe impl Send for Mutex {}
 unsafe impl Sync for StaticMutex {}
 unsafe impl Send for StaticMutex {}
 
-impl Wrapped for StaticKernelObject<k_mutex> {
-    type T = Mutex;
-
-    /// Mutex initializers take no argument.
-    type I = ();
-
-    fn get_wrapped(&self, _arg: Self::I) -> Mutex {
-        let ptr = self.value.get();
-        unsafe {
-            k_mutex_init(ptr);
-        }
-        Mutex {
-            item: Fixed::Static(ptr),
-        }
-    }
-}
-
 impl fmt::Debug for Mutex {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "sys::Mutex {:?}", self.item.get())
+        // SAFETY: Address is just gotten to print for diagnostics.
+        write!(f, "sys::Mutex {:?}", unsafe { self.item.get() })
     }
 }
 
 /// A Condition Variable
 ///
 /// Lightweight wrappers for Zephyr's `k_condvar`.
 pub struct Condvar {
-    /// The underlying `k_condvar`.
-    item: Fixed<k_condvar>,
+    item: ZephyrObject<k_condvar>,
 }
 
 #[doc(hidden)]
@@ -144,13 +129,8 @@ impl Condvar {
     /// Create a new Condvar.
     ///
     /// Create a new dynamically allocated Condvar.  The Condvar can only be used from system threads.
-    #[cfg(CONFIG_RUST_ALLOC)]
-    pub fn new() -> Result<Condvar> {
-        let item: Fixed<k_condvar> = Fixed::new(unsafe { mem::zeroed() });
-        unsafe {
-            to_result_void(k_condvar_init(item.get()))?;
-        }
-        Ok(Condvar { item })
+    pub const fn new() -> Condvar {
+        Condvar { item: <ZephyrObject<k_condvar>>::new_raw() }
     }
 
     /// Wait for someone else using this mutex/condvar pair to notify.
@@ -184,25 +164,18 @@ impl Condvar {
     }
 }
 
-impl fmt::Debug for Condvar {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "sys::Condvar {:?}", self.item.get())
+impl ObjectInit<k_condvar> for ZephyrObject<k_condvar> {
+    fn init(item: *mut k_condvar) {
+        // SAFETY: ZephyrObject handles initialization and move prevention.
+        unsafe {
+            k_condvar_init(item);
+        }
     }
 }
 
-impl Wrapped for StaticCondvar {
-    type T = Condvar;
-
-    /// Condvar initializers take no argument.
-    type I = ();
-
-    fn get_wrapped(&self, _arg: Self::I) -> Condvar {
-        let ptr = self.value.get();
-        unsafe {
-            k_condvar_init(ptr);
-        }
-        Condvar {
-            item: Fixed::Static(ptr),
-        }
+impl fmt::Debug for Condvar {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        // SAFETY: Just getting the address to print.
+        write!(f, "sys::Condvar {:?}", unsafe { self.item.get() })
     }
 }

zephyr/src/work.rs

@@ -194,7 +194,7 @@ use zephyr_sys::{
 };
 
 use crate::{
-    error::to_result_void, kio::ContextExt, object::Fixed, simpletls::StaticTls, sync::Arc,
+    error::to_result_void, kio::ContextExt, object::Fixed, simpletls::SimpleTls, sync::{Arc, Mutex},
     sys::thread::ThreadStack, time::Timeout,
 };
 
@@ -275,7 +275,7 @@ impl WorkQueueBuilder {
             // SAFETY: This associates the workqueue with the thread ID that runs it.  The thread is
             // a pointer into this work item, which will not move, because of the Fixed.
             let this = &mut *item.get();
-            WORK_QUEUES.insert(&this.thread, WorkQueueRef(item.get()));
+            WORK_QUEUES.lock().unwrap().insert(&this.thread, WorkQueueRef(item.get()));
 
             // SAFETY: Start work queue thread.  The main issue here is that the work queue cannot
             // be deallocated once the thread has started.  We enforce this by making Drop panic.
@@ -340,7 +340,7 @@ impl Drop for WorkQueue {
 ///
 /// This is a little bit messy as we don't have a lazy mechanism, so we have to handle this a bit
 /// manually right now.
-static WORK_QUEUES: StaticTls<WorkQueueRef> = StaticTls::new();
+static WORK_QUEUES: Mutex<SimpleTls<WorkQueueRef>> = Mutex::new(SimpleTls::new());
 
 /// For the queue mapping, we need a simple wrapper around the underlying pointer, one that doesn't
 /// implement stop.
@@ -353,7 +353,7 @@ unsafe impl Sync for WorkQueueRef {}
 
 /// Retrieve the current work queue, if we are running within one.
 pub fn get_current_workq() -> Option<*mut k_work_q> {
-    WORK_QUEUES.get().map(|wq| wq.0)
+    WORK_QUEUES.lock().unwrap().get().map(|wq| wq.0)
 }
 
 /// A Rust wrapper for `k_poll_signal`.