diff options
| -rw-r--r-- | embassy-executor/CHANGELOG.md | 1 | ||||
| -rw-r--r-- | embassy-executor/src/raw/timer_queue.rs | 96 | ||||
| -rw-r--r-- | embassy-executor/src/raw/util.rs | 5 | ||||
| -rw-r--r-- | embassy-time-queue-driver/src/lib.rs | 11 | ||||
| -rw-r--r-- | embassy-time-queue-driver/src/queue_integrated.rs | 78 |
5 files changed, 96 insertions, 95 deletions
diff --git a/embassy-executor/CHANGELOG.md b/embassy-executor/CHANGELOG.md index 1aef57a70..068156210 100644 --- a/embassy-executor/CHANGELOG.md +++ b/embassy-executor/CHANGELOG.md | |||
| @@ -10,7 +10,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 | |||
| 10 | - embassy-executor no longer provides an `embassy-time-queue-driver` implementation | 10 | - embassy-executor no longer provides an `embassy-time-queue-driver` implementation |
| 11 | - Added `TaskRef::executor` to obtain a reference to a task's executor | 11 | - Added `TaskRef::executor` to obtain a reference to a task's executor |
| 12 | - integrated-timers are no longer processed when polling the executor. | 12 | - integrated-timers are no longer processed when polling the executor. |
| 13 | - `raw::timer_queue::TimerQueue` is now public. | ||
| 14 | 13 | ||
| 15 | ## 0.6.3 - 2024-11-12 | 14 | ## 0.6.3 - 2024-11-12 |
| 16 | 15 | ||
diff --git a/embassy-executor/src/raw/timer_queue.rs b/embassy-executor/src/raw/timer_queue.rs index e0a22f4d4..46e346c1b 100644 --- a/embassy-executor/src/raw/timer_queue.rs +++ b/embassy-executor/src/raw/timer_queue.rs | |||
| @@ -1,99 +1,25 @@ | |||
| 1 | //! Timer queue operations. | 1 | //! Timer queue operations. |
| 2 | use core::cmp::min; | ||
| 3 | 2 | ||
| 4 | use super::util::SyncUnsafeCell; | 3 | use core::cell::Cell; |
| 4 | |||
| 5 | use super::TaskRef; | 5 | use super::TaskRef; |
| 6 | 6 | ||
| 7 | /// An item in the timer queue. | 7 | /// An item in the timer queue. |
| 8 | pub struct TimerQueueItem { | 8 | pub struct TimerQueueItem { |
| 9 | next: SyncUnsafeCell<Option<TaskRef>>, | 9 | /// The next item in the queue. |
| 10 | expires_at: SyncUnsafeCell<u64>, | 10 | pub next: Cell<Option<TaskRef>>, |
| 11 | } | ||
| 12 | 11 | ||
| 13 | impl TimerQueueItem { | 12 | /// The time at which this item expires. |
| 14 | pub(crate) const fn new() -> Self { | 13 | pub expires_at: Cell<u64>, |
| 15 | Self { | ||
| 16 | next: SyncUnsafeCell::new(None), | ||
| 17 | expires_at: SyncUnsafeCell::new(0), | ||
| 18 | } | ||
| 19 | } | ||
| 20 | } | 14 | } |
| 21 | 15 | ||
| 22 | /// A timer queue, with items integrated into tasks. | 16 | unsafe impl Sync for TimerQueueItem {} |
| 23 | pub struct TimerQueue { | ||
| 24 | head: SyncUnsafeCell<Option<TaskRef>>, | ||
| 25 | } | ||
| 26 | 17 | ||
| 27 | impl TimerQueue { | 18 | impl TimerQueueItem { |
| 28 | /// Creates a new timer queue. | 19 | pub(crate) const fn new() -> Self { |
| 29 | pub const fn new() -> Self { | ||
| 30 | Self { | 20 | Self { |
| 31 | head: SyncUnsafeCell::new(None), | 21 | next: Cell::new(None), |
| 32 | } | 22 | expires_at: Cell::new(0), |
| 33 | } | ||
| 34 | |||
| 35 | /// Schedules a task to run at a specific time. | ||
| 36 | /// | ||
| 37 | /// If this function returns `true`, the called should find the next expiration time and set | ||
| 38 | /// a new alarm for that time. | ||
| 39 | pub fn schedule_wake(&mut self, at: u64, p: TaskRef) -> bool { | ||
| 40 | unsafe { | ||
| 41 | let item = p.timer_queue_item(); | ||
| 42 | if item.next.get().is_none() { | ||
| 43 | // If not in the queue, add it and update. | ||
| 44 | let prev = self.head.replace(Some(p)); | ||
| 45 | item.next.set(prev); | ||
| 46 | } else if at <= item.expires_at.get() { | ||
| 47 | // If expiration is sooner than previously set, update. | ||
| 48 | } else { | ||
| 49 | // Task does not need to be updated. | ||
| 50 | return false; | ||
| 51 | } | ||
| 52 | |||
| 53 | item.expires_at.set(at); | ||
| 54 | true | ||
| 55 | } | ||
| 56 | } | ||
| 57 | |||
| 58 | /// Dequeues expired timers and returns the next alarm time. | ||
| 59 | /// | ||
| 60 | /// The provided callback will be called for each expired task. Tasks that never expire | ||
| 61 | /// will be removed, but the callback will not be called. | ||
| 62 | pub fn next_expiration(&mut self, now: u64) -> u64 { | ||
| 63 | let mut next_expiration = u64::MAX; | ||
| 64 | |||
| 65 | self.retain(|p| { | ||
| 66 | let item = p.timer_queue_item(); | ||
| 67 | let expires = unsafe { item.expires_at.get() }; | ||
| 68 | |||
| 69 | if expires <= now { | ||
| 70 | // Timer expired, process task. | ||
| 71 | super::wake_task(p); | ||
| 72 | false | ||
| 73 | } else { | ||
| 74 | // Timer didn't yet expire, or never expires. | ||
| 75 | next_expiration = min(next_expiration, expires); | ||
| 76 | expires != u64::MAX | ||
| 77 | } | ||
| 78 | }); | ||
| 79 | |||
| 80 | next_expiration | ||
| 81 | } | ||
| 82 | |||
| 83 | fn retain(&self, mut f: impl FnMut(TaskRef) -> bool) { | ||
| 84 | unsafe { | ||
| 85 | let mut prev = &self.head; | ||
| 86 | while let Some(p) = prev.get() { | ||
| 87 | let item = p.timer_queue_item(); | ||
| 88 | if f(p) { | ||
| 89 | // Skip to next | ||
| 90 | prev = &item.next; | ||
| 91 | } else { | ||
| 92 | // Remove it | ||
| 93 | prev.set(item.next.get()); | ||
| 94 | item.next.set(None); | ||
| 95 | } | ||
| 96 | } | ||
| 97 | } | 23 | } |
| 98 | } | 24 | } |
| 99 | } | 25 | } |
diff --git a/embassy-executor/src/raw/util.rs b/embassy-executor/src/raw/util.rs index e2633658a..c46085e45 100644 --- a/embassy-executor/src/raw/util.rs +++ b/embassy-executor/src/raw/util.rs | |||
| @@ -54,9 +54,4 @@ impl<T> SyncUnsafeCell<T> { | |||
| 54 | { | 54 | { |
| 55 | *self.value.get() | 55 | *self.value.get() |
| 56 | } | 56 | } |
| 57 | |||
| 58 | #[cfg(feature = "integrated-timers")] | ||
| 59 | pub unsafe fn replace(&self, value: T) -> T { | ||
| 60 | core::mem::replace(&mut *self.value.get(), value) | ||
| 61 | } | ||
| 62 | } | 57 | } |
diff --git a/embassy-time-queue-driver/src/lib.rs b/embassy-time-queue-driver/src/lib.rs index c5e989854..0c78921ed 100644 --- a/embassy-time-queue-driver/src/lib.rs +++ b/embassy-time-queue-driver/src/lib.rs | |||
| @@ -22,9 +22,9 @@ | |||
| 22 | //! ); | 22 | //! ); |
| 23 | //! ``` | 23 | //! ``` |
| 24 | //! | 24 | //! |
| 25 | //! You can also use the `queue_generic` or the `embassy_executor::raw::timer_queue` modules to | 25 | //! You can also use the `queue_generic` or the `queue_integrated` modules to implement your own |
| 26 | //! implement your own timer queue. These modules contain queue implementations which you can wrap | 26 | //! timer queue. These modules contain queue implementations which you can wrap and tailor to |
| 27 | //! and tailor to your needs. | 27 | //! your needs. |
| 28 | //! | 28 | //! |
| 29 | //! If you are providing an embassy-executor implementation besides a timer queue, you can choose to | 29 | //! If you are providing an embassy-executor implementation besides a timer queue, you can choose to |
| 30 | //! expose the `integrated-timers` feature in your implementation. This feature stores timer items | 30 | //! expose the `integrated-timers` feature in your implementation. This feature stores timer items |
| @@ -49,7 +49,10 @@ | |||
| 49 | //! embassy_time_queue_driver::timer_queue_impl!(static QUEUE: MyTimerQueue = MyTimerQueue{}); | 49 | //! embassy_time_queue_driver::timer_queue_impl!(static QUEUE: MyTimerQueue = MyTimerQueue{}); |
| 50 | //! ``` | 50 | //! ``` |
| 51 | 51 | ||
| 52 | #[cfg(not(feature = "integrated-timers"))] | ||
| 52 | pub mod queue_generic; | 53 | pub mod queue_generic; |
| 54 | #[cfg(feature = "integrated-timers")] | ||
| 55 | pub mod queue_integrated; | ||
| 53 | 56 | ||
| 54 | use core::cell::RefCell; | 57 | use core::cell::RefCell; |
| 55 | use core::task::Waker; | 58 | use core::task::Waker; |
| @@ -89,7 +92,7 @@ macro_rules! timer_queue_impl { | |||
| 89 | } | 92 | } |
| 90 | 93 | ||
| 91 | #[cfg(feature = "integrated-timers")] | 94 | #[cfg(feature = "integrated-timers")] |
| 92 | type InnerQueue = embassy_executor::raw::timer_queue::TimerQueue; | 95 | type InnerQueue = queue_integrated::TimerQueue; |
| 93 | 96 | ||
| 94 | #[cfg(not(feature = "integrated-timers"))] | 97 | #[cfg(not(feature = "integrated-timers"))] |
| 95 | type InnerQueue = queue_generic::Queue; | 98 | type InnerQueue = queue_generic::Queue; |
diff --git a/embassy-time-queue-driver/src/queue_integrated.rs b/embassy-time-queue-driver/src/queue_integrated.rs new file mode 100644 index 000000000..cb0f79356 --- /dev/null +++ b/embassy-time-queue-driver/src/queue_integrated.rs | |||
| @@ -0,0 +1,78 @@ | |||
| 1 | //! Timer queue operations. | ||
| 2 | use core::cell::Cell; | ||
| 3 | use core::cmp::min; | ||
| 4 | |||
| 5 | use embassy_executor::raw::TaskRef; | ||
| 6 | |||
| 7 | /// A timer queue, with items integrated into tasks. | ||
| 8 | pub struct TimerQueue { | ||
| 9 | head: Cell<Option<TaskRef>>, | ||
| 10 | } | ||
| 11 | |||
| 12 | impl TimerQueue { | ||
| 13 | /// Creates a new timer queue. | ||
| 14 | pub const fn new() -> Self { | ||
| 15 | Self { head: Cell::new(None) } | ||
| 16 | } | ||
| 17 | |||
| 18 | /// Schedules a task to run at a specific time. | ||
| 19 | /// | ||
| 20 | /// If this function returns `true`, the called should find the next expiration time and set | ||
| 21 | /// a new alarm for that time. | ||
| 22 | pub fn schedule_wake(&mut self, at: u64, p: TaskRef) -> bool { | ||
| 23 | let item = p.timer_queue_item(); | ||
| 24 | if item.next.get().is_none() { | ||
| 25 | // If not in the queue, add it and update. | ||
| 26 | let prev = self.head.replace(Some(p)); | ||
| 27 | item.next.set(prev); | ||
| 28 | } else if at <= item.expires_at.get() { | ||
| 29 | // If expiration is sooner than previously set, update. | ||
| 30 | } else { | ||
| 31 | // Task does not need to be updated. | ||
| 32 | return false; | ||
| 33 | } | ||
| 34 | |||
| 35 | item.expires_at.set(at); | ||
| 36 | true | ||
| 37 | } | ||
| 38 | |||
| 39 | /// Dequeues expired timers and returns the next alarm time. | ||
| 40 | /// | ||
| 41 | /// The provided callback will be called for each expired task. Tasks that never expire | ||
| 42 | /// will be removed, but the callback will not be called. | ||
| 43 | pub fn next_expiration(&mut self, now: u64) -> u64 { | ||
| 44 | let mut next_expiration = u64::MAX; | ||
| 45 | |||
| 46 | self.retain(|p| { | ||
| 47 | let item = p.timer_queue_item(); | ||
| 48 | let expires = item.expires_at.get(); | ||
| 49 | |||
| 50 | if expires <= now { | ||
| 51 | // Timer expired, process task. | ||
| 52 | embassy_executor::raw::wake_task(p); | ||
| 53 | false | ||
| 54 | } else { | ||
| 55 | // Timer didn't yet expire, or never expires. | ||
| 56 | next_expiration = min(next_expiration, expires); | ||
| 57 | expires != u64::MAX | ||
| 58 | } | ||
| 59 | }); | ||
| 60 | |||
| 61 | next_expiration | ||
| 62 | } | ||
| 63 | |||
| 64 | fn retain(&self, mut f: impl FnMut(TaskRef) -> bool) { | ||
| 65 | let mut prev = &self.head; | ||
| 66 | while let Some(p) = prev.get() { | ||
| 67 | let item = p.timer_queue_item(); | ||
| 68 | if f(p) { | ||
| 69 | // Skip to next | ||
| 70 | prev = &item.next; | ||
| 71 | } else { | ||
| 72 | // Remove it | ||
| 73 | prev.set(item.next.get()); | ||
| 74 | item.next.set(None); | ||
| 75 | } | ||
| 76 | } | ||
| 77 | } | ||
| 78 | } | ||
