max/embassy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Use of a NoopMutex

Browse Source
This commit is contained in:
huntc 2021-07-09 12:13:07 +10:00
parent 5f87c7808c
commit f159beec1c
3 changed files with 78 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

109
embassy/src/util/mpsc.rs
View File

@ -49,6 +49,7 @@ use futures::Future;
use super::CriticalSectionMutex; use super::CriticalSectionMutex;
use super::Mutex; use super::Mutex;
use super::NoopMutex;
use super::ThreadModeMutex; use super::ThreadModeMutex;
/// Send values to the associated `Receiver`. /// Send values to the associated `Receiver`.
@ -96,10 +97,10 @@ unsafe impl<'ch, M, T, const N: usize> Sync for Receiver<'ch, M, T, N> where M:
//// ////
/// ```compile_fail /// ```compile_fail
/// use embassy::util::mpsc; /// use embassy::util::mpsc;
/// use embassy::util::mpsc::{Channel, ChannelCell, WithThreadModeOnly}; /// use embassy::util::mpsc::{Channel, WithThreadModeOnly};
/// ///
/// let (sender, receiver) = { /// let (sender, receiver) = {
/// let mut channel = ChannelCell::new(Channel::<WithThreadModeOnly, u32, 3>::with_thread_mode_only()); /// let mut channel = Channel::<WithThreadModeOnly, u32, 3>::with_thread_mode_only();
/// mpsc::split(&channel) /// mpsc::split(&channel)
/// }; /// };
/// ``` /// ```
@ -464,10 +465,10 @@ impl<T, const N: usize> Channel<WithCriticalSections, T, N> {
/// ///
/// ``` /// ```
/// use embassy::util::mpsc; /// use embassy::util::mpsc;
/// use embassy::util::mpsc::{Channel, ChannelCell, WithCriticalSections}; /// use embassy::util::mpsc::{Channel, WithCriticalSections};
/// ///
/// // Declare a bounded channel of 3 u32s. /// // Declare a bounded channel of 3 u32s.
/// let mut channel = ChannelCell::new(mpsc::Channel::<WithCriticalSections, u32, 3>::with_critical_sections()); /// let mut channel = Channel::<WithCriticalSections, u32, 3>::with_critical_sections();
/// // once we have a channel, obtain its sender and receiver /// // once we have a channel, obtain its sender and receiver
/// let (sender, receiver) = mpsc::split(&channel); /// let (sender, receiver) = mpsc::split(&channel);
/// ``` /// ```
@ -488,10 +489,10 @@ impl<T, const N: usize> Channel<WithThreadModeOnly, T, N> {
/// ///
/// ``` no_run /// ``` no_run
/// use embassy::util::mpsc; /// use embassy::util::mpsc;
/// use embassy::util::mpsc::{Channel, ChannelCell, WithThreadModeOnly}; /// use embassy::util::mpsc::{Channel, WithThreadModeOnly};
/// ///
/// // Declare a bounded channel of 3 u32s. /// // Declare a bounded channel of 3 u32s.
/// let mut channel = ChannelCell::new(Channel::<WithThreadModeOnly, u32, 3>::with_thread_mode_only()); /// let mut channel = Channel::<WithThreadModeOnly, u32, 3>::with_thread_mode_only();
/// // once we have a channel, obtain its sender and receiver /// // once we have a channel, obtain its sender and receiver
/// let (sender, receiver) = mpsc::split(&channel); /// let (sender, receiver) = mpsc::split(&channel);
/// ``` /// ```
@ -502,6 +503,27 @@ impl<T, const N: usize> Channel<WithThreadModeOnly, T, N> {
} }
} }
pub type WithNoThreads = NoopMutex<()>;
impl<T, const N: usize> Channel<WithNoThreads, T, N> {
/// Establish a new bounded channel for within a single thread. To create one:
///
/// ```
/// use embassy::util::mpsc;
/// use embassy::util::mpsc::{Channel, WithNoThreads};
///
/// // Declare a bounded channel of 3 u32s.
/// let mut channel = Channel::<WithNoThreads, u32, 3>::with_no_threads();
/// // once we have a channel, obtain its sender and receiver
/// let (sender, receiver) = mpsc::split(&channel);
/// ```
pub const fn with_no_threads() -> Self {
let mutex = NoopMutex::new(());
let state = ChannelState::new();
Channel { mutex, state }
}
}
impl<M, T, const N: usize> Channel<M, T, N> impl<M, T, const N: usize> Channel<M, T, N>
where where
M: Mutex<Data = ()>, M: Mutex<Data = ()>,
@ -675,43 +697,6 @@ mod tests {
} }
} }
/// A mutex that does nothing - useful for our testing purposes
pub struct NoopMutex<T> {
inner: UnsafeCell<T>,
}
impl<T> NoopMutex<T> {
pub const fn new(value: T) -> Self {
NoopMutex {
inner: UnsafeCell::new(value),
}
}
}
impl<T> NoopMutex<T> {
pub fn borrow(&self) -> &T {
unsafe { &*self.inner.get() }
}
}
impl<T> Mutex for NoopMutex<T> {
type Data = T;
fn lock<R>(&mut self, f: impl FnOnce(&Self::Data) -> R) -> R {
f(self.borrow())
}
}
pub type WithNoThreads = NoopMutex<()>;
impl<T, const N: usize> Channel<WithNoThreads, T, N> {
pub const fn with_no_threads() -> Self {
let mutex = NoopMutex::new(());
let state = ChannelState::new();
Channel { mutex, state }
}
}
#[test] #[test]
fn sending_once() { fn sending_once() {
let mut c = Channel::<WithNoThreads, u32, 3>::with_no_threads(); let mut c = Channel::<WithNoThreads, u32, 3>::with_no_threads();
@ -772,7 +757,7 @@ mod tests {
#[test] #[test]
fn simple_send_and_receive() { fn simple_send_and_receive() {
let c = ChannelCell::new(Channel::<WithNoThreads, u32, 3>::with_no_threads()); let c = Channel::<WithNoThreads, u32, 3>::with_no_threads();
let (s, r) = split(&c); let (s, r) = split(&c);
assert!(s.clone().try_send(1).is_ok()); assert!(s.clone().try_send(1).is_ok());
assert_eq!(r.try_recv().unwrap(), 1); assert_eq!(r.try_recv().unwrap(), 1);
@ -780,7 +765,7 @@ mod tests {
#[test] #[test]
fn should_close_without_sender() { fn should_close_without_sender() {
let c = ChannelCell::new(Channel::<WithNoThreads, u32, 3>::with_no_threads()); let c = Channel::<WithNoThreads, u32, 3>::with_no_threads();
let (s, r) = split(&c); let (s, r) = split(&c);
drop(s); drop(s);
match r.try_recv() { match r.try_recv() {
@ -791,7 +776,7 @@ mod tests {
#[test] #[test]
fn should_close_once_drained() { fn should_close_once_drained() {
let c = ChannelCell::new(Channel::<WithNoThreads, u32, 3>::with_no_threads()); let c = Channel::<WithNoThreads, u32, 3>::with_no_threads();
let (s, r) = split(&c); let (s, r) = split(&c);
assert!(s.try_send(1).is_ok()); assert!(s.try_send(1).is_ok());
drop(s); drop(s);
@ -804,7 +789,7 @@ mod tests {
#[test] #[test]
fn should_reject_send_when_receiver_dropped() { fn should_reject_send_when_receiver_dropped() {
let c = ChannelCell::new(Channel::<WithNoThreads, u32, 3>::with_no_threads()); let c = Channel::<WithNoThreads, u32, 3>::with_no_threads();
let (s, r) = split(&c); let (s, r) = split(&c);
drop(r); drop(r);
match s.try_send(1) { match s.try_send(1) {
@ -815,7 +800,7 @@ mod tests {
#[test] #[test]
fn should_reject_send_when_channel_closed() { fn should_reject_send_when_channel_closed() {
let c = ChannelCell::new(Channel::<WithNoThreads, u32, 3>::with_no_threads()); let c = Channel::<WithNoThreads, u32, 3>::with_no_threads();
let (s, mut r) = split(&c); let (s, mut r) = split(&c);
assert!(s.try_send(1).is_ok()); assert!(s.try_send(1).is_ok());
r.close(); r.close();
@ -831,8 +816,8 @@ mod tests {
async fn receiver_closes_when_sender_dropped_async() { async fn receiver_closes_when_sender_dropped_async() {
let executor = ThreadPool::new().unwrap(); let executor = ThreadPool::new().unwrap();
static mut CHANNEL: ChannelCell<Channel<WithCriticalSections, u32, 3>> = static mut CHANNEL: Channel<WithCriticalSections, u32, 3> =
ChannelCell::new(Channel::with_critical_sections()); Channel::with_critical_sections();
let (s, mut r) = split(unsafe { &CHANNEL }); let (s, mut r) = split(unsafe { &CHANNEL });
assert!(executor assert!(executor
.spawn(async move { .spawn(async move {
@ -846,8 +831,8 @@ mod tests {
async fn receiver_receives_given_try_send_async() { async fn receiver_receives_given_try_send_async() {
let executor = ThreadPool::new().unwrap(); let executor = ThreadPool::new().unwrap();
static mut CHANNEL: ChannelCell<Channel<WithCriticalSections, u32, 3>> = static mut CHANNEL: Channel<WithCriticalSections, u32, 3> =
ChannelCell::new(Channel::with_critical_sections()); Channel::with_critical_sections();
let (s, mut r) = split(unsafe { &CHANNEL }); let (s, mut r) = split(unsafe { &CHANNEL });
assert!(executor assert!(executor
.spawn(async move { .spawn(async move {
@ -859,8 +844,8 @@ mod tests {
#[futures_test::test] #[futures_test::test]
async fn sender_send_completes_if_capacity() { async fn sender_send_completes_if_capacity() {
static mut CHANNEL: ChannelCell<Channel<WithCriticalSections, u32, 1>> = static mut CHANNEL: Channel<WithCriticalSections, u32, 1> =
ChannelCell::new(Channel::with_critical_sections()); Channel::with_critical_sections();
let (s, mut r) = split(unsafe { &CHANNEL }); let (s, mut r) = split(unsafe { &CHANNEL });
assert!(s.send(1).await.is_ok()); assert!(s.send(1).await.is_ok());
assert_eq!(r.recv().await, Some(1)); assert_eq!(r.recv().await, Some(1));
@ -868,8 +853,8 @@ mod tests {
#[futures_test::test] #[futures_test::test]
async fn sender_send_completes_if_closed() { async fn sender_send_completes_if_closed() {
static mut CHANNEL: ChannelCell<Channel<WithCriticalSections, u32, 1>> = static mut CHANNEL: Channel<WithCriticalSections, u32, 1> =
ChannelCell::new(Channel::with_critical_sections()); Channel::with_critical_sections();
let (s, r) = split(unsafe { &CHANNEL }); let (s, r) = split(unsafe { &CHANNEL });
drop(r); drop(r);
match s.send(1).await { match s.send(1).await {
@ -882,8 +867,8 @@ mod tests {
async fn senders_sends_wait_until_capacity() { async fn senders_sends_wait_until_capacity() {
let executor = ThreadPool::new().unwrap(); let executor = ThreadPool::new().unwrap();
static mut CHANNEL: ChannelCell<Channel<WithCriticalSections, u32, 1>> = static mut CHANNEL: Channel<WithCriticalSections, u32, 1> =
ChannelCell::new(Channel::with_critical_sections()); Channel::with_critical_sections();
let (s0, mut r) = split(unsafe { &CHANNEL }); let (s0, mut r) = split(unsafe { &CHANNEL });
assert!(s0.try_send(1).is_ok()); assert!(s0.try_send(1).is_ok());
let s1 = s0.clone(); let s1 = s0.clone();
@ -902,8 +887,8 @@ mod tests {
#[futures_test::test] #[futures_test::test]
async fn sender_close_completes_if_closing() { async fn sender_close_completes_if_closing() {
static mut CHANNEL: ChannelCell<Channel<WithCriticalSections, u32, 1>> = static mut CHANNEL: Channel<WithCriticalSections, u32, 1> =
ChannelCell::new(Channel::with_critical_sections()); Channel::with_critical_sections();
let (s, mut r) = split(unsafe { &CHANNEL }); let (s, mut r) = split(unsafe { &CHANNEL });
r.close(); r.close();
s.closed().await; s.closed().await;
@ -911,8 +896,8 @@ mod tests {
#[futures_test::test] #[futures_test::test]
async fn sender_close_completes_if_closed() { async fn sender_close_completes_if_closed() {
static mut CHANNEL: ChannelCell<Channel<WithCriticalSections, u32, 1>> = static mut CHANNEL: Channel<WithCriticalSections, u32, 1> =
ChannelCell::new(Channel::with_critical_sections()); Channel::with_critical_sections();
let (s, r) = split(unsafe { &CHANNEL }); let (s, r) = split(unsafe { &CHANNEL });
drop(r); drop(r);
s.closed().await; s.closed().await;

27
embassy/src/util/mutex.rs
View File

@ -105,3 +105,30 @@ pub fn in_thread_mode() -> bool {
return cortex_m::peripheral::SCB::vect_active() return cortex_m::peripheral::SCB::vect_active()
== cortex_m::peripheral::scb::VectActive::ThreadMode; == cortex_m::peripheral::scb::VectActive::ThreadMode;
} }
/// A "mutex" that does nothing and cannot be shared between threads.
pub struct NoopMutex<T> {
inner: UnsafeCell<T>,
}
impl<T> NoopMutex<T> {
pub const fn new(value: T) -> Self {
NoopMutex {
inner: UnsafeCell::new(value),
}
}
}
impl<T> NoopMutex<T> {
pub fn borrow(&self) -> &T {
unsafe { &*self.inner.get() }
}
}
impl<T> Mutex for NoopMutex<T> {
type Data = T;
fn lock<R>(&mut self, f: impl FnOnce(&Self::Data) -> R) -> R {
f(self.borrow())
}
}

8
examples/nrf/src/bin/mpsc.rs
View File

@ -16,17 +16,17 @@ use embassy::util::{mpsc, Forever};
use embassy_nrf::gpio::{Level, Output, OutputDrive}; use embassy_nrf::gpio::{Level, Output, OutputDrive};
use embassy_nrf::Peripherals; use embassy_nrf::Peripherals;
use embedded_hal::digital::v2::OutputPin; use embedded_hal::digital::v2::OutputPin;
use mpsc::{Channel, Sender, WithThreadModeOnly}; use mpsc::{Channel, Sender, WithNoThreads};
enum LedState { enum LedState {
On, On,
Off, Off,
} }
static CHANNEL: Forever<Channel<WithThreadModeOnly, LedState, 1>> = Forever::new(); static CHANNEL: Forever<Channel<WithNoThreads, LedState, 1>> = Forever::new();
#[embassy::task(pool_size = 1)] #[embassy::task(pool_size = 1)]
async fn my_task(sender: Sender<'static, WithThreadModeOnly, LedState, 1>) { async fn my_task(sender: Sender<'static, WithNoThreads, LedState, 1>) {
loop { loop {
let _ = sender.send(LedState::On).await; let _ = sender.send(LedState::On).await;
Timer::after(Duration::from_secs(1)).await; Timer::after(Duration::from_secs(1)).await;
@ -39,7 +39,7 @@ async fn my_task(sender: Sender<'static, WithThreadModeOnly, LedState, 1>) {
async fn main(spawner: Spawner, p: Peripherals) { async fn main(spawner: Spawner, p: Peripherals) {
let mut led = Output::new(p.P0_13, Level::Low, OutputDrive::Standard); let mut led = Output::new(p.P0_13, Level::Low, OutputDrive::Standard);
let channel = CHANNEL.put(Channel::with_thread_mode_only()); let channel = CHANNEL.put(Channel::with_no_threads());
let (sender, mut receiver) = mpsc::split(channel); let (sender, mut receiver) = mpsc::split(channel);
spawner.spawn(my_task(sender)).unwrap(); spawner.spawn(my_task(sender)).unwrap();