mirror/rtic
1
0
Fork 0
mirror of https://github.com/rtic-rs/rtic.git synced 2025-12-16 12:55:23 +01:00
Code Issues Projects Releases Packages Wiki Activity

Add blocking version of rtic_sync::arbiter::{i2c,spi}::ArbiterDevice

Browse source
This commit is contained in:
Emil Fresk 2024-06-23 10:33:02 +02:00
parent 95616b3c59
commit f42147948d
4 changed files with 342 additions and 191 deletions
Download patch file Download diff file Expand all files Collapse all files

7
rtic-sync/CHANGELOG.md
View file

@ -7,7 +7,14 @@ For each category, _Added_, _Changed_, _Fixed_ add new entries at the top!
## [Unreleased]
### Added
- Add `arbiter::{i2c, spi}::BlockingArbiterDevice` which allows sharing of `embedded_hal` (non-async) buses. This also helps during initialization of RTIC apps as you can use the arbiter while in `init`. After initialization is complete, convert an `BlockingArbiterDevice` into an `ArbiterDevice` using `BlockingArbiterDevice::into_non_blocking()`.
### Fixed
- Avoid a critical section when a `send`-link is popped and when returning `free_slot`.
### Changed
- Add `loom` support.

194
rtic-sync/src/arbiter.rs
View file

@ -29,10 +29,12 @@ use core::ops::{Deref, DerefMut};
use core::pin::Pin;
use core::task::{Poll, Waker};
use portable_atomic::{fence, AtomicBool, Ordering};
use rtic_common::dropper::OnDrop;
use rtic_common::wait_queue::{Link, WaitQueue};
pub mod i2c;
pub mod spi;
/// This is needed to make the async closure in `send` accept that we "share"
/// the link possible between threads.
#[derive(Clone)]
@ -191,196 +193,6 @@ impl<T> DerefMut for ExclusiveAccess<'_, T> {
}
}
/// SPI bus sharing using [`Arbiter`]
pub mod spi {
use super::Arbiter;
use embedded_hal::digital::OutputPin;
use embedded_hal_async::{
delay::DelayNs,
spi::{ErrorType, Operation, SpiBus, SpiDevice},
};
use embedded_hal_bus::spi::DeviceError;
/// [`Arbiter`]-based shared bus implementation.
pub struct ArbiterDevice<'a, BUS, CS, D> {
bus: &'a Arbiter<BUS>,
cs: CS,
delay: D,
}
impl<'a, BUS, CS, D> ArbiterDevice<'a, BUS, CS, D> {
/// Create a new [`ArbiterDevice`].
pub fn new(bus: &'a Arbiter<BUS>, cs: CS, delay: D) -> Self {
Self { bus, cs, delay }
}
}
impl<BUS, CS, D> ErrorType for ArbiterDevice<'_, BUS, CS, D>
where
BUS: ErrorType,
CS: OutputPin,
{
type Error = DeviceError<BUS::Error, CS::Error>;
}
impl<Word, BUS, CS, D> SpiDevice<Word> for ArbiterDevice<'_, BUS, CS, D>
where
Word: Copy + 'static,
BUS: SpiBus<Word>,
CS: OutputPin,
D: DelayNs,
{
async fn transaction(
&mut self,
operations: &mut [Operation<'_, Word>],
) -> Result<(), DeviceError<BUS::Error, CS::Error>> {
let mut bus = self.bus.access().await;
self.cs.set_low().map_err(DeviceError::Cs)?;
let op_res = 'ops: {
for op in operations {
let res = match op {
Operation::Read(buf) => bus.read(buf).await,
Operation::Write(buf) => bus.write(buf).await,
Operation::Transfer(read, write) => bus.transfer(read, write).await,
Operation::TransferInPlace(buf) => bus.transfer_in_place(buf).await,
Operation::DelayNs(ns) => match bus.flush().await {
Err(e) => Err(e),
Ok(()) => {
self.delay.delay_ns(*ns).await;
Ok(())
}
},
};
if let Err(e) = res {
break 'ops Err(e);
}
}
Ok(())
};
// On failure, it's important to still flush and deassert CS.
let flush_res = bus.flush().await;
let cs_res = self.cs.set_high();
op_res.map_err(DeviceError::Spi)?;
flush_res.map_err(DeviceError::Spi)?;
cs_res.map_err(DeviceError::Cs)?;
Ok(())
}
}
}
/// I2C bus sharing using [`Arbiter`]
///
/// An Example how to use it in RTIC application:
/// ```text
/// #[app(device = some_hal, peripherals = true, dispatchers = [TIM16])]
/// mod app {
/// use core::mem::MaybeUninit;
/// use rtic_sync::{arbiter::{i2c::ArbiterDevice, Arbiter},
///
/// #[shared]
/// struct Shared {}
///
/// #[local]
/// struct Local {
/// ens160: Ens160<ArbiterDevice<'static, I2c<'static, I2C1>>>,
/// }
///
/// #[init(local = [
/// i2c_arbiter: MaybeUninit<Arbiter<I2c<'static, I2C1>>> = MaybeUninit::uninit(),
/// ])]
/// fn init(cx: init::Context) -> (Shared, Local) {
/// let i2c = I2c::new(cx.device.I2C1);
/// let i2c_arbiter = cx.local.i2c_arbiter.write(Arbiter::new(i2c));
/// let ens160 = Ens160::new(ArbiterDevice::new(i2c_arbiter), 0x52);
///
/// i2c_sensors::spawn(i2c_arbiter).ok();
///
/// (Shared {}, Local { ens160 })
/// }
///
/// #[task(local = [ens160])]
/// async fn i2c_sensors(cx: i2c_sensors::Context, i2c: &'static Arbiter<I2c<'static, I2C1>>) {
/// use sensor::Asensor;
///
/// loop {
/// // Use scope to make sure I2C access is dropped.
/// {
/// // Read from sensor driver that wants to use I2C directly.
/// let mut i2c = i2c.access().await;
/// let status = Asensor::status(&mut i2c).await;
/// }
///
/// // Read ENS160 sensor.
/// let eco2 = cx.local.ens160.eco2().await;
/// }
/// }
/// }
/// ```
pub mod i2c {
use super::Arbiter;
use embedded_hal::i2c::{AddressMode, ErrorType, Operation};
use embedded_hal_async::i2c::I2c;
/// [`Arbiter`]-based shared bus implementation for I2C.
pub struct ArbiterDevice<'a, BUS> {
bus: &'a Arbiter<BUS>,
}
impl<'a, BUS> ArbiterDevice<'a, BUS> {
/// Create a new [`ArbiterDevice`] for I2C.
pub fn new(bus: &'a Arbiter<BUS>) -> Self {
Self { bus }
}
}
impl<BUS> ErrorType for ArbiterDevice<'_, BUS>
where
BUS: ErrorType,
{
type Error = BUS::Error;
}
impl<BUS, A> I2c<A> for ArbiterDevice<'_, BUS>
where
BUS: I2c<A>,
A: AddressMode,
{
async fn read(&mut self, address: A, read: &mut [u8]) -> Result<(), Self::Error> {
let mut bus = self.bus.access().await;
bus.read(address, read).await
}
async fn write(&mut self, address: A, write: &[u8]) -> Result<(), Self::Error> {
let mut bus = self.bus.access().await;
bus.write(address, write).await
}
async fn write_read(
&mut self,
address: A,
write: &[u8],
read: &mut [u8],
) -> Result<(), Self::Error> {
let mut bus = self.bus.access().await;
bus.write_read(address, write, read).await
}
async fn transaction(
&mut self,
address: A,
operations: &mut [Operation<'_>],
) -> Result<(), Self::Error> {
let mut bus = self.bus.access().await;
bus.transaction(address, operations).await
}
}
}
#[cfg(not(loom))]
#[cfg(test)]
mod tests {

168
rtic-sync/src/arbiter/i2c.rs Normal file
View file

@ -0,0 +1,168 @@
//! I2C bus sharing using [`Arbiter`]
//!
//! An Example how to use it in RTIC application:
//! ```text
//! #[app(device = some_hal, peripherals = true, dispatchers = [TIM16])]
//! mod app {
//! use core::mem::MaybeUninit;
//! use rtic_sync::{arbiter::{i2c::ArbiterDevice, Arbiter},
//!
//! #[shared]
//! struct Shared {}
//!
//! #[local]
//! struct Local {
//! ens160: Ens160<ArbiterDevice<'static, I2c<'static, I2C1>>>,
//! }
//!
//! #[init(local = [
//! i2c_arbiter: MaybeUninit<Arbiter<I2c<'static, I2C1>>> = MaybeUninit::uninit(),
//! ])]
//! fn init(cx: init::Context) -> (Shared, Local) {
//! let i2c = I2c::new(cx.device.I2C1);
//! let i2c_arbiter = cx.local.i2c_arbiter.write(Arbiter::new(i2c));
//! let ens160 = Ens160::new(ArbiterDevice::new(i2c_arbiter), 0x52);
//!
//! i2c_sensors::spawn(i2c_arbiter).ok();
//!
//! (Shared {}, Local { ens160 })
//! }
//!
//! #[task(local = [ens160])]
//! async fn i2c_sensors(cx: i2c_sensors::Context, i2c: &'static Arbiter<I2c<'static, I2C1>>) {
//! use sensor::Asensor;
//!
//! loop {
//! // Use scope to make sure I2C access is dropped.
//! {
//! // Read from sensor driver that wants to use I2C directly.
//! let mut i2c = i2c.access().await;
//! let status = Asensor::status(&mut i2c).await;
//! }
//!
//! // Read ENS160 sensor.
//! let eco2 = cx.local.ens160.eco2().await;
//! }
//! }
//! }
//! ```
use super::Arbiter;
use embedded_hal::i2c::{AddressMode, ErrorType, I2c as BlockingI2c, Operation};
use embedded_hal_async::i2c::I2c as AsyncI2c;
/// [`Arbiter`]-based shared bus implementation for I2C.
pub struct ArbiterDevice<'a, BUS> {
bus: &'a Arbiter<BUS>,
}
impl<'a, BUS> ArbiterDevice<'a, BUS> {
/// Create a new [`ArbiterDevice`] for I2C.
pub fn new(bus: &'a Arbiter<BUS>) -> Self {
Self { bus }
}
}
impl<BUS> ErrorType for ArbiterDevice<'_, BUS>
where
BUS: ErrorType,
{
type Error = BUS::Error;
}
impl<BUS, A> AsyncI2c<A> for ArbiterDevice<'_, BUS>
where
BUS: AsyncI2c<A>,
A: AddressMode,
{
async fn read(&mut self, address: A, read: &mut [u8]) -> Result<(), Self::Error> {
let mut bus = self.bus.access().await;
bus.read(address, read).await
}
async fn write(&mut self, address: A, write: &[u8]) -> Result<(), Self::Error> {
let mut bus = self.bus.access().await;
bus.write(address, write).await
}
async fn write_read(
&mut self,
address: A,
write: &[u8],
read: &mut [u8],
) -> Result<(), Self::Error> {
let mut bus = self.bus.access().await;
bus.write_read(address, write, read).await
}
async fn transaction(
&mut self,
address: A,
operations: &mut [Operation<'_>],
) -> Result<(), Self::Error> {
let mut bus = self.bus.access().await;
bus.transaction(address, operations).await
}
}
/// [`Arbiter`]-based shared bus implementation for I2C.
pub struct BlockingArbiterDevice<'a, BUS> {
bus: &'a Arbiter<BUS>,
}
impl<'a, BUS> BlockingArbiterDevice<'a, BUS> {
/// Create a new [`BlockingArbiterDevice`] for I2C.
pub fn new(bus: &'a Arbiter<BUS>) -> Self {
Self { bus }
}
/// Create an `ArbiterDevice` from an `BlockingArbiterDevice`.
pub fn into_non_blocking(self) -> ArbiterDevice<'a, BUS>
where
BUS: AsyncI2c,
{
ArbiterDevice { bus: self.bus }
}
}
impl<'a, BUS> ErrorType for BlockingArbiterDevice<'a, BUS>
where
BUS: ErrorType,
{
type Error = BUS::Error;
}
impl<'a, BUS, A> AsyncI2c<A> for BlockingArbiterDevice<'a, BUS>
where
BUS: BlockingI2c<A>,
A: AddressMode,
{
async fn read(&mut self, address: A, read: &mut [u8]) -> Result<(), Self::Error> {
let mut bus = self.bus.access().await;
bus.read(address, read)
}
async fn write(&mut self, address: A, write: &[u8]) -> Result<(), Self::Error> {
let mut bus = self.bus.access().await;
bus.write(address, write)
}
async fn write_read(
&mut self,
address: A,
write: &[u8],
read: &mut [u8],
) -> Result<(), Self::Error> {
let mut bus = self.bus.access().await;
bus.write_read(address, write, read)
}
async fn transaction(
&mut self,
address: A,
operations: &mut [Operation<'_>],
) -> Result<(), Self::Error> {
let mut bus = self.bus.access().await;
bus.transaction(address, operations)
}
}

164
rtic-sync/src/arbiter/spi.rs Normal file
View file

@ -0,0 +1,164 @@
//! SPI bus sharing using [`Arbiter`]
use super::Arbiter;
use embedded_hal::digital::OutputPin;
use embedded_hal::spi::SpiBus as BlockingSpiBus;
use embedded_hal_async::{
delay::DelayNs,
spi::{ErrorType, Operation, SpiBus as AsyncSpiBus, SpiDevice},
};
use embedded_hal_bus::spi::DeviceError;
/// [`Arbiter`]-based shared bus implementation.
pub struct ArbiterDevice<'a, BUS, CS, D> {
bus: &'a Arbiter<BUS>,
cs: CS,
delay: D,
}
impl<'a, BUS, CS, D> ArbiterDevice<'a, BUS, CS, D> {
/// Create a new [`ArbiterDevice`].
pub fn new(bus: &'a Arbiter<BUS>, cs: CS, delay: D) -> Self {
Self { bus, cs, delay }
}
}
impl<BUS, CS, D> ErrorType for ArbiterDevice<'_, BUS, CS, D>
where
BUS: ErrorType,
CS: OutputPin,
{
type Error = DeviceError<BUS::Error, CS::Error>;
}
impl<Word, BUS, CS, D> SpiDevice<Word> for ArbiterDevice<'_, BUS, CS, D>
where
Word: Copy + 'static,
BUS: AsyncSpiBus<Word>,
CS: OutputPin,
D: DelayNs,
{
async fn transaction(
&mut self,
operations: &mut [Operation<'_, Word>],
) -> Result<(), DeviceError<BUS::Error, CS::Error>> {
let mut bus = self.bus.access().await;
self.cs.set_low().map_err(DeviceError::Cs)?;
let op_res = 'ops: {
for op in operations {
let res = match op {
Operation::Read(buf) => bus.read(buf).await,
Operation::Write(buf) => bus.write(buf).await,
Operation::Transfer(read, write) => bus.transfer(read, write).await,
Operation::TransferInPlace(buf) => bus.transfer_in_place(buf).await,
Operation::DelayNs(ns) => match bus.flush().await {
Err(e) => Err(e),
Ok(()) => {
self.delay.delay_ns(*ns).await;
Ok(())
}
},
};
if let Err(e) = res {
break 'ops Err(e);
}
}
Ok(())
};
// On failure, it's important to still flush and deassert CS.
let flush_res = bus.flush().await;
let cs_res = self.cs.set_high();
op_res.map_err(DeviceError::Spi)?;
flush_res.map_err(DeviceError::Spi)?;
cs_res.map_err(DeviceError::Cs)?;
Ok(())
}
}
/// [`Arbiter`]-based shared bus implementation.
pub struct BlockingArbiterDevice<'a, BUS, CS, D> {
bus: &'a Arbiter<BUS>,
cs: CS,
delay: D,
}
impl<'a, BUS, CS, D> BlockingArbiterDevice<'a, BUS, CS, D> {
/// Create a new [`BlockingArbiterDevice`].
pub fn new(bus: &'a Arbiter<BUS>, cs: CS, delay: D) -> Self {
Self { bus, cs, delay }
}
/// Create an `ArbiterDevice` from an `BlockingArbiterDevice`.
pub fn into_non_blocking(self) -> ArbiterDevice<'a, BUS, CS, D>
where
BUS: AsyncSpiBus,
{
ArbiterDevice {
bus: self.bus,
cs: self.cs,
delay: self.delay,
}
}
}
impl<'a, BUS, CS, D> ErrorType for BlockingArbiterDevice<'a, BUS, CS, D>
where
BUS: ErrorType,
CS: OutputPin,
{
type Error = DeviceError<BUS::Error, CS::Error>;
}
impl<'a, Word, BUS, CS, D> SpiDevice<Word> for BlockingArbiterDevice<'a, BUS, CS, D>
where
Word: Copy + 'static,
BUS: BlockingSpiBus<Word>,
CS: OutputPin,
D: DelayNs,
{
async fn transaction(
&mut self,
operations: &mut [Operation<'_, Word>],
) -> Result<(), DeviceError<BUS::Error, CS::Error>> {
let mut bus = self.bus.access().await;
self.cs.set_low().map_err(DeviceError::Cs)?;
let op_res = 'ops: {
for op in operations {
let res = match op {
Operation::Read(buf) => bus.read(buf),
Operation::Write(buf) => bus.write(buf),
Operation::Transfer(read, write) => bus.transfer(read, write),
Operation::TransferInPlace(buf) => bus.transfer_in_place(buf),
Operation::DelayNs(ns) => match bus.flush() {
Err(e) => Err(e),
Ok(()) => {
self.delay.delay_ns(*ns).await;
Ok(())
}
},
};
if let Err(e) = res {
break 'ops Err(e);
}
}
Ok(())
};
// On failure, it's important to still flush and deassert CS.
let flush_res = bus.flush();
let cs_res = self.cs.set_high();
op_res.map_err(DeviceError::Spi)?;
flush_res.map_err(DeviceError::Spi)?;
cs_res.map_err(DeviceError::Cs)?;
Ok(())
}
}