diff --git a/services/ticktimer-server/src/api.rs b/services/ticktimer-server/src/api.rs index fd4f40b97..5bd272cce 100644 --- a/services/ticktimer-server/src/api.rs +++ b/services/ticktimer-server/src/api.rs @@ -15,8 +15,14 @@ pub(crate) enum Opcode { /// force a WDT update PingWdt = 4, - /// return the version string of Xous. We bury it here because this is a small, lightweight server we can rebuild on every run. + /// Return the version string of Xous. We bury it here because this is a small, lightweight server we can rebuild on every run. GetVersion = 5, + + /// Lock the given Mutex, blocking if it is already locked + LockMutex = 6, + + /// Unlock the given Mutex + UnlockMutex = 7, } #[derive(rkyv::Archive, rkyv::Serialize, rkyv::Deserialize)] diff --git a/services/ticktimer-server/src/lib.rs b/services/ticktimer-server/src/lib.rs index 7ae3a106d..90f695cf0 100644 --- a/services/ticktimer-server/src/lib.rs +++ b/services/ticktimer-server/src/lib.rs @@ -66,11 +66,61 @@ impl Ticktimer { version: xous_ipc::String::new(), }; let mut buf = xous_ipc::Buffer::into_buf(alloc).expect("couldn't convert version request"); - buf.lend_mut(self.conn, api::Opcode::GetVersion.to_u32().unwrap()).expect("couldn't get version"); - let v = buf.to_original::().expect("couldn't revert buffer"); + buf.lend_mut(self.conn, api::Opcode::GetVersion.to_u32().unwrap()) + .expect("couldn't get version"); + let v = buf + .to_original::() + .expect("couldn't revert buffer"); String::from(v.version.as_str().unwrap()) } + + /// Lock the given Mutex. Blocks until the Mutex is locked. + /// + /// Note that Mutexes start out in a `Locked` state and move into an `Unlocked` state by calling + /// `Unlock` on their pointer. For example, the following will probably block forever: + /// + /// `TickTimer.lock_mutex(1)` + /// + /// In order to create a new Mutex, you must first `Unlock` it. For example, the following is + /// allowed: + /// + /// `TickTimer.unlock_mutex(1)` + /// `TickTimer.lock_mutex(1)` + /// `TickTimer.unlock_mutex(1)` + /// + /// Arguments: + /// + /// * mtx: A `usize` referring to the Mutex. This is probably a pointer, but can be any `usize` + pub fn lock_mutex(&self, mtx: usize) { + send_message( + self.conn, + xous::Message::new_blocking_scalar( + api::Opcode::LockMutex.to_usize().unwrap(), + mtx, + 0, + 0, + 0, + ), + ) + .expect("couldn't lock mutex"); + } + + /// Unlock the given Mutex. Does not block. If the Mutex is not locked, then it will be + /// "doubly-unlocked". That is, if you Unlock a mutex twice, then you can Lock it twice + /// without blocking. + /// + /// Arguments: + /// + /// * mtx: A `usize` referring to the Mutex. This is probably a pointer, but can be any `usize` + pub fn unlock_mutex(&self, mtx: usize) { + send_message( + self.conn, + xous::Message::new_scalar(api::Opcode::UnlockMutex.to_usize().unwrap(), mtx, 0, 0, 0), + ) + .expect("couldn't lock mutex"); + } } + use core::sync::atomic::{AtomicU32, Ordering}; static REFCOUNT: AtomicU32 = AtomicU32::new(0); impl Drop for Ticktimer { diff --git a/services/ticktimer-server/src/main.rs b/services/ticktimer-server/src/main.rs index a9f231440..601e6829c 100644 --- a/services/ticktimer-server/src/main.rs +++ b/services/ticktimer-server/src/main.rs @@ -475,7 +475,7 @@ fn xmain() -> ! { log::info!("****************************************************************"); log::info!("Welcome to Xous {}", version::SEMVER); - #[cfg(not(feature="no-timestamp"))] + #[cfg(not(feature = "no-timestamp"))] log::info!("Built on {}", version::TIMESTAMP); log::info!("****************************************************************"); @@ -495,8 +495,27 @@ fn xmain() -> ! { let xns = xous_names::XousNames::new().unwrap(); let sr_cid = xous::connect(ticktimer_server).expect("couldn't create suspend callback connection"); - let mut susres = susres::Susres::new(Some(SuspendOrder::Last), &xns, api::Opcode::SuspendResume as u32, sr_cid) - .expect("couldn't create suspend/resume object"); + let mut susres = susres::Susres::new( + Some(SuspendOrder::Last), + &xns, + api::Opcode::SuspendResume as u32, + sr_cid, + ) + .expect("couldn't create suspend/resume object"); + + use std::collections::{HashMap, HashSet, VecDeque}; + + // A list of mutexes that should be allowed to run immediately, because the + // thread they were waiting on has already unlocked the mutex. This occurs + // when a thread attempts to Lock a Mutex and fails, then gets preempted + // before it has a chance to send the `LockMutex` message to us. + let mut mutex_ready_hash: HashMap, HashSet> = HashMap::new(); + + // A list of message IDs that are waiting to lock a Mutex. These are processes + // that have attempted to lock a Mutex and failed, and have sent us the `LockMutex` + // message. This queue is drained by threads sending `UnlockMutex` to us. + let mut mutex_hash: HashMap, HashMap>> = + HashMap::new(); loop { #[cfg(feature = "watchdog")] @@ -540,9 +559,60 @@ fn xmain() -> ! { ticktimer.reset_wdt(); } Some(api::Opcode::GetVersion) => { - let mut buf = unsafe{xous_ipc::Buffer::from_memory_message_mut(msg.body.memory_message_mut().unwrap())}; + let mut buf = unsafe { + xous_ipc::Buffer::from_memory_message_mut( + msg.body.memory_message_mut().unwrap(), + ) + }; buf.replace(version::get_version()).unwrap(); } + Some(api::Opcode::LockMutex) => { + let pid = msg.sender.pid(); + if !msg.body.is_blocking() { + info!("sender made LockMutex request that was not blocking"); + break; + } + if let Some(scalar) = msg.body.scalar_message() { + let ready = mutex_ready_hash.entry(pid).or_default(); + + // If this item is in the Ready list, return right away without blocking + if ready.remove(&scalar.arg1) { + xous::return_scalar(msg.sender, 0).unwrap(); + break; + } + + // This item is not in the Ready list, so add our sender to the list of processes + // to get called when UnlockMutex is invoked + let awaiting = mutex_hash.entry(pid).or_default(); + + // Add this to the end of the list of entries to call so that when `UnlockMutex` is sent + // the message will get a response. + let mutex_entry = awaiting.entry(scalar.arg1).or_default(); + mutex_entry.push_back(msg.sender); + } + } + Some(api::Opcode::UnlockMutex) => { + let pid = msg.sender.pid(); + if msg.body.is_blocking() { + info!("sender made UnlockMutex request that was blocking"); + break; + } + if let Some(scalar) = msg.body.scalar_message() { + // Get a list of awaiting mutexes for this process + let awaiting = mutex_hash.entry(pid).or_default(); + + // Get the vector of awaiting mutex entries. + let mutex_entry = awaiting.entry(scalar.arg1).or_default(); + + // If there's something waiting in the queue, respond to that message + if let Some(sender) = mutex_entry.pop_front() { + xous::return_scalar(sender, 0).unwrap(); + } else { + // Otherwise, mark this scalar as being ready to run + mutex_ready_hash.entry(pid).or_default().insert(scalar.arg1); + } + } + } None => { error!("couldn't convert opcode"); break;