itsscb/embassy
1
0
Fork 0
mirror of https://github.com/embassy-rs/embassy.git synced 2025-09-27 04:10:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor integrated-timers

Browse Source
This commit is contained in:
Dániel Buga 2024-11-26 23:54:21 +01:00
parent 406d377b75
commit 5a5495aac4
No known key found for this signature in database
32 changed files with 611 additions and 1182 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.github/ci/test.sh vendored
View File

@ -17,7 +17,7 @@ cargo test --manifest-path ./embassy-futures/Cargo.toml
cargo test --manifest-path ./embassy-sync/Cargo.toml
cargo test --manifest-path ./embassy-embedded-hal/Cargo.toml
cargo test --manifest-path ./embassy-hal-internal/Cargo.toml
cargo test --manifest-path ./embassy-time/Cargo.toml --features generic-queue,mock-driver
cargo test --manifest-path ./embassy-time/Cargo.toml --features mock-driver
cargo test --manifest-path ./embassy-time-driver/Cargo.toml
cargo test --manifest-path ./embassy-boot/Cargo.toml

4
ci-xtensa.sh
View File

@ -24,7 +24,9 @@ cargo batch \
--- build --release --manifest-path embassy-executor/Cargo.toml --target xtensa-esp32-none-elf --features arch-spin,executor-thread \
--- build --release --manifest-path embassy-executor/Cargo.toml --target xtensa-esp32-none-elf --features arch-spin,executor-thread,integrated-timers \
--- build --release --manifest-path embassy-sync/Cargo.toml --target xtensa-esp32s2-none-elf --features defmt \
--- build --release --manifest-path embassy-time/Cargo.toml --target xtensa-esp32s2-none-elf --features defmt,defmt-timestamp-uptime,generic-queue-8,mock-driver \
--- build --release --manifest-path embassy-time/Cargo.toml --target xtensa-esp32s2-none-elf --features defmt,defmt-timestamp-uptime,mock-driver \
--- build --release --manifest-path embassy-time-queue-driver/Cargo.toml --target xtensa-esp32s2-none-elf --features integrated-timers \
--- build --release --manifest-path embassy-time-queue-driver/Cargo.toml --target xtensa-esp32s2-none-elf --features generic-queue-8 \
--- build --release --manifest-path embassy-net/Cargo.toml --target xtensa-esp32-none-elf --features defmt,tcp,udp,dns,proto-ipv4,medium-ethernet,packet-trace \
--- build --release --manifest-path embassy-net/Cargo.toml --target xtensa-esp32-none-elf --features defmt,tcp,udp,dns,proto-ipv4,multicast,medium-ethernet \
--- build --release --manifest-path embassy-net/Cargo.toml --target xtensa-esp32-none-elf --features defmt,tcp,udp,dns,dhcpv4,medium-ethernet \

4
ci.sh
View File

@ -45,7 +45,9 @@ cargo batch \
--- build --release --manifest-path embassy-executor/Cargo.toml --target riscv32imac-unknown-none-elf --features arch-riscv32,executor-thread \
--- build --release --manifest-path embassy-executor/Cargo.toml --target riscv32imac-unknown-none-elf --features arch-riscv32,executor-thread,integrated-timers \
--- build --release --manifest-path embassy-sync/Cargo.toml --target thumbv6m-none-eabi --features defmt \
--- build --release --manifest-path embassy-time/Cargo.toml --target thumbv6m-none-eabi --features defmt,defmt-timestamp-uptime,generic-queue-8,mock-driver \
--- build --release --manifest-path embassy-time/Cargo.toml --target thumbv6m-none-eabi --features defmt,defmt-timestamp-uptime,mock-driver \
--- build --release --manifest-path embassy-time-queue-driver/Cargo.toml --target thumbv6m-none-eabi --features integrated-timers \
--- build --release --manifest-path embassy-time-queue-driver/Cargo.toml --target thumbv6m-none-eabi --features generic-queue-8 \
--- build --release --manifest-path embassy-net/Cargo.toml --target thumbv7em-none-eabi --features defmt,tcp,udp,dns,proto-ipv4,medium-ethernet,packet-trace \
--- build --release --manifest-path embassy-net/Cargo.toml --target thumbv7em-none-eabi --features defmt,tcp,udp,dns,proto-ipv4,multicast,medium-ethernet \
--- build --release --manifest-path embassy-net/Cargo.toml --target thumbv7em-none-eabi --features defmt,tcp,udp,dns,dhcpv4,medium-ethernet \

2
embassy-executor/Cargo.toml
View File

@ -68,7 +68,7 @@ nightly = ["embassy-executor-macros/nightly"]
turbowakers = []
## Use the executor-integrated `embassy-time` timer queue.
integrated-timers = ["dep:embassy-time-driver", "dep:embassy-time-queue-driver"]
integrated-timers = ["dep:embassy-time-driver"]
#! ### Architecture
_arch = [] # some arch was picked

4
embassy-executor/src/arch/avr.rs
View File

@ -53,10 +53,6 @@ mod thread {
///
/// This function never returns.
pub fn run(&'static mut self, init: impl FnOnce(Spawner)) -> ! {
unsafe {
self.inner.initialize();
}
init(self.inner.spawner());
loop {

6
embassy-executor/src/arch/cortex_m.rs
View File

@ -98,9 +98,6 @@ mod thread {
///
/// This function never returns.
pub fn run(&'static mut self, init: impl FnOnce(Spawner)) -> ! {
unsafe {
self.inner.initialize();
}
init(self.inner.spawner());
loop {
@ -210,9 +207,6 @@ mod interrupt {
}
let executor = unsafe { (&*self.executor.get()).assume_init_ref() };
unsafe {
executor.initialize();
}
unsafe { NVIC::unmask(irq) }

4
embassy-executor/src/arch/riscv32.rs
View File

@ -54,10 +54,6 @@ mod thread {
///
/// This function never returns.
pub fn run(&'static mut self, init: impl FnOnce(Spawner)) -> ! {
unsafe {
self.inner.initialize();
}
init(self.inner.spawner());
loop {

4
embassy-executor/src/arch/spin.rs
View File

@ -48,10 +48,6 @@ mod thread {
///
/// This function never returns.
pub fn run(&'static mut self, init: impl FnOnce(Spawner)) -> ! {
unsafe {
self.inner.initialize();
}
init(self.inner.spawner());
loop {

4
embassy-executor/src/arch/std.rs
View File

@ -55,10 +55,6 @@ mod thread {
///
/// This function never returns.
pub fn run(&'static mut self, init: impl FnOnce(Spawner)) -> ! {
unsafe {
self.inner.initialize();
}
init(self.inner.spawner());
loop {

4
embassy-executor/src/arch/wasm.rs
View File

@ -70,10 +70,6 @@ mod thread {
/// - a `static mut` (unsafe)
/// - a local variable in a function you know never returns (like `fn main() -> !`), upgrading its lifetime with `transmute`. (unsafe)
pub fn start(&'static mut self, init: impl FnOnce(Spawner)) {
unsafe {
self.inner.initialize();
}
unsafe {
let executor = &self.inner;
let future = Closure::new(move |_| {

135
embassy-executor/src/raw/mod.rs
View File

@ -17,7 +17,7 @@ mod run_queue;
mod state;
#[cfg(feature = "integrated-timers")]
mod timer_queue;
pub mod timer_queue;
#[cfg(feature = "trace")]
mod trace;
pub(crate) mod util;
@ -31,9 +31,6 @@ use core::pin::Pin;
use core::ptr::NonNull;
use core::task::{Context, Poll};
#[cfg(feature = "integrated-timers")]
use embassy_time_driver::AlarmHandle;
use self::run_queue::{RunQueue, RunQueueItem};
use self::state::State;
use self::util::{SyncUnsafeCell, UninitCell};
@ -47,8 +44,7 @@ pub(crate) struct TaskHeader {
pub(crate) executor: SyncUnsafeCell<Option<&'static SyncExecutor>>,
poll_fn: SyncUnsafeCell<Option<unsafe fn(TaskRef)>>,
#[cfg(feature = "integrated-timers")]
pub(crate) expires_at: SyncUnsafeCell<u64>,
/// Integrated timer queue storage. This field should not be accessed outside of the timer queue.
#[cfg(feature = "integrated-timers")]
pub(crate) timer_queue_item: timer_queue::TimerQueueItem,
}
@ -80,6 +76,12 @@ impl TaskRef {
unsafe { self.ptr.as_ref() }
}
/// Returns a reference to the executor that the task is currently running on.
#[cfg(feature = "integrated-timers")]
pub unsafe fn executor(self) -> Option<&'static Executor> {
self.header().executor.get().map(|e| Executor::wrap(e))
}
/// The returned pointer is valid for the entire TaskStorage.
pub(crate) fn as_ptr(self) -> *const TaskHeader {
self.ptr.as_ptr()
@ -120,8 +122,6 @@ impl<F: Future + 'static> TaskStorage<F> {
// Note: this is lazily initialized so that a static `TaskStorage` will go in `.bss`
poll_fn: SyncUnsafeCell::new(None),
#[cfg(feature = "integrated-timers")]
expires_at: SyncUnsafeCell::new(0),
#[cfg(feature = "integrated-timers")]
timer_queue_item: timer_queue::TimerQueueItem::new(),
},
@ -160,9 +160,6 @@ impl<F: Future + 'static> TaskStorage<F> {
Poll::Ready(_) => {
this.future.drop_in_place();
this.raw.state.despawn();
#[cfg(feature = "integrated-timers")]
this.raw.expires_at.set(u64::MAX);
}
Poll::Pending => {}
}
@ -316,34 +313,16 @@ impl Pender {
pub(crate) struct SyncExecutor {
run_queue: RunQueue,
pender: Pender,
#[cfg(feature = "integrated-timers")]
pub(crate) timer_queue: timer_queue::TimerQueue,
#[cfg(feature = "integrated-timers")]
alarm: AlarmHandle,
}
impl SyncExecutor {
pub(crate) fn new(pender: Pender) -> Self {
#[cfg(feature = "integrated-timers")]
let alarm = unsafe { unwrap!(embassy_time_driver::allocate_alarm()) };
Self {
run_queue: RunQueue::new(),
pender,
#[cfg(feature = "integrated-timers")]
timer_queue: timer_queue::TimerQueue::new(),
#[cfg(feature = "integrated-timers")]
alarm,
}
}
pub(crate) unsafe fn initialize(&'static self) {
#[cfg(feature = "integrated-timers")]
embassy_time_driver::set_alarm_callback(self.alarm, Self::alarm_callback, self as *const _ as *mut ());
}
/// Enqueue a task in the task queue
///
/// # Safety
@ -360,12 +339,6 @@ impl SyncExecutor {
}
}
#[cfg(feature = "integrated-timers")]
fn alarm_callback(ctx: *mut ()) {
let this: &Self = unsafe { &*(ctx as *const Self) };
this.pender.pend();
}
pub(super) unsafe fn spawn(&'static self, task: TaskRef) {
task.header().executor.set(Some(self));
@ -379,56 +352,27 @@ impl SyncExecutor {
///
/// Same as [`Executor::poll`], plus you must only call this on the thread this executor was created.
pub(crate) unsafe fn poll(&'static self) {
#[allow(clippy::never_loop)]
loop {
#[cfg(feature = "integrated-timers")]
self.timer_queue
.dequeue_expired(embassy_time_driver::now(), wake_task_no_pend);
self.run_queue.dequeue_all(|p| {
let task = p.header();
self.run_queue.dequeue_all(|p| {
let task = p.header();
#[cfg(feature = "integrated-timers")]
task.expires_at.set(u64::MAX);
if !task.state.run_dequeue() {
// If task is not running, ignore it. This can happen in the following scenario:
// - Task gets dequeued, poll starts
// - While task is being polled, it gets woken. It gets placed in the queue.
// - Task poll finishes, returning done=true
// - RUNNING bit is cleared, but the task is already in the queue.
return;
}
#[cfg(feature = "trace")]
trace::task_exec_begin(self, &p);
// Run the task
task.poll_fn.get().unwrap_unchecked()(p);
#[cfg(feature = "trace")]
trace::task_exec_end(self, &p);
// Enqueue or update into timer_queue
#[cfg(feature = "integrated-timers")]
self.timer_queue.update(p);
});
#[cfg(feature = "integrated-timers")]
{
// If this is already in the past, set_alarm might return false
// In that case do another poll loop iteration.
let next_expiration = self.timer_queue.next_expiration();
if embassy_time_driver::set_alarm(self.alarm, next_expiration) {
break;
}
if !task.state.run_dequeue() {
// If task is not running, ignore it. This can happen in the following scenario:
// - Task gets dequeued, poll starts
// - While task is being polled, it gets woken. It gets placed in the queue.
// - Task poll finishes, returning done=true
// - RUNNING bit is cleared, but the task is already in the queue.
return;
}
#[cfg(not(feature = "integrated-timers"))]
{
break;
}
}
#[cfg(feature = "trace")]
trace::task_exec_begin(self, &p);
// Run the task
task.poll_fn.get().unwrap_unchecked()(p);
#[cfg(feature = "trace")]
trace::task_exec_end(self, &p);
});
#[cfg(feature = "trace")]
trace::executor_idle(self)
@ -494,15 +438,6 @@ impl Executor {
}
}
/// Initializes the executor.
///
/// # Safety
///
/// This function must be called once before any other method is called.
pub unsafe fn initialize(&'static self) {
self.inner.initialize();
}
/// Spawn a task in this executor.
///
/// # Safety
@ -575,21 +510,3 @@ pub fn wake_task_no_pend(task: TaskRef) {
}
}
}
#[cfg(feature = "integrated-timers")]
struct TimerQueue;
#[cfg(feature = "integrated-timers")]
impl embassy_time_queue_driver::TimerQueue for TimerQueue {
fn schedule_wake(&'static self, at: u64, waker: &core::task::Waker) {
let task = waker::task_from_waker(waker);
let task = task.header();
unsafe {
let expires_at = task.expires_at.get();
task.expires_at.set(expires_at.min(at));
}
}
}
#[cfg(feature = "integrated-timers")]
embassy_time_queue_driver::timer_queue_impl!(static TIMER_QUEUE: TimerQueue = TimerQueue);

89
embassy-executor/src/raw/timer_queue.rs
View File

@ -1,75 +1,100 @@
//! Timer queue operations.
use core::cmp::min;
use super::util::SyncUnsafeCell;
use super::TaskRef;
use crate::raw::util::SyncUnsafeCell;
pub(crate) struct TimerQueueItem {
next: SyncUnsafeCell<Option<TaskRef>>,
expires_at: SyncUnsafeCell<u64>,
}
impl TimerQueueItem {
pub const fn new() -> Self {
Self {
next: SyncUnsafeCell::new(None),
expires_at: SyncUnsafeCell::new(0),
}
}
}
pub(crate) struct TimerQueue {
/// A timer queue, with items integrated into tasks.
pub struct TimerQueue {
head: SyncUnsafeCell<Option<TaskRef>>,
}
impl TimerQueue {
/// Creates a new timer queue.
pub const fn new() -> Self {
Self {
head: SyncUnsafeCell::new(None),
}
}
pub(crate) unsafe fn update(&self, p: TaskRef) {
let task = p.header();
if task.expires_at.get() != u64::MAX {
/// Schedules a task to run at a specific time.
///
/// If this function returns `true`, the called should find the next expiration time and set
/// a new alarm for that time.
pub fn schedule_wake(&mut self, at: u64, p: TaskRef) -> bool {
unsafe {
let task = p.header();
let item = &task.timer_queue_item;
if task.state.timer_enqueue() {
task.timer_queue_item.next.set(self.head.get());
self.head.set(Some(p));
// If not in the queue, add it and update.
let prev = self.head.replace(Some(p));
item.next.set(prev);
} else if at <= item.expires_at.get() {
// If expiration is sooner than previously set, update.
} else {
// Task does not need to be updated.
return false;
}
item.expires_at.set(at);
true
}
}
pub(crate) unsafe fn next_expiration(&self) -> u64 {
let mut res = u64::MAX;
self.retain(|p| {
let task = p.header();
let expires = task.expires_at.get();
res = min(res, expires);
expires != u64::MAX
});
res
}
/// Dequeues expired timers and returns the next alarm time.
///
/// The provided callback will be called for each expired task. Tasks that never expire
/// will be removed, but the callback will not be called.
pub fn next_expiration(&mut self, now: u64) -> u64 {
let mut next_expiration = u64::MAX;
pub(crate) unsafe fn dequeue_expired(&self, now: u64, on_task: impl Fn(TaskRef)) {
self.retain(|p| {
let task = p.header();
if task.expires_at.get() <= now {
on_task(p);
let item = &task.timer_queue_item;
let expires = unsafe { item.expires_at.get() };
if expires <= now {
// Timer expired, process task.
super::wake_task(p);
false
} else {
true
// Timer didn't yet expire, or never expires.
next_expiration = min(next_expiration, expires);
expires != u64::MAX
}
});
next_expiration
}
pub(crate) unsafe fn retain(&self, mut f: impl FnMut(TaskRef) -> bool) {
let mut prev = &self.head;
while let Some(p) = prev.get() {
let task = p.header();
if f(p) {
// Skip to next
prev = &task.timer_queue_item.next;
} else {
// Remove it
prev.set(task.timer_queue_item.next.get());
task.state.timer_dequeue();
fn retain(&self, mut f: impl FnMut(TaskRef) -> bool) {
unsafe {
let mut prev = &self.head;
while let Some(p) = prev.get() {
let task = p.header();
let item = &task.timer_queue_item;
if f(p) {
// Skip to next
prev = &item.next;
} else {
// Remove it
prev.set(item.next.get());
task.state.timer_dequeue();
}
}
}
}

5
embassy-executor/src/raw/util.rs
View File

@ -54,4 +54,9 @@ impl<T> SyncUnsafeCell<T> {
{
*self.value.get()
}
#[cfg(feature = "integrated-timers")]
pub unsafe fn replace(&self, value: T) -> T {
core::mem::replace(&mut *self.value.get(), value)
}
}

3
embassy-executor/tests/test.rs
View File

@ -40,9 +40,6 @@ fn setup() -> (&'static Executor, Trace) {
let trace = Trace::new();
let context = Box::leak(Box::new(trace.clone())) as *mut _ as *mut ();
let executor = &*Box::leak(Box::new(Executor::new(context)));
unsafe {
executor.initialize();
}
(executor, trace)
}

3
embassy-nrf/Cargo.toml
View File

@ -119,7 +119,7 @@ _nrf52 = ["_ppi"]
_nrf51 = ["_ppi"]
_nrf91 = []
_time-driver = ["dep:embassy-time-driver", "embassy-time-driver?/tick-hz-32_768"]
_time-driver = ["dep:embassy-time-driver", "embassy-time-driver?/tick-hz-32_768", "dep:embassy-time-queue-driver"]
# trustzone state.
_s = []
@ -135,6 +135,7 @@ _nrf52832_anomaly_109 = []
[dependencies]
embassy-time-driver = { version = "0.1", path = "../embassy-time-driver", optional = true }
embassy-time-queue-driver = { version = "0.1", path = "../embassy-time-queue-driver", optional = true }
embassy-time = { version = "0.3.2", path = "../embassy-time", optional = true }
embassy-sync = { version = "0.6.1", path = "../embassy-sync" }
embassy-hal-internal = {version = "0.2.0", path = "../embassy-hal-internal", features = ["cortex-m", "prio-bits-3"] }

117
embassy-nrf/src/time_driver.rs
View File

@ -1,11 +1,11 @@
use core::cell::Cell;
use core::sync::atomic::{compiler_fence, AtomicU32, AtomicU8, Ordering};
use core::{mem, ptr};
use core::sync::atomic::{compiler_fence, AtomicU32, Ordering};
use critical_section::CriticalSection;
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
use embassy_sync::blocking_mutex::CriticalSectionMutex as Mutex;
use embassy_time_driver::{AlarmHandle, Driver};
use embassy_time_driver::Driver;
use embassy_time_queue_driver::GlobalTimerQueue;
use crate::interrupt::InterruptExt;
use crate::{interrupt, pac};
@ -94,11 +94,6 @@ mod test {
struct AlarmState {
timestamp: Cell<u64>,
// This is really a Option<(fn(*mut ()), *mut ())>
// but fn pointers aren't allowed in const yet
callback: Cell<*const ()>,
ctx: Cell<*mut ()>,
}
unsafe impl Send for AlarmState {}
@ -107,26 +102,20 @@ impl AlarmState {
const fn new() -> Self {
Self {
timestamp: Cell::new(u64::MAX),
callback: Cell::new(ptr::null()),
ctx: Cell::new(ptr::null_mut()),
}
}
}
const ALARM_COUNT: usize = 3;
struct RtcDriver {
/// Number of 2^23 periods elapsed since boot.
period: AtomicU32,
alarm_count: AtomicU8,
/// Timestamp at which to fire alarm. u64::MAX if no alarm is scheduled.
alarms: Mutex<[AlarmState; ALARM_COUNT]>,
alarms: Mutex<AlarmState>,
}
embassy_time_driver::time_driver_impl!(static DRIVER: RtcDriver = RtcDriver {
period: AtomicU32::new(0),
alarm_count: AtomicU8::new(0),
alarms: Mutex::const_new(CriticalSectionRawMutex::new(), [const {AlarmState::new()}; ALARM_COUNT]),
alarms: Mutex::const_new(CriticalSectionRawMutex::new(), AlarmState::new()),
});
impl RtcDriver {
@ -169,13 +158,12 @@ impl RtcDriver {
self.next_period();
}
for n in 0..ALARM_COUNT {
if r.events_compare(n).read() == 1 {
r.events_compare(n).write_value(0);
critical_section::with(|cs| {
self.trigger_alarm(n, cs);
})
}
let n = 0;
if r.events_compare(n).read() == 1 {
r.events_compare(n).write_value(0);
critical_section::with(|cs| {
self.trigger_alarm(cs);
});
}
}
@ -186,75 +174,33 @@ impl RtcDriver {
self.period.store(period, Ordering::Relaxed);
let t = (period as u64) << 23;
for n in 0..ALARM_COUNT {
let alarm = &self.alarms.borrow(cs)[n];
let at = alarm.timestamp.get();
let n = 0;
let alarm = &self.alarms.borrow(cs);
let at = alarm.timestamp.get();
if at < t + 0xc00000 {
// just enable it. `set_alarm` has already set the correct CC val.
r.intenset().write(|w| w.0 = compare_n(n));
}
if at < t + 0xc00000 {
// just enable it. `set_alarm` has already set the correct CC val.
r.intenset().write(|w| w.0 = compare_n(n));
}
})
}
fn get_alarm<'a>(&'a self, cs: CriticalSection<'a>, alarm: AlarmHandle) -> &'a AlarmState {
// safety: we're allowed to assume the AlarmState is created by us, and
// we never create one that's out of bounds.
unsafe { self.alarms.borrow(cs).get_unchecked(alarm.id() as usize) }
}
fn trigger_alarm(&self, n: usize, cs: CriticalSection) {
fn trigger_alarm(&self, cs: CriticalSection) {
let n = 0;
let r = rtc();
r.intenclr().write(|w| w.0 = compare_n(n));
let alarm = &self.alarms.borrow(cs)[n];
let alarm = &self.alarms.borrow(cs);
alarm.timestamp.set(u64::MAX);
// Call after clearing alarm, so the callback can set another alarm.
// safety:
// - we can ignore the possiblity of `f` being unset (null) because of the safety contract of `allocate_alarm`.
// - other than that we only store valid function pointers into alarm.callback
let f: fn(*mut ()) = unsafe { mem::transmute(alarm.callback.get()) };
f(alarm.ctx.get());
}
}
impl Driver for RtcDriver {
fn now(&self) -> u64 {
// `period` MUST be read before `counter`, see comment at the top for details.
let period = self.period.load(Ordering::Relaxed);
compiler_fence(Ordering::Acquire);
let counter = rtc().counter().read().0;
calc_now(period, counter)
TIMER_QUEUE_DRIVER.dispatch();
}
unsafe fn allocate_alarm(&self) -> Option<AlarmHandle> {
critical_section::with(|_| {
let id = self.alarm_count.load(Ordering::Relaxed);
if id < ALARM_COUNT as u8 {
self.alarm_count.store(id + 1, Ordering::Relaxed);
Some(AlarmHandle::new(id))
} else {
None
}
})
}
fn set_alarm_callback(&self, alarm: AlarmHandle, callback: fn(*mut ()), ctx: *mut ()) {
fn set_alarm(&self, timestamp: u64) -> bool {
critical_section::with(|cs| {
let alarm = self.get_alarm(cs, alarm);
alarm.callback.set(callback as *const ());
alarm.ctx.set(ctx);
})
}
fn set_alarm(&self, alarm: AlarmHandle, timestamp: u64) -> bool {
critical_section::with(|cs| {
let n = alarm.id() as _;
let alarm = self.get_alarm(cs, alarm);
let n = 0;
let alarm = &self.alarms.borrow(cs);
alarm.timestamp.set(timestamp);
let r = rtc();
@ -304,6 +250,16 @@ impl Driver for RtcDriver {
}
}
impl Driver for RtcDriver {
fn now(&self) -> u64 {
// `period` MUST be read before `counter`, see comment at the top for details.
let period = self.period.load(Ordering::Relaxed);
compiler_fence(Ordering::Acquire);
let counter = rtc().counter().read().0;
calc_now(period, counter)
}
}
#[cfg(feature = "_nrf54l")]
#[cfg(feature = "rt")]
#[interrupt]
@ -321,3 +277,8 @@ fn RTC1() {
pub(crate) fn init(irq_prio: crate::interrupt::Priority) {
DRIVER.init(irq_prio)
}
embassy_time_queue_driver::timer_queue_impl!(
static TIMER_QUEUE_DRIVER: GlobalTimerQueue
= GlobalTimerQueue::new(|next_expiration| DRIVER.set_alarm(next_expiration))
);

3
embassy-rp/Cargo.toml
View File

@ -40,7 +40,7 @@ critical-section-impl = ["critical-section/restore-state-u8"]
unstable-pac = []
## Enable the timer for use with `embassy-time` with a 1MHz tick rate.
time-driver = ["dep:embassy-time-driver", "embassy-time-driver?/tick-hz-1_000_000"]
time-driver = ["dep:embassy-time-driver", "embassy-time-driver?/tick-hz-1_000_000", "dep:embassy-time-queue-driver"]
## Enable ROM function cache. This will store the address of a ROM function when first used, improving performance of subsequent calls.
rom-func-cache = []
@ -110,6 +110,7 @@ binary-info = ["rt", "dep:rp-binary-info", "rp-binary-info?/binary-info"]
[dependencies]
embassy-sync = { version = "0.6.1", path = "../embassy-sync" }
embassy-time-driver = { version = "0.1", path = "../embassy-time-driver", optional = true }
embassy-time-queue-driver = { version = "0.1", path = "../embassy-time-queue-driver", optional = true }
embassy-time = { version = "0.3.2", path = "../embassy-time" }
embassy-futures = { version = "0.1.0", path = "../embassy-futures" }
embassy-hal-internal = {version = "0.2.0", path = "../embassy-hal-internal", features = ["cortex-m", "prio-bits-2"] }

130
embassy-rp/src/time_driver.rs
View File

@ -1,11 +1,10 @@
//! Timer driver.
use core::cell::Cell;
use atomic_polyfill::{AtomicU8, Ordering};
use critical_section::CriticalSection;
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
use embassy_sync::blocking_mutex::Mutex;
use embassy_time_driver::{AlarmHandle, Driver};
use embassy_time_driver::Driver;
use embassy_time_queue_driver::GlobalTimerQueue;
#[cfg(feature = "rp2040")]
use pac::TIMER;
#[cfg(feature = "_rp235x")]
@ -16,23 +15,17 @@ use crate::{interrupt, pac};
struct AlarmState {
timestamp: Cell<u64>,
callback: Cell<Option<(fn(*mut ()), *mut ())>>,
}
unsafe impl Send for AlarmState {}
const ALARM_COUNT: usize = 4;
struct TimerDriver {
alarms: Mutex<CriticalSectionRawMutex, [AlarmState; ALARM_COUNT]>,
next_alarm: AtomicU8,
alarms: Mutex<CriticalSectionRawMutex, AlarmState>,
}
embassy_time_driver::time_driver_impl!(static DRIVER: TimerDriver = TimerDriver{
alarms: Mutex::const_new(CriticalSectionRawMutex::new(), [const{AlarmState {
alarms: Mutex::const_new(CriticalSectionRawMutex::new(), AlarmState {
timestamp: Cell::new(0),
callback: Cell::new(None),
}}; ALARM_COUNT]),
next_alarm: AtomicU8::new(0),
}),
});
impl Driver for TimerDriver {
@ -46,34 +39,13 @@ impl Driver for TimerDriver {
}
}
}
}
unsafe fn allocate_alarm(&self) -> Option<AlarmHandle> {
let id = self.next_alarm.fetch_update(Ordering::AcqRel, Ordering::Acquire, |x| {
if x < ALARM_COUNT as u8 {
Some(x + 1)
} else {
None
}
});
match id {
Ok(id) => Some(AlarmHandle::new(id)),
Err(_) => None,
}
}
fn set_alarm_callback(&self, alarm: AlarmHandle, callback: fn(*mut ()), ctx: *mut ()) {
let n = alarm.id() as usize;
impl TimerDriver {
fn set_alarm(&self, timestamp: u64) -> bool {
let n = 0;
critical_section::with(|cs| {
let alarm = &self.alarms.borrow(cs)[n];
alarm.callback.set(Some((callback, ctx)));
})
}
fn set_alarm(&self, alarm: AlarmHandle, timestamp: u64) -> bool {
let n = alarm.id() as usize;
critical_section::with(|cs| {
let alarm = &self.alarms.borrow(cs)[n];
let alarm = &self.alarms.borrow(cs);
alarm.timestamp.set(timestamp);
// Arm it.
@ -96,15 +68,14 @@ impl Driver for TimerDriver {
}
})
}
}
impl TimerDriver {
fn check_alarm(&self, n: usize) {
fn check_alarm(&self) {
let n = 0;
critical_section::with(|cs| {
let alarm = &self.alarms.borrow(cs)[n];
let alarm = &self.alarms.borrow(cs);
let timestamp = alarm.timestamp.get();
if timestamp <= self.now() {
self.trigger_alarm(n, cs)
self.trigger_alarm()
} else {
// Not elapsed, arm it again.
// This can happen if it was set more than 2^32 us in the future.
@ -116,17 +87,8 @@ impl TimerDriver {
TIMER.intr().write(|w| w.set_alarm(n, true));
}
fn trigger_alarm(&self, n: usize, cs: CriticalSection) {
// disarm
TIMER.armed().write(|w| w.set_armed(1 << n));
let alarm = &self.alarms.borrow(cs)[n];
alarm.timestamp.set(u64::MAX);
// Call after clearing alarm, so the callback can set another alarm.
if let Some((f, ctx)) = alarm.callback.get() {
f(ctx);
}
fn trigger_alarm(&self) {
TIMER_QUEUE_DRIVER.dispatch();
}
}
@ -134,79 +96,37 @@ impl TimerDriver {
pub unsafe fn init() {
// init alarms
critical_section::with(|cs| {
let alarms = DRIVER.alarms.borrow(cs);
for a in alarms {
a.timestamp.set(u64::MAX);
}
let alarm = DRIVER.alarms.borrow(cs);
alarm.timestamp.set(u64::MAX);
});
// enable all irqs
// enable irq
TIMER.inte().write(|w| {
w.set_alarm(0, true);
w.set_alarm(1, true);
w.set_alarm(2, true);
w.set_alarm(3, true);
});
#[cfg(feature = "rp2040")]
{
interrupt::TIMER_IRQ_0.enable();
interrupt::TIMER_IRQ_1.enable();
interrupt::TIMER_IRQ_2.enable();
interrupt::TIMER_IRQ_3.enable();
}
#[cfg(feature = "_rp235x")]
{
interrupt::TIMER0_IRQ_0.enable();
interrupt::TIMER0_IRQ_1.enable();
interrupt::TIMER0_IRQ_2.enable();
interrupt::TIMER0_IRQ_3.enable();
}
}
#[cfg(all(feature = "rt", feature = "rp2040"))]
#[interrupt]
fn TIMER_IRQ_0() {
DRIVER.check_alarm(0)
}
#[cfg(all(feature = "rt", feature = "rp2040"))]
#[interrupt]
fn TIMER_IRQ_1() {
DRIVER.check_alarm(1)
}
#[cfg(all(feature = "rt", feature = "rp2040"))]
#[interrupt]
fn TIMER_IRQ_2() {
DRIVER.check_alarm(2)
}
#[cfg(all(feature = "rt", feature = "rp2040"))]
#[interrupt]
fn TIMER_IRQ_3() {
DRIVER.check_alarm(3)
DRIVER.check_alarm()
}
#[cfg(all(feature = "rt", feature = "_rp235x"))]
#[interrupt]
fn TIMER0_IRQ_0() {
DRIVER.check_alarm(0)
DRIVER.check_alarm()
}
#[cfg(all(feature = "rt", feature = "_rp235x"))]
#[interrupt]
fn TIMER0_IRQ_1() {
DRIVER.check_alarm(1)
}
#[cfg(all(feature = "rt", feature = "_rp235x"))]
#[interrupt]
fn TIMER0_IRQ_2() {
DRIVER.check_alarm(2)
}
#[cfg(all(feature = "rt", feature = "_rp235x"))]
#[interrupt]
fn TIMER0_IRQ_3() {
DRIVER.check_alarm(3)
}
embassy_time_queue_driver::timer_queue_impl!(
static TIMER_QUEUE_DRIVER: GlobalTimerQueue
= GlobalTimerQueue::new(|next_expiration| DRIVER.set_alarm(next_expiration))
);

4
embassy-stm32/Cargo.toml
View File

@ -45,6 +45,7 @@ rustdoc-args = ["--cfg", "docsrs"]
embassy-sync = { version = "0.6.1", path = "../embassy-sync" }
embassy-time = { version = "0.3.2", path = "../embassy-time", optional = true }
embassy-time-driver = { version = "0.1", path = "../embassy-time-driver", optional = true }
embassy-time-queue-driver = { version = "0.1", path = "../embassy-time-queue-driver", optional = true }
embassy-futures = { version = "0.1.0", path = "../embassy-futures" }
embassy-hal-internal = {version = "0.2.0", path = "../embassy-hal-internal", features = ["cortex-m", "prio-bits-4"] }
embassy-embedded-hal = {version = "0.2.0", path = "../embassy-embedded-hal", default-features = false }
@ -125,6 +126,7 @@ defmt = [
exti = []
low-power = [ "dep:embassy-executor", "embassy-executor?/arch-cortex-m", "time" ]
low-power-debug-with-sleep = []
integrated-timers = ["dep:embassy-executor", "_time-driver"]
## Automatically generate `memory.x` file using [`stm32-metapac`](https://docs.rs/stm32-metapac/)
memory-x = ["stm32-metapac/memory-x"]
@ -149,7 +151,7 @@ time = ["dep:embassy-time", "embassy-embedded-hal/time"]
# Features starting with `_` are for internal use only. They're not intended
# to be enabled by other crates, and are not covered by semver guarantees.
_time-driver = ["dep:embassy-time-driver", "time"]
_time-driver = ["dep:embassy-time-driver", "time", "dep:embassy-time-queue-driver"]
## Use any time driver
time-driver-any = ["_time-driver"]

3
embassy-stm32/src/low_power.rs
View File

@ -256,9 +256,6 @@ impl Executor {
/// This function never returns.
pub fn run(&'static mut self, init: impl FnOnce(Spawner)) -> ! {
let executor = unsafe { EXECUTOR.as_mut().unwrap() };
unsafe {
executor.inner.initialize();
}
init(executor.inner.spawner());
loop {

154
embassy-stm32/src/time_driver.rs
View File

@ -1,13 +1,13 @@
#![allow(non_snake_case)]
use core::cell::Cell;
use core::sync::atomic::{compiler_fence, AtomicU32, AtomicU8, Ordering};
use core::{mem, ptr};
use core::sync::atomic::{compiler_fence, AtomicU32, Ordering};
use critical_section::CriticalSection;
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
use embassy_sync::blocking_mutex::Mutex;
use embassy_time_driver::{AlarmHandle, Driver, TICK_HZ};
use embassy_time_driver::{Driver, TICK_HZ};
use embassy_time_queue_driver::GlobalTimerQueue;
use stm32_metapac::timer::{regs, TimGp16};
use crate::interrupt::typelevel::Interrupt;
@ -24,18 +24,6 @@ use crate::{interrupt, peripherals};
// additional CC capabilities to provide timer alarms to embassy-time. embassy-time requires AT LEAST
// one alarm to be allocatable, which means timers that only have CC1, such as TIM16/TIM17, are not
// candidates for use as an embassy-time driver provider. (a.k.a 1CH and 1CH_CMP are not, others are good.)
//
// The values of ALARM_COUNT below are not the TOTAL CC registers available, but rather the number
// available after reserving CC1 for regular time keeping. For example, TIM2 has four CC registers:
// CC1, CC2, CC3, and CC4, so it can provide ALARM_COUNT = 3.
cfg_if::cfg_if! {
if #[cfg(any(time_driver_tim9, time_driver_tim12, time_driver_tim15, time_driver_tim21, time_driver_tim22))] {
const ALARM_COUNT: usize = 1;
} else {
const ALARM_COUNT: usize = 3;
}
}
#[cfg(time_driver_tim1)]
type T = peripherals::TIM1;
@ -208,11 +196,6 @@ fn calc_now(period: u32, counter: u16) -> u64 {
struct AlarmState {
timestamp: Cell<u64>,
// This is really a Option<(fn(*mut ()), *mut ())>
// but fn pointers aren't allowed in const yet
callback: Cell<*const ()>,
ctx: Cell<*mut ()>,
}
unsafe impl Send for AlarmState {}
@ -221,8 +204,6 @@ impl AlarmState {
const fn new() -> Self {
Self {
timestamp: Cell::new(u64::MAX),
callback: Cell::new(ptr::null()),
ctx: Cell::new(ptr::null_mut()),
}
}
}
@ -230,17 +211,14 @@ impl AlarmState {
pub(crate) struct RtcDriver {
/// Number of 2^15 periods elapsed since boot.
period: AtomicU32,
alarm_count: AtomicU8,
/// Timestamp at which to fire alarm. u64::MAX if no alarm is scheduled.
alarms: Mutex<CriticalSectionRawMutex, [AlarmState; ALARM_COUNT]>,
alarm: Mutex<CriticalSectionRawMutex, AlarmState>,
#[cfg(feature = "low-power")]
rtc: Mutex<CriticalSectionRawMutex, Cell<Option<&'static Rtc>>>,
}
embassy_time_driver::time_driver_impl!(static DRIVER: RtcDriver = RtcDriver {
period: AtomicU32::new(0),
alarm_count: AtomicU8::new(0),
alarms: Mutex::const_new(CriticalSectionRawMutex::new(), [const{AlarmState::new()}; ALARM_COUNT]),
alarm: Mutex::const_new(CriticalSectionRawMutex::new(), AlarmState::new()),
#[cfg(feature = "low-power")]
rtc: Mutex::const_new(CriticalSectionRawMutex::new(), Cell::new(None)),
});
@ -289,7 +267,7 @@ impl RtcDriver {
let r = regs_gp16();
// XXX: reduce the size of this critical section ?
critical_section::with(|cs| {
critical_section::with(|_cs| {
let sr = r.sr().read();
let dier = r.dier().read();
@ -308,10 +286,9 @@ impl RtcDriver {
self.next_period();
}
for n in 0..ALARM_COUNT {
if sr.ccif(n + 1) && dier.ccie(n + 1) {
self.trigger_alarm(n, cs);
}
let n = 0;
if sr.ccif(n + 1) && dier.ccie(n + 1) {
self.trigger_alarm();
}
})
}
@ -326,36 +303,20 @@ impl RtcDriver {
critical_section::with(move |cs| {
r.dier().modify(move |w| {
for n in 0..ALARM_COUNT {
let alarm = &self.alarms.borrow(cs)[n];
let at = alarm.timestamp.get();
let n = 0;
let alarm = self.alarm.borrow(cs);
let at = alarm.timestamp.get();
if at < t + 0xc000 {
// just enable it. `set_alarm` has already set the correct CCR val.
w.set_ccie(n + 1, true);
}
if at < t + 0xc000 {
// just enable it. `set_alarm` has already set the correct CCR val.
w.set_ccie(n + 1, true);
}
})
})
}
fn get_alarm<'a>(&'a self, cs: CriticalSection<'a>, alarm: AlarmHandle) -> &'a AlarmState {
// safety: we're allowed to assume the AlarmState is created by us, and
// we never create one that's out of bounds.
unsafe { self.alarms.borrow(cs).get_unchecked(alarm.id() as usize) }
}
fn trigger_alarm(&self, n: usize, cs: CriticalSection) {
let alarm = &self.alarms.borrow(cs)[n];
alarm.timestamp.set(u64::MAX);
// Call after clearing alarm, so the callback can set another alarm.
// safety:
// - we can ignore the possibility of `f` being unset (null) because of the safety contract of `allocate_alarm`.
// - other than that we only store valid function pointers into alarm.callback
let f: fn(*mut ()) = unsafe { mem::transmute(alarm.callback.get()) };
f(alarm.ctx.get());
fn trigger_alarm(&self) {
TIMER_QUEUE_DRIVER.dispatch();
}
/*
@ -367,14 +328,7 @@ impl RtcDriver {
fn time_until_next_alarm(&self, cs: CriticalSection) -> embassy_time::Duration {
let now = self.now() + 32;
embassy_time::Duration::from_ticks(
self.alarms
.borrow(cs)
.iter()
.map(|alarm: &AlarmState| alarm.timestamp.get().saturating_sub(now))
.min()
.unwrap_or(u64::MAX),
)
embassy_time::Duration::from_ticks(self.alarm.borrow(cs).timestamp.get().saturating_sub(now))
}
#[cfg(feature = "low-power")]
@ -409,15 +363,12 @@ impl RtcDriver {
self.period.store(period, Ordering::SeqCst);
regs_gp16().cnt().write(|w| w.set_cnt(cnt as u16));
// Now, recompute all alarms
for i in 0..self.alarm_count.load(Ordering::Relaxed) as usize {
let alarm_handle = unsafe { AlarmHandle::new(i as u8) };
let alarm = self.get_alarm(cs, alarm_handle);
// Now, recompute alarm
let alarm = self.alarm.borrow(cs);
if !self.set_alarm(alarm_handle, alarm.timestamp.get()) {
// If the alarm timestamp has passed, we need to trigger it
self.trigger_alarm(i, cs);
}
if !self.set_alarm(alarm.timestamp.get()) {
// If the alarm timestamp has passed, we need to trigger it
self.trigger_alarm();
}
}
@ -489,46 +440,13 @@ impl RtcDriver {
regs_gp16().cr1().modify(|w| w.set_cen(true));
})
}
}
impl Driver for RtcDriver {
fn now(&self) -> u64 {
let r = regs_gp16();
let period = self.period.load(Ordering::Relaxed);
compiler_fence(Ordering::Acquire);
let counter = r.cnt().read().cnt();
calc_now(period, counter)
}
unsafe fn allocate_alarm(&self) -> Option<AlarmHandle> {
critical_section::with(|_| {
let id = self.alarm_count.load(Ordering::Relaxed);
if id < ALARM_COUNT as u8 {
self.alarm_count.store(id + 1, Ordering::Relaxed);
Some(AlarmHandle::new(id))
} else {
None
}
})
}
fn set_alarm_callback(&self, alarm: AlarmHandle, callback: fn(*mut ()), ctx: *mut ()) {
critical_section::with(|cs| {
let alarm = self.get_alarm(cs, alarm);
alarm.callback.set(callback as *const ());
alarm.ctx.set(ctx);
})
}
fn set_alarm(&self, alarm: AlarmHandle, timestamp: u64) -> bool {
fn set_alarm(&self, timestamp: u64) -> bool {
critical_section::with(|cs| {
let r = regs_gp16();
let n = alarm.id() as usize;
let alarm = self.get_alarm(cs, alarm);
alarm.timestamp.set(timestamp);
let n = 0;
self.alarm.borrow(cs).timestamp.set(timestamp);
let t = self.now();
if timestamp <= t {
@ -536,7 +454,7 @@ impl Driver for RtcDriver {
// Disarm the alarm and return `false` to indicate that.
r.dier().modify(|w| w.set_ccie(n + 1, false));
alarm.timestamp.set(u64::MAX);
self.alarm.borrow(cs).timestamp.set(u64::MAX);
return false;
}
@ -558,7 +476,7 @@ impl Driver for RtcDriver {
// It is the caller's responsibility to handle this ambiguity.
r.dier().modify(|w| w.set_ccie(n + 1, false));
alarm.timestamp.set(u64::MAX);
self.alarm.borrow(cs).timestamp.set(u64::MAX);
return false;
}
@ -569,6 +487,17 @@ impl Driver for RtcDriver {
}
}
impl Driver for RtcDriver {
fn now(&self) -> u64 {
let r = regs_gp16();
let period = self.period.load(Ordering::Relaxed);
compiler_fence(Ordering::Acquire);
let counter = r.cnt().read().cnt();
calc_now(period, counter)
}
}
#[cfg(feature = "low-power")]
pub(crate) fn get_driver() -> &'static RtcDriver {
&DRIVER
@ -577,3 +506,8 @@ pub(crate) fn get_driver() -> &'static RtcDriver {
pub(crate) fn init(cs: CriticalSection) {
DRIVER.init(cs)
}
embassy_time_queue_driver::timer_queue_impl!(
static TIMER_QUEUE_DRIVER: GlobalTimerQueue
= GlobalTimerQueue::new(|next_expiration| DRIVER.set_alarm(next_expiration))
);

135
embassy-time-driver/src/lib.rs
View File

@ -21,8 +21,8 @@
//!
//! Instead of the usual "trait + generic params" approach, calls from embassy to the driver are done via `extern` functions.
//!
//! `embassy` internally defines the driver functions as `extern "Rust" { fn _embassy_time_now() -> u64; }` and calls them.
//! The driver crate defines the functions as `#[no_mangle] fn _embassy_time_now() -> u64`. The linker will resolve the
//! `embassy` internally defines the driver function as `extern "Rust" { fn _embassy_time_now() -> u64; }` and calls it.
//! The driver crate defines the function as `#[no_mangle] fn _embassy_time_now() -> u64`. The linker will resolve the
//! calls from the `embassy` crate to call into the driver crate.
//!
//! If there is none or multiple drivers in the crate tree, linking will fail.
@ -38,7 +38,7 @@
//! # Example
//!
//! ```
//! use embassy_time_driver::{Driver, AlarmHandle};
//! use embassy_time_driver::Driver;
//!
//! struct MyDriver{} // not public!
//!
@ -46,15 +46,6 @@
//! fn now(&self) -> u64 {
//! todo!()
//! }
//! unsafe fn allocate_alarm(&self) -> Option<AlarmHandle> {
//! todo!()
//! }
//! fn set_alarm_callback(&self, alarm: AlarmHandle, callback: fn(*mut ()), ctx: *mut ()) {
//! todo!()
//! }
//! fn set_alarm(&self, alarm: AlarmHandle, timestamp: u64) -> bool {
//! todo!()
//! }
//! }
//!
//! embassy_time_driver::time_driver_impl!(static DRIVER: MyDriver = MyDriver{});
@ -70,28 +61,6 @@ mod tick;
/// This value is specified by the [`tick-*` Cargo features](crate#tick-rate)
pub const TICK_HZ: u64 = tick::TICK_HZ;
/// Alarm handle, assigned by the driver.
#[derive(Clone, Copy)]
pub struct AlarmHandle {
id: u8,
}
impl AlarmHandle {
/// Create an AlarmHandle
///
/// Safety: May only be called by the current global Driver impl.
/// The impl is allowed to rely on the fact that all `AlarmHandle` instances
/// are created by itself in unsafe code (e.g. indexing operations)
pub unsafe fn new(id: u8) -> Self {
Self { id }
}
/// Get the ID of the AlarmHandle.
pub fn id(&self) -> u8 {
self.id
}
}
/// Time driver
pub trait Driver: Send + Sync + 'static {
/// Return the current timestamp in ticks.
@ -105,76 +74,10 @@ pub trait Driver: Send + Sync + 'static {
/// you MUST extend them to 64-bit, for example by counting overflows in software,
/// or chaining multiple timers together.
fn now(&self) -> u64;
/// Try allocating an alarm handle. Returns None if no alarms left.
/// Initially the alarm has no callback set, and a null `ctx` pointer.
///
/// The allocated alarm is a reusable resource and can be used multiple times.
/// Once the alarm has fired, it remains allocated and can be set again without needing
/// to be reallocated.
///
/// # Safety
/// It is UB to make the alarm fire before setting a callback.
unsafe fn allocate_alarm(&self) -> Option<AlarmHandle>;
/// Set the callback function to be called when the alarm triggers.
/// The callback may be called from any context (interrupt or thread mode).
///
/// The callback is maintained after the alarm has fired. Callers do not need
/// to set a callback again before setting another alarm, unless they want to
/// change the callback function or context.
fn set_alarm_callback(&self, alarm: AlarmHandle, callback: fn(*mut ()), ctx: *mut ());
/// Set an alarm at the given timestamp.
///
/// ## Behavior
///
/// If `timestamp` is in the future, `set_alarm` schedules calling the callback function
/// at that time, and returns `true`.
///
/// If `timestamp` is in the past, `set_alarm` has two allowed behaviors. Implementations can pick whether to:
///
/// - Schedule calling the callback function "immediately", as if the requested timestamp was "now+epsilon" and return `true`, or
/// - Not schedule the callback, and return `false`.
///
/// Callers must ensure to behave correctly with either behavior.
///
/// When callback is called, it is guaranteed that `now()` will return a value greater than or equal to `timestamp`.
///
/// ## Reentrancy
///
/// Calling the callback from `set_alarm` synchronously is not allowed. If the implementation chooses the first option above,
/// it must still call the callback from another context (i.e. an interrupt handler or background thread), it's not allowed
/// to call it synchronously in the context `set_alarm` is running.
///
/// The reason for the above is callers are explicitly permitted to do both of:
/// - Lock a mutex in the alarm callback.
/// - Call `set_alarm` while having that mutex locked.
///
/// If `set_alarm` called the callback synchronously, it'd cause a deadlock or panic because it'd cause the
/// mutex to be locked twice reentrantly in the same context.
///
/// ## Overwriting alarms
///
/// Only one alarm can be active at a time for each `AlarmHandle`. This overwrites any previously-set alarm if any.
///
/// ## Unsetting the alarm
///
/// There is no `unset_alarm` API. Instead, callers can call `set_alarm` with `timestamp` set to `u64::MAX`.
///
/// This allows for more efficient implementations, since they don't need to distinguish between the "alarm set" and
/// "alarm not set" cases, thanks to the fact "Alarm set for u64::MAX" is effectively equivalent for "alarm not set".
///
/// This means implementations need to be careful to avoid timestamp overflows. The recommendation is to make `timestamp`
/// be in the same units as hardware ticks to avoid any conversions, which makes avoiding overflow easier.
fn set_alarm(&self, alarm: AlarmHandle, timestamp: u64) -> bool;
}
extern "Rust" {
fn _embassy_time_now() -> u64;
fn _embassy_time_allocate_alarm() -> Option<AlarmHandle>;
fn _embassy_time_set_alarm_callback(alarm: AlarmHandle, callback: fn(*mut ()), ctx: *mut ());
fn _embassy_time_set_alarm(alarm: AlarmHandle, timestamp: u64) -> bool;
}
/// See [`Driver::now`]
@ -182,23 +85,6 @@ pub fn now() -> u64 {
unsafe { _embassy_time_now() }
}
/// See [`Driver::allocate_alarm`]
///
/// Safety: it is UB to make the alarm fire before setting a callback.
pub unsafe fn allocate_alarm() -> Option<AlarmHandle> {
_embassy_time_allocate_alarm()
}
/// See [`Driver::set_alarm_callback`]
pub fn set_alarm_callback(alarm: AlarmHandle, callback: fn(*mut ()), ctx: *mut ()) {
unsafe { _embassy_time_set_alarm_callback(alarm, callback, ctx) }
}
/// See [`Driver::set_alarm`]
pub fn set_alarm(alarm: AlarmHandle, timestamp: u64) -> bool {
unsafe { _embassy_time_set_alarm(alarm, timestamp) }
}
/// Set the time Driver implementation.
///
/// See the module documentation for an example.
@ -211,20 +97,5 @@ macro_rules! time_driver_impl {
fn _embassy_time_now() -> u64 {
<$t as $crate::Driver>::now(&$name)
}
#[no_mangle]
unsafe fn _embassy_time_allocate_alarm() -> Option<$crate::AlarmHandle> {
<$t as $crate::Driver>::allocate_alarm(&$name)
}
#[no_mangle]
fn _embassy_time_set_alarm_callback(alarm: $crate::AlarmHandle, callback: fn(*mut ()), ctx: *mut ()) {
<$t as $crate::Driver>::set_alarm_callback(&$name, alarm, callback, ctx)
}
#[no_mangle]
fn _embassy_time_set_alarm(alarm: $crate::AlarmHandle, timestamp: u64) -> bool {
<$t as $crate::Driver>::set_alarm(&$name, alarm, timestamp)
}
};
}

33
embassy-time-queue-driver/Cargo.toml
View File

@ -20,6 +20,39 @@ categories = [
# This is especially common when mixing crates from crates.io and git.
links = "embassy-time-queue"
[dependencies]
critical-section = "1.2.0"
heapless = "0.8"
embassy-executor = { version = "0.6.3", path = "../embassy-executor", optional = true }
embassy-time-driver = { version = "0.1.0", path = "../embassy-time-driver" }
[features]
#! ### Generic Queue
## Use the executor-integrated `embassy-time` timer queue. The timer items are stored inside
## the task headers, so you do not need to set a capacity for the queue.
## To use this you must have a time driver provided.
##
## If this feature is not enabled, a generic queue is available with a configurable capacity.
integrated-timers = ["embassy-executor/integrated-timers"]
#! The following features set how many timers are used for the generic queue. At most one
#! `generic-queue-*` feature can be enabled. If none is enabled, a default of 64 timers is used.
#!
#! When using embassy-time from libraries, you should *not* enable any `generic-queue-*` feature, to allow the
#! end user to pick.
## Generic Queue with 8 timers
generic-queue-8 = []
## Generic Queue with 16 timers
generic-queue-16 = []
## Generic Queue with 32 timers
generic-queue-32 = []
## Generic Queue with 64 timers
generic-queue-64 = []
## Generic Queue with 128 timers
generic-queue-128 = []
[package.metadata.embassy_docs]
src_base = "https://github.com/embassy-rs/embassy/blob/embassy-time-queue-driver-v$VERSION/embassy-time-queue-driver/src/"
src_base_git = "https://github.com/embassy-rs/embassy/blob/$COMMIT/embassy-time-queue-driver/src/"

136
embassy-time-queue-driver/src/lib.rs
View File

@ -6,7 +6,29 @@
//!
//! - Define a struct `MyTimerQueue`
//! - Implement [`TimerQueue`] for it
//! - Register it as the global timer queue with [`timer_queue_impl`](crate::timer_queue_impl).
//! - Register it as the global timer queue with [`timer_queue_impl`].
//! - Ensure that you process the timer queue when `schedule_wake` is due. This usually involves
//! waking expired tasks, finding the next expiration time and setting an alarm.
//!
//! If a single global timer queue is sufficient for you, you can use the
//! [`GlobalTimerQueue`] type, which is a wrapper around a global timer queue
//! protected by a critical section.
//!
//! ```
//! use embassy_time_queue_driver::GlobalTimerQueue;
//! embassy_time_queue_driver::timer_queue_impl!(
//! static TIMER_QUEUE_DRIVER: GlobalTimerQueue
//! = GlobalTimerQueue::new(|next_expiration| todo!("Set an alarm"))
//! );
//! ```
//!
//! You can also use the `queue_generic` or the `embassy_executor::raw::timer_queue` modules to
//! implement your own timer queue. These modules contain queue implementations which you can wrap
//! and tailor to your needs.
//!
//! If you are providing an embassy-executor implementation besides a timer queue, you can choose to
//! expose the `integrated-timers` feature in your implementation. This feature stores timer items
//! in the tasks themselves, so you don't need a fixed-size queue or dynamic memory allocation.
//!
//! ## Example
//!
@ -14,7 +36,7 @@
//! use core::task::Waker;
//!
//! use embassy_time::Instant;
//! use embassy_time::queue::{TimerQueue};
//! use embassy_time::queue::TimerQueue;
//!
//! struct MyTimerQueue{}; // not public!
//!
@ -26,11 +48,18 @@
//!
//! embassy_time_queue_driver::timer_queue_impl!(static QUEUE: MyTimerQueue = MyTimerQueue{});
//! ```
pub mod queue_generic;
use core::cell::RefCell;
use core::task::Waker;
use critical_section::Mutex;
/// Timer queue
pub trait TimerQueue {
/// Schedules a waker in the queue to be awoken at moment `at`.
///
/// If this moment is in the past, the waker might be awoken immediately.
fn schedule_wake(&'static self, at: u64, waker: &Waker);
}
@ -58,3 +87,106 @@ macro_rules! timer_queue_impl {
}
};
}
#[cfg(feature = "integrated-timers")]
type InnerQueue = embassy_executor::raw::timer_queue::TimerQueue;
#[cfg(not(feature = "integrated-timers"))]
type InnerQueue = queue_generic::Queue;
/// A timer queue implementation that can be used as a global timer queue.
///
/// This implementation is not thread-safe, and should be protected by a mutex of some sort.
pub struct GenericTimerQueue<F: Fn(u64) -> bool> {
queue: InnerQueue,
set_alarm: F,
}
impl<F: Fn(u64) -> bool> GenericTimerQueue<F> {
/// Creates a new timer queue.
///
/// `set_alarm` is a function that should set the next alarm time. The function should
/// return `true` if the alarm was set, and `false` if the alarm was in the past.
pub const fn new(set_alarm: F) -> Self {
Self {
queue: InnerQueue::new(),
set_alarm,
}
}
/// Schedules a task to run at a specific time, and returns whether any changes were made.
pub fn schedule_wake(&mut self, at: u64, waker: &core::task::Waker) {
#[cfg(feature = "integrated-timers")]
let waker = embassy_executor::raw::task_from_waker(waker);
if self.queue.schedule_wake(at, waker) {
self.dispatch()
}
}
/// Dequeues expired timers and returns the next alarm time.
pub fn next_expiration(&mut self, now: u64) -> u64 {
self.queue.next_expiration(now)
}
/// Handle the alarm.
///
/// Call this function when the next alarm is due.
pub fn dispatch(&mut self) {
let mut next_expiration = self.next_expiration(embassy_time_driver::now());
while !(self.set_alarm)(next_expiration) {
// next_expiration is in the past, dequeue and find a new expiration
next_expiration = self.next_expiration(next_expiration);
}
}
}
/// A [`GenericTimerQueue`] protected by a critical section. Directly useable as a [`TimerQueue`].
pub struct GlobalTimerQueue {
inner: Mutex<RefCell<GenericTimerQueue<fn(u64) -> bool>>>,
}
impl GlobalTimerQueue {
/// Creates a new timer queue.
///
/// `set_alarm` is a function that should set the next alarm time. The function should
/// return `true` if the alarm was set, and `false` if the alarm was in the past.
pub const fn new(set_alarm: fn(u64) -> bool) -> Self {
Self {
inner: Mutex::new(RefCell::new(GenericTimerQueue::new(set_alarm))),
}
}
/// Schedules a task to run at a specific time, and returns whether any changes were made.
pub fn schedule_wake(&self, at: u64, waker: &core::task::Waker) {
critical_section::with(|cs| {
let mut inner = self.inner.borrow_ref_mut(cs);
inner.schedule_wake(at, waker);
});
}
/// Dequeues expired timers and returns the next alarm time.
pub fn next_expiration(&self, now: u64) -> u64 {
critical_section::with(|cs| {
let mut inner = self.inner.borrow_ref_mut(cs);
inner.next_expiration(now)
})
}
/// Handle the alarm.
///
/// Call this function when the next alarm is due.
pub fn dispatch(&self) {
critical_section::with(|cs| {
let mut inner = self.inner.borrow_ref_mut(cs);
inner.dispatch()
})
}
}
impl TimerQueue for GlobalTimerQueue {
fn schedule_wake(&'static self, at: u64, waker: &Waker) {
GlobalTimerQueue::schedule_wake(self, at, waker)
}
}

146
embassy-time-queue-driver/src/queue_generic.rs Normal file
View File

@ -0,0 +1,146 @@
//! Generic timer queue implementations.
//!
//! Time queue drivers may use this to simplify their implementation.
use core::cmp::{min, Ordering};
use core::task::Waker;
use heapless::Vec;
#[derive(Debug)]
struct Timer {
at: u64,
waker: Waker,
}
impl PartialEq for Timer {
fn eq(&self, other: &Self) -> bool {
self.at == other.at
}
}
impl Eq for Timer {}
impl PartialOrd for Timer {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
self.at.partial_cmp(&other.at)
}
}
impl Ord for Timer {
fn cmp(&self, other: &Self) -> Ordering {
self.at.cmp(&other.at)
}
}
/// A timer queue with a pre-determined capacity.
pub struct ConstGenericQueue<const QUEUE_SIZE: usize> {
queue: Vec<Timer, QUEUE_SIZE>,
}
impl<const QUEUE_SIZE: usize> ConstGenericQueue<QUEUE_SIZE> {
/// Creates a new timer queue.
pub const fn new() -> Self {
Self { queue: Vec::new() }
}
/// Schedules a task to run at a specific time, and returns whether any changes were made.
///
/// If this function returns `true`, the called should find the next expiration time and set
/// a new alarm for that time.
pub fn schedule_wake(&mut self, at: u64, waker: &Waker) -> bool {
self.queue
.iter_mut()
.find(|timer| timer.waker.will_wake(waker))
.map(|timer| {
if timer.at > at {
timer.at = at;
true
} else {
false
}
})
.unwrap_or_else(|| {
let mut timer = Timer {
waker: waker.clone(),
at,
};
loop {
match self.queue.push(timer) {
Ok(()) => break,
Err(e) => timer = e,
}
self.queue.pop().unwrap().waker.wake();
}
true
})
}
/// Dequeues expired timers and returns the next alarm time.
pub fn next_expiration(&mut self, now: u64) -> u64 {
let mut next_alarm = u64::MAX;
let mut i = 0;
while i < self.queue.len() {
let timer = &self.queue[i];
if timer.at <= now {
let timer = self.queue.swap_remove(i);
timer.waker.wake();
} else {
next_alarm = min(next_alarm, timer.at);
i += 1;
}
}
next_alarm
}
}
#[cfg(feature = "generic-queue-8")]
const QUEUE_SIZE: usize = 8;
#[cfg(feature = "generic-queue-16")]
const QUEUE_SIZE: usize = 16;
#[cfg(feature = "generic-queue-32")]
const QUEUE_SIZE: usize = 32;
#[cfg(feature = "generic-queue-64")]
const QUEUE_SIZE: usize = 64;
#[cfg(feature = "generic-queue-128")]
const QUEUE_SIZE: usize = 128;
#[cfg(not(any(
feature = "generic-queue-8",
feature = "generic-queue-16",
feature = "generic-queue-32",
feature = "generic-queue-64",
feature = "generic-queue-128"
)))]
const QUEUE_SIZE: usize = 64;
/// A timer queue with a pre-determined capacity.
pub struct Queue {
queue: ConstGenericQueue<QUEUE_SIZE>,
}
impl Queue {
/// Creates a new timer queue.
pub const fn new() -> Self {
Self {
queue: ConstGenericQueue::new(),
}
}
/// Schedules a task to run at a specific time, and returns whether any changes were made.
///
/// If this function returns `true`, the called should find the next expiration time and set
/// a new alarm for that time.
pub fn schedule_wake(&mut self, at: u64, waker: &Waker) -> bool {
self.queue.schedule_wake(at, waker)
}
/// Dequeues expired timers and returns the next alarm time.
pub fn next_expiration(&mut self, now: u64) -> u64 {
self.queue.next_expiration(now)
}
}

24
embassy-time/Cargo.toml
View File

@ -42,29 +42,6 @@ defmt-timestamp-uptime-tus = ["defmt"]
## Create a `MockDriver` that can be manually advanced for testing purposes.
mock-driver = ["tick-hz-1_000_000"]
#! ### Generic Queue
## Create a global, generic queue that can be used with any executor.
## To use this you must have a time driver provided.
generic-queue = []
#! The following features set how many timers are used for the generic queue. At most one
#! `generic-queue-*` feature can be enabled. If none is enabled, a default of 64 timers is used.
#!
#! When using embassy-time from libraries, you should *not* enable any `generic-queue-*` feature, to allow the
#! end user to pick.
## Generic Queue with 8 timers
generic-queue-8 = ["generic-queue"]
## Generic Queue with 16 timers
generic-queue-16 = ["generic-queue"]
## Generic Queue with 32 timers
generic-queue-32 = ["generic-queue"]
## Generic Queue with 64 timers
generic-queue-64 = ["generic-queue"]
## Generic Queue with 128 timers
generic-queue-128 = ["generic-queue"]
#! ### Tick Rate
#!
#! At most 1 `tick-*` feature can be enabled. If none is enabled, a default of 1MHz is used.
@ -419,7 +396,6 @@ embedded-hal-async = { version = "1.0" }
futures-util = { version = "0.3.17", default-features = false }
critical-section = "1.1"
cfg-if = "1.0.0"
heapless = "0.8"
document-features = "0.2.7"

85
embassy-time/src/driver_mock.rs
View File

@ -1,7 +1,7 @@
use core::cell::RefCell;
use critical_section::Mutex as CsMutex;
use embassy_time_driver::{AlarmHandle, Driver};
use embassy_time_driver::Driver;
use crate::{Duration, Instant};
@ -60,15 +60,13 @@ impl MockDriver {
let now = inner.now.as_ticks();
inner
.alarm
.as_mut()
.filter(|alarm| alarm.timestamp <= now)
.map(|alarm| {
alarm.timestamp = u64::MAX;
if inner.alarm.timestamp <= now {
inner.alarm.timestamp = u64::MAX;
(alarm.callback, alarm.ctx)
})
Some((inner.alarm.callback, inner.alarm.ctx))
} else {
None
}
})
};
@ -76,68 +74,48 @@ impl MockDriver {
(callback)(ctx);
}
}
}
impl Driver for MockDriver {
fn now(&self) -> u64 {
critical_section::with(|cs| self.0.borrow_ref(cs).now).as_ticks()
}
unsafe fn allocate_alarm(&self) -> Option<AlarmHandle> {
/// Configures a callback to be called when the alarm fires.
pub fn set_alarm_callback(&self, callback: fn(*mut ()), ctx: *mut ()) {
critical_section::with(|cs| {
let mut inner = self.0.borrow_ref_mut(cs);
if inner.alarm.is_some() {
None
} else {
inner.alarm.replace(AlarmState::new());
Some(AlarmHandle::new(0))
}
})
}
fn set_alarm_callback(&self, _alarm: AlarmHandle, callback: fn(*mut ()), ctx: *mut ()) {
critical_section::with(|cs| {
let mut inner = self.0.borrow_ref_mut(cs);
let Some(alarm) = inner.alarm.as_mut() else {
panic!("Alarm not allocated");
};
alarm.callback = callback;
alarm.ctx = ctx;
inner.alarm.callback = callback;
inner.alarm.ctx = ctx;
});
}
fn set_alarm(&self, _alarm: AlarmHandle, timestamp: u64) -> bool {
/// Sets the alarm to fire at the specified timestamp.
pub fn set_alarm(&self, timestamp: u64) -> bool {
critical_section::with(|cs| {
let mut inner = self.0.borrow_ref_mut(cs);
if timestamp <= inner.now.as_ticks() {
false
} else {
let Some(alarm) = inner.alarm.as_mut() else {
panic!("Alarm not allocated");
};
alarm.timestamp = timestamp;
inner.alarm.timestamp = timestamp;
true
}
})
}
}
impl Driver for MockDriver {
fn now(&self) -> u64 {
critical_section::with(|cs| self.0.borrow_ref(cs).now).as_ticks()
}
}
struct InnerMockDriver {
now: Instant,
alarm: Option<AlarmState>,
alarm: AlarmState,
}
impl InnerMockDriver {
const fn new() -> Self {
Self {
now: Instant::from_ticks(0),
alarm: None,
alarm: AlarmState::new(),
}
}
}
@ -189,8 +167,7 @@ mod tests {
setup();
let driver = MockDriver::get();
let alarm = unsafe { AlarmHandle::new(0) };
assert_eq!(false, driver.set_alarm(alarm, driver.now()));
assert_eq!(false, driver.set_alarm(driver.now()));
}
#[test]
@ -199,23 +176,11 @@ mod tests {
setup();
let driver = MockDriver::get();
let alarm = unsafe { driver.allocate_alarm() }.expect("No alarms available");
static mut CALLBACK_CALLED: bool = false;
let ctx = &mut () as *mut ();
driver.set_alarm_callback(alarm, |_| unsafe { CALLBACK_CALLED = true }, ctx);
driver.set_alarm(alarm, driver.now() + 1);
driver.set_alarm_callback(|_| unsafe { CALLBACK_CALLED = true }, core::ptr::null_mut());
driver.set_alarm(driver.now() + 1);
assert_eq!(false, unsafe { CALLBACK_CALLED });
driver.advance(Duration::from_secs(1));
assert_eq!(true, unsafe { CALLBACK_CALLED });
}
#[test]
#[serial]
fn test_allocate_alarm() {
setup();
let driver = MockDriver::get();
assert!(unsafe { driver.allocate_alarm() }.is_some());
assert!(unsafe { driver.allocate_alarm() }.is_none());
}
}

119
embassy-time/src/driver_std.rs
View File

@ -1,53 +1,38 @@
use core::sync::atomic::{AtomicU8, Ordering};
use std::cell::{RefCell, UnsafeCell};
use std::mem::MaybeUninit;
use std::sync::{Condvar, Mutex, Once};
use std::time::{Duration as StdDuration, Instant as StdInstant};
use std::{mem, ptr, thread};
use std::{ptr, thread};
use critical_section::Mutex as CsMutex;
use embassy_time_driver::{AlarmHandle, Driver};
const ALARM_COUNT: usize = 4;
use embassy_time_driver::Driver;
use embassy_time_queue_driver::GlobalTimerQueue;
struct AlarmState {
timestamp: u64,
// This is really a Option<(fn(*mut ()), *mut ())>
// but fn pointers aren't allowed in const yet
callback: *const (),
ctx: *mut (),
}
unsafe impl Send for AlarmState {}
impl AlarmState {
const fn new() -> Self {
Self {
timestamp: u64::MAX,
callback: ptr::null(),
ctx: ptr::null_mut(),
}
Self { timestamp: u64::MAX }
}
}
struct TimeDriver {
alarm_count: AtomicU8,
once: Once,
// The STD Driver implementation requires the alarms' mutex to be reentrant, which the STD Mutex isn't
// The STD Driver implementation requires the alarm's mutex to be reentrant, which the STD Mutex isn't
// Fortunately, mutexes based on the `critical-section` crate are reentrant, because the critical sections
// themselves are reentrant
alarms: UninitCell<CsMutex<RefCell<[AlarmState; ALARM_COUNT]>>>,
alarm: UninitCell<CsMutex<RefCell<AlarmState>>>,
zero_instant: UninitCell<StdInstant>,
signaler: UninitCell<Signaler>,
}
embassy_time_driver::time_driver_impl!(static DRIVER: TimeDriver = TimeDriver {
alarm_count: AtomicU8::new(0),
once: Once::new(),
alarms: UninitCell::uninit(),
alarm: UninitCell::uninit(),
zero_instant: UninitCell::uninit(),
signaler: UninitCell::uninit(),
});
@ -55,8 +40,8 @@ embassy_time_driver::time_driver_impl!(static DRIVER: TimeDriver = TimeDriver {
impl TimeDriver {
fn init(&self) {
self.once.call_once(|| unsafe {
self.alarms
.write(CsMutex::new(RefCell::new([const { AlarmState::new() }; ALARM_COUNT])));
self.alarm
.write(CsMutex::new(RefCell::new(const { AlarmState::new() })));
self.zero_instant.write(StdInstant::now());
self.signaler.write(Signaler::new());
@ -70,36 +55,13 @@ impl TimeDriver {
let now = DRIVER.now();
let next_alarm = critical_section::with(|cs| {
let alarms = unsafe { DRIVER.alarms.as_ref() }.borrow(cs);
loop {
let pending = alarms
.borrow_mut()
.iter_mut()
.find(|alarm| alarm.timestamp <= now)
.map(|alarm| {
alarm.timestamp = u64::MAX;
let mut alarm = unsafe { DRIVER.alarm.as_ref() }.borrow_ref_mut(cs);
if alarm.timestamp <= now {
alarm.timestamp = u64::MAX;
(alarm.callback, alarm.ctx)
});
if let Some((callback, ctx)) = pending {
// safety:
// - we can ignore the possiblity of `f` being unset (null) because of the safety contract of `allocate_alarm`.
// - other than that we only store valid function pointers into alarm.callback
let f: fn(*mut ()) = unsafe { mem::transmute(callback) };
f(ctx);
} else {
// No alarm due
break;
}
TIMER_QUEUE_DRIVER.dispatch();
}
alarms
.borrow()
.iter()
.map(|alarm| alarm.timestamp)
.min()
.unwrap_or(u64::MAX)
alarm.timestamp
});
// Ensure we don't overflow
@ -110,6 +72,17 @@ impl TimeDriver {
unsafe { DRIVER.signaler.as_ref() }.wait_until(until);
}
}
fn set_alarm(&self, timestamp: u64) -> bool {
self.init();
critical_section::with(|cs| {
let mut alarm = unsafe { self.alarm.as_ref() }.borrow_ref_mut(cs);
alarm.timestamp = timestamp;
unsafe { self.signaler.as_ref() }.signal();
});
true
}
}
impl Driver for TimeDriver {
@ -119,43 +92,6 @@ impl Driver for TimeDriver {
let zero = unsafe { self.zero_instant.read() };
StdInstant::now().duration_since(zero).as_micros() as u64
}
unsafe fn allocate_alarm(&self) -> Option<AlarmHandle> {
let id = self.alarm_count.fetch_update(Ordering::AcqRel, Ordering::Acquire, |x| {
if x < ALARM_COUNT as u8 {
Some(x + 1)
} else {
None
}
});
match id {
Ok(id) => Some(AlarmHandle::new(id)),
Err(_) => None,
}
}
fn set_alarm_callback(&self, alarm: AlarmHandle, callback: fn(*mut ()), ctx: *mut ()) {
self.init();
critical_section::with(|cs| {
let mut alarms = unsafe { self.alarms.as_ref() }.borrow_ref_mut(cs);
let alarm = &mut alarms[alarm.id() as usize];
alarm.callback = callback as *const ();
alarm.ctx = ctx;
});
}
fn set_alarm(&self, alarm: AlarmHandle, timestamp: u64) -> bool {
self.init();
critical_section::with(|cs| {
let mut alarms = unsafe { self.alarms.as_ref() }.borrow_ref_mut(cs);
let alarm = &mut alarms[alarm.id() as usize];
alarm.timestamp = timestamp;
unsafe { self.signaler.as_ref() }.signal();
});
true
}
}
struct Signaler {
@ -228,3 +164,8 @@ impl<T: Copy> UninitCell<T> {
ptr::read(self.as_mut_ptr())
}
}
embassy_time_queue_driver::timer_queue_impl!(
static TIMER_QUEUE_DRIVER: GlobalTimerQueue
= GlobalTimerQueue::new(|next_expiration| DRIVER.set_alarm(next_expiration))
);

80
embassy-time/src/driver_wasm.rs
View File

@ -1,28 +1,22 @@
use core::sync::atomic::{AtomicU8, Ordering};
use std::cell::UnsafeCell;
use std::mem::MaybeUninit;
use std::ptr;
use std::sync::{Mutex, Once};
use embassy_time_driver::{AlarmHandle, Driver};
use embassy_time_driver::Driver;
use embassy_time_queue_driver::GlobalTimerQueue;
use wasm_bindgen::prelude::*;
use wasm_timer::Instant as StdInstant;
const ALARM_COUNT: usize = 4;
struct AlarmState {
token: Option<f64>,
closure: Option<Closure<dyn FnMut() + 'static>>,
}
unsafe impl Send for AlarmState {}
impl AlarmState {
const fn new() -> Self {
Self {
token: None,
closure: None,
}
Self { token: None }
}
}
@ -33,66 +27,32 @@ extern "C" {
}
struct TimeDriver {
alarm_count: AtomicU8,
once: Once,
alarms: UninitCell<Mutex<[AlarmState; ALARM_COUNT]>>,
alarm: UninitCell<Mutex<AlarmState>>,
zero_instant: UninitCell<StdInstant>,
closure: UninitCell<Closure<dyn FnMut()>>,
}
embassy_time_driver::time_driver_impl!(static DRIVER: TimeDriver = TimeDriver {
alarm_count: AtomicU8::new(0),
once: Once::new(),
alarms: UninitCell::uninit(),
alarm: UninitCell::uninit(),
zero_instant: UninitCell::uninit(),
closure: UninitCell::uninit()
});
impl TimeDriver {
fn init(&self) {
self.once.call_once(|| unsafe {
self.alarms
.write(Mutex::new([const { AlarmState::new() }; ALARM_COUNT]));
self.alarm.write(Mutex::new(const { AlarmState::new() }));
self.zero_instant.write(StdInstant::now());
self.closure
.write(Closure::new(Box::new(|| TIMER_QUEUE_DRIVER.dispatch())));
});
}
}
impl Driver for TimeDriver {
fn now(&self) -> u64 {
fn set_alarm(&self, timestamp: u64) -> bool {
self.init();
let zero = unsafe { self.zero_instant.read() };
StdInstant::now().duration_since(zero).as_micros() as u64
}
unsafe fn allocate_alarm(&self) -> Option<AlarmHandle> {
let id = self.alarm_count.fetch_update(Ordering::AcqRel, Ordering::Acquire, |x| {
if x < ALARM_COUNT as u8 {
Some(x + 1)
} else {
None
}
});
match id {
Ok(id) => Some(AlarmHandle::new(id)),
Err(_) => None,
}
}
fn set_alarm_callback(&self, alarm: AlarmHandle, callback: fn(*mut ()), ctx: *mut ()) {
self.init();
let mut alarms = unsafe { self.alarms.as_ref() }.lock().unwrap();
let alarm = &mut alarms[alarm.id() as usize];
alarm.closure.replace(Closure::new(move || {
callback(ctx);
}));
}
fn set_alarm(&self, alarm: AlarmHandle, timestamp: u64) -> bool {
self.init();
let mut alarms = unsafe { self.alarms.as_ref() }.lock().unwrap();
let alarm = &mut alarms[alarm.id() as usize];
let mut alarm = unsafe { self.alarm.as_ref() }.lock().unwrap();
if let Some(token) = alarm.token {
clearTimeout(token);
}
@ -102,13 +62,22 @@ impl Driver for TimeDriver {
false
} else {
let timeout = (timestamp - now) as u32;
alarm.token = Some(setTimeout(alarm.closure.as_ref().unwrap(), timeout / 1000));
alarm.token = Some(setTimeout(unsafe { self.closure.as_ref() }, timeout / 1000));
true
}
}
}
impl Driver for TimeDriver {
fn now(&self) -> u64 {
self.init();
let zero = unsafe { self.zero_instant.read() };
StdInstant::now().duration_since(zero).as_micros() as u64
}
}
pub(crate) struct UninitCell<T>(MaybeUninit<UnsafeCell<T>>);
unsafe impl<T> Send for UninitCell<T> {}
unsafe impl<T> Sync for UninitCell<T> {}
@ -139,3 +108,8 @@ impl<T: Copy> UninitCell<T> {
ptr::read(self.as_mut_ptr())
}
}
embassy_time_queue_driver::timer_queue_impl!(
static TIMER_QUEUE_DRIVER: GlobalTimerQueue
= GlobalTimerQueue::new(|next_expiration| DRIVER.set_alarm(next_expiration))
);

2
embassy-time/src/lib.rs
View File

@ -25,8 +25,6 @@ pub use driver_mock::MockDriver;
mod driver_std;
#[cfg(feature = "wasm")]
mod driver_wasm;
#[cfg(feature = "generic-queue")]
mod queue_generic;
pub use delay::{block_for, Delay};
pub use duration::Duration;

346
embassy-time/src/queue_generic.rs
View File

@ -1,346 +0,0 @@
use core::cell::RefCell;
use core::cmp::{min, Ordering};
use core::task::Waker;
use critical_section::Mutex;
use embassy_time_driver::{allocate_alarm, set_alarm, set_alarm_callback, AlarmHandle};
use embassy_time_queue_driver::TimerQueue;
use heapless::Vec;
use crate::Instant;
#[cfg(feature = "generic-queue-8")]
const QUEUE_SIZE: usize = 8;
#[cfg(feature = "generic-queue-16")]
const QUEUE_SIZE: usize = 16;
#[cfg(feature = "generic-queue-32")]
const QUEUE_SIZE: usize = 32;
#[cfg(feature = "generic-queue-64")]
const QUEUE_SIZE: usize = 64;
#[cfg(feature = "generic-queue-128")]
const QUEUE_SIZE: usize = 128;
#[cfg(not(any(
feature = "generic-queue-8",
feature = "generic-queue-16",
feature = "generic-queue-32",
feature = "generic-queue-64",
feature = "generic-queue-128"
)))]
const QUEUE_SIZE: usize = 64;
#[derive(Debug)]
struct Timer {
at: Instant,
waker: Waker,
}
impl PartialEq for Timer {
fn eq(&self, other: &Self) -> bool {
self.at == other.at
}
}
impl Eq for Timer {}
impl PartialOrd for Timer {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
self.at.partial_cmp(&other.at)
}
}
impl Ord for Timer {
fn cmp(&self, other: &Self) -> Ordering {
self.at.cmp(&other.at)
}
}
struct InnerQueue {
queue: Vec<Timer, QUEUE_SIZE>,
alarm: AlarmHandle,
}
impl InnerQueue {
fn schedule_wake(&mut self, at: Instant, waker: &Waker) {
self.queue
.iter_mut()
.find(|timer| timer.waker.will_wake(waker))
.map(|timer| {
timer.at = min(timer.at, at);
})
.unwrap_or_else(|| {
let mut timer = Timer {
waker: waker.clone(),
at,
};
loop {
match self.queue.push(timer) {
Ok(()) => break,
Err(e) => timer = e,
}
self.queue.pop().unwrap().waker.wake();
}
});
// Don't wait for the alarm callback to trigger and directly
// dispatch all timers that are already due
//
// Then update the alarm if necessary
self.dispatch();
}
fn dispatch(&mut self) {
loop {
let now = Instant::now();
let mut next_alarm = Instant::MAX;
let mut i = 0;
while i < self.queue.len() {
let timer = &self.queue[i];
if timer.at <= now {
let timer = self.queue.swap_remove(i);
timer.waker.wake();
} else {
next_alarm = min(next_alarm, timer.at);
i += 1;
}
}
if self.update_alarm(next_alarm) {
break;
}
}
}
fn update_alarm(&mut self, next_alarm: Instant) -> bool {
if next_alarm == Instant::MAX {
true
} else {
set_alarm(self.alarm, next_alarm.as_ticks())
}
}
fn handle_alarm(&mut self) {
self.dispatch();
}
}
struct Queue {
inner: Mutex<RefCell<Option<InnerQueue>>>,
}
impl Queue {
const fn new() -> Self {
Self {
inner: Mutex::new(RefCell::new(None)),
}
}
fn schedule_wake(&'static self, at: Instant, waker: &Waker) {
critical_section::with(|cs| {
let mut inner = self.inner.borrow_ref_mut(cs);
inner
.get_or_insert_with(|| {
let handle = unsafe { allocate_alarm() }.unwrap();
set_alarm_callback(handle, Self::handle_alarm_callback, self as *const _ as _);
InnerQueue {
queue: Vec::new(),
alarm: handle,
}
})
.schedule_wake(at, waker)
});
}
fn handle_alarm(&self) {
critical_section::with(|cs| self.inner.borrow_ref_mut(cs).as_mut().unwrap().handle_alarm())
}
fn handle_alarm_callback(ctx: *mut ()) {
unsafe { (ctx as *const Self).as_ref().unwrap() }.handle_alarm();
}
}
impl TimerQueue for Queue {
fn schedule_wake(&'static self, at: u64, waker: &Waker) {
Queue::schedule_wake(self, Instant::from_ticks(at), waker);
}
}
embassy_time_queue_driver::timer_queue_impl!(static QUEUE: Queue = Queue::new());
#[cfg(test)]
#[cfg(feature = "mock-driver")]
mod tests {
use core::sync::atomic::{AtomicBool, Ordering};
use core::task::Waker;
use std::sync::Arc;
use std::task::Wake;
use serial_test::serial;
use crate::driver_mock::MockDriver;
use crate::queue_generic::QUEUE;
use crate::{Duration, Instant};
struct TestWaker {
pub awoken: AtomicBool,
}
impl Wake for TestWaker {
fn wake(self: Arc<Self>) {
self.awoken.store(true, Ordering::Relaxed);
}
fn wake_by_ref(self: &Arc<Self>) {
self.awoken.store(true, Ordering::Relaxed);
}
}
fn test_waker() -> (Arc<TestWaker>, Waker) {
let arc = Arc::new(TestWaker {
awoken: AtomicBool::new(false),
});
let waker = Waker::from(arc.clone());
(arc, waker)
}
fn setup() {
MockDriver::get().reset();
critical_section::with(|cs| *QUEUE.inner.borrow_ref_mut(cs) = None);
}
fn queue_len() -> usize {
critical_section::with(|cs| {
QUEUE
.inner
.borrow_ref(cs)
.as_ref()
.map(|inner| inner.queue.iter().count())
.unwrap_or(0)
})
}
#[test]
#[serial]
fn test_schedule() {
setup();
assert_eq!(queue_len(), 0);
let (flag, waker) = test_waker();
QUEUE.schedule_wake(Instant::from_secs(1), &waker);
assert!(!flag.awoken.load(Ordering::Relaxed));
assert_eq!(queue_len(), 1);
}
#[test]
#[serial]
fn test_schedule_same() {
setup();
let (_flag, waker) = test_waker();
QUEUE.schedule_wake(Instant::from_secs(1), &waker);
assert_eq!(queue_len(), 1);
QUEUE.schedule_wake(Instant::from_secs(1), &waker);
assert_eq!(queue_len(), 1);
QUEUE.schedule_wake(Instant::from_secs(100), &waker);
assert_eq!(queue_len(), 1);
let (_flag2, waker2) = test_waker();
QUEUE.schedule_wake(Instant::from_secs(100), &waker2);
assert_eq!(queue_len(), 2);
}
#[test]
#[serial]
fn test_trigger() {
setup();
let (flag, waker) = test_waker();
QUEUE.schedule_wake(Instant::from_secs(100), &waker);
assert!(!flag.awoken.load(Ordering::Relaxed));
MockDriver::get().advance(Duration::from_secs(99));
assert!(!flag.awoken.load(Ordering::Relaxed));
assert_eq!(queue_len(), 1);
MockDriver::get().advance(Duration::from_secs(1));
assert!(flag.awoken.load(Ordering::Relaxed));
assert_eq!(queue_len(), 0);
}
#[test]
#[serial]
fn test_immediate_trigger() {
setup();
let (flag, waker) = test_waker();
QUEUE.schedule_wake(Instant::from_secs(100), &waker);
MockDriver::get().advance(Duration::from_secs(50));
let (flag2, waker2) = test_waker();
QUEUE.schedule_wake(Instant::from_secs(40), &waker2);
assert!(!flag.awoken.load(Ordering::Relaxed));
assert!(flag2.awoken.load(Ordering::Relaxed));
assert_eq!(queue_len(), 1);
}
#[test]
#[serial]
fn test_queue_overflow() {
setup();
for i in 1..super::QUEUE_SIZE {
let (flag, waker) = test_waker();
QUEUE.schedule_wake(Instant::from_secs(310), &waker);
assert_eq!(queue_len(), i);
assert!(!flag.awoken.load(Ordering::Relaxed));
}
let (flag, waker) = test_waker();
QUEUE.schedule_wake(Instant::from_secs(300), &waker);
assert_eq!(queue_len(), super::QUEUE_SIZE);
assert!(!flag.awoken.load(Ordering::Relaxed));
let (flag2, waker2) = test_waker();
QUEUE.schedule_wake(Instant::from_secs(305), &waker2);
assert_eq!(queue_len(), super::QUEUE_SIZE);
assert!(flag.awoken.load(Ordering::Relaxed));
let (_flag3, waker3) = test_waker();
QUEUE.schedule_wake(Instant::from_secs(320), &waker3);
assert_eq!(queue_len(), super::QUEUE_SIZE);
assert!(flag2.awoken.load(Ordering::Relaxed));
}
}

3
examples/nrf52840-rtic/Cargo.toml
View File

@ -9,7 +9,8 @@ rtic = { version = "2", features = ["thumbv7-backend"] }
embassy-futures = { version = "0.1.0", path = "../../embassy-futures" }
embassy-sync = { version = "0.6.1", path = "../../embassy-sync", features = ["defmt"] }
embassy-time = { version = "0.3.2", path = "../../embassy-time", features = [ "defmt", "defmt-timestamp-uptime", "generic-queue"] }
embassy-time = { version = "0.3.2", path = "../../embassy-time", features = [ "defmt", "defmt-timestamp-uptime"] }
embassy-time-queue-driver = { version = "0.1.0", path = "../../embassy-time-queue-driver" }
embassy-nrf = { version = "0.2.0", path = "../../embassy-nrf", features = [ "defmt", "nrf52840", "time-driver-rtc1", "gpiote", "unstable-pac", "time"] }
defmt = "0.3"