aboutsummaryrefslogtreecommitdiff
path: root/embassy-time/src/queue_generic.rs
diff options
context:
space:
mode:
authorDániel Buga <[email protected]>2024-11-26 23:54:21 +0100
committerDániel Buga <[email protected]>2024-12-10 21:31:42 +0100
commit5a5495aac43d75610735f2ca80fb6c8e8f31ed71 (patch)
tree7a4336917894730692589359e9d1a285ec5a0a05 /embassy-time/src/queue_generic.rs
parent406d377b7564d16e12b7fae4f42c0c709bf4f243 (diff)
Refactor integrated-timers
Diffstat (limited to 'embassy-time/src/queue_generic.rs')
-rw-r--r--embassy-time/src/queue_generic.rs346
1 files changed, 0 insertions, 346 deletions
diff --git a/embassy-time/src/queue_generic.rs b/embassy-time/src/queue_generic.rs
deleted file mode 100644
index 0068edae8..000000000
--- a/embassy-time/src/queue_generic.rs
+++ /dev/null
@@ -1,346 +0,0 @@
1use core::cell::RefCell;
2use core::cmp::{min, Ordering};
3use core::task::Waker;
4
5use critical_section::Mutex;
6use embassy_time_driver::{allocate_alarm, set_alarm, set_alarm_callback, AlarmHandle};
7use embassy_time_queue_driver::TimerQueue;
8use heapless::Vec;
9
10use crate::Instant;
11
12#[cfg(feature = "generic-queue-8")]
13const QUEUE_SIZE: usize = 8;
14#[cfg(feature = "generic-queue-16")]
15const QUEUE_SIZE: usize = 16;
16#[cfg(feature = "generic-queue-32")]
17const QUEUE_SIZE: usize = 32;
18#[cfg(feature = "generic-queue-64")]
19const QUEUE_SIZE: usize = 64;
20#[cfg(feature = "generic-queue-128")]
21const QUEUE_SIZE: usize = 128;
22#[cfg(not(any(
23 feature = "generic-queue-8",
24 feature = "generic-queue-16",
25 feature = "generic-queue-32",
26 feature = "generic-queue-64",
27 feature = "generic-queue-128"
28)))]
29const QUEUE_SIZE: usize = 64;
30
31#[derive(Debug)]
32struct Timer {
33 at: Instant,
34 waker: Waker,
35}
36
37impl PartialEq for Timer {
38 fn eq(&self, other: &Self) -> bool {
39 self.at == other.at
40 }
41}
42
43impl Eq for Timer {}
44
45impl PartialOrd for Timer {
46 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
47 self.at.partial_cmp(&other.at)
48 }
49}
50
51impl Ord for Timer {
52 fn cmp(&self, other: &Self) -> Ordering {
53 self.at.cmp(&other.at)
54 }
55}
56
57struct InnerQueue {
58 queue: Vec<Timer, QUEUE_SIZE>,
59 alarm: AlarmHandle,
60}
61
62impl InnerQueue {
63 fn schedule_wake(&mut self, at: Instant, waker: &Waker) {
64 self.queue
65 .iter_mut()
66 .find(|timer| timer.waker.will_wake(waker))
67 .map(|timer| {
68 timer.at = min(timer.at, at);
69 })
70 .unwrap_or_else(|| {
71 let mut timer = Timer {
72 waker: waker.clone(),
73 at,
74 };
75
76 loop {
77 match self.queue.push(timer) {
78 Ok(()) => break,
79 Err(e) => timer = e,
80 }
81
82 self.queue.pop().unwrap().waker.wake();
83 }
84 });
85
86 // Don't wait for the alarm callback to trigger and directly
87 // dispatch all timers that are already due
88 //
89 // Then update the alarm if necessary
90 self.dispatch();
91 }
92
93 fn dispatch(&mut self) {
94 loop {
95 let now = Instant::now();
96
97 let mut next_alarm = Instant::MAX;
98
99 let mut i = 0;
100 while i < self.queue.len() {
101 let timer = &self.queue[i];
102 if timer.at <= now {
103 let timer = self.queue.swap_remove(i);
104 timer.waker.wake();
105 } else {
106 next_alarm = min(next_alarm, timer.at);
107 i += 1;
108 }
109 }
110
111 if self.update_alarm(next_alarm) {
112 break;
113 }
114 }
115 }
116
117 fn update_alarm(&mut self, next_alarm: Instant) -> bool {
118 if next_alarm == Instant::MAX {
119 true
120 } else {
121 set_alarm(self.alarm, next_alarm.as_ticks())
122 }
123 }
124
125 fn handle_alarm(&mut self) {
126 self.dispatch();
127 }
128}
129
130struct Queue {
131 inner: Mutex<RefCell<Option<InnerQueue>>>,
132}
133
134impl Queue {
135 const fn new() -> Self {
136 Self {
137 inner: Mutex::new(RefCell::new(None)),
138 }
139 }
140
141 fn schedule_wake(&'static self, at: Instant, waker: &Waker) {
142 critical_section::with(|cs| {
143 let mut inner = self.inner.borrow_ref_mut(cs);
144
145 inner
146 .get_or_insert_with(|| {
147 let handle = unsafe { allocate_alarm() }.unwrap();
148 set_alarm_callback(handle, Self::handle_alarm_callback, self as *const _ as _);
149 InnerQueue {
150 queue: Vec::new(),
151 alarm: handle,
152 }
153 })
154 .schedule_wake(at, waker)
155 });
156 }
157
158 fn handle_alarm(&self) {
159 critical_section::with(|cs| self.inner.borrow_ref_mut(cs).as_mut().unwrap().handle_alarm())
160 }
161
162 fn handle_alarm_callback(ctx: *mut ()) {
163 unsafe { (ctx as *const Self).as_ref().unwrap() }.handle_alarm();
164 }
165}
166
167impl TimerQueue for Queue {
168 fn schedule_wake(&'static self, at: u64, waker: &Waker) {
169 Queue::schedule_wake(self, Instant::from_ticks(at), waker);
170 }
171}
172
173embassy_time_queue_driver::timer_queue_impl!(static QUEUE: Queue = Queue::new());
174
175#[cfg(test)]
176#[cfg(feature = "mock-driver")]
177mod tests {
178 use core::sync::atomic::{AtomicBool, Ordering};
179 use core::task::Waker;
180 use std::sync::Arc;
181 use std::task::Wake;
182
183 use serial_test::serial;
184
185 use crate::driver_mock::MockDriver;
186 use crate::queue_generic::QUEUE;
187 use crate::{Duration, Instant};
188
189 struct TestWaker {
190 pub awoken: AtomicBool,
191 }
192
193 impl Wake for TestWaker {
194 fn wake(self: Arc<Self>) {
195 self.awoken.store(true, Ordering::Relaxed);
196 }
197
198 fn wake_by_ref(self: &Arc<Self>) {
199 self.awoken.store(true, Ordering::Relaxed);
200 }
201 }
202
203 fn test_waker() -> (Arc<TestWaker>, Waker) {
204 let arc = Arc::new(TestWaker {
205 awoken: AtomicBool::new(false),
206 });
207 let waker = Waker::from(arc.clone());
208
209 (arc, waker)
210 }
211
212 fn setup() {
213 MockDriver::get().reset();
214 critical_section::with(|cs| *QUEUE.inner.borrow_ref_mut(cs) = None);
215 }
216
217 fn queue_len() -> usize {
218 critical_section::with(|cs| {
219 QUEUE
220 .inner
221 .borrow_ref(cs)
222 .as_ref()
223 .map(|inner| inner.queue.iter().count())
224 .unwrap_or(0)
225 })
226 }
227
228 #[test]
229 #[serial]
230 fn test_schedule() {
231 setup();
232
233 assert_eq!(queue_len(), 0);
234
235 let (flag, waker) = test_waker();
236
237 QUEUE.schedule_wake(Instant::from_secs(1), &waker);
238
239 assert!(!flag.awoken.load(Ordering::Relaxed));
240 assert_eq!(queue_len(), 1);
241 }
242
243 #[test]
244 #[serial]
245 fn test_schedule_same() {
246 setup();
247
248 let (_flag, waker) = test_waker();
249
250 QUEUE.schedule_wake(Instant::from_secs(1), &waker);
251
252 assert_eq!(queue_len(), 1);
253
254 QUEUE.schedule_wake(Instant::from_secs(1), &waker);
255
256 assert_eq!(queue_len(), 1);
257
258 QUEUE.schedule_wake(Instant::from_secs(100), &waker);
259
260 assert_eq!(queue_len(), 1);
261
262 let (_flag2, waker2) = test_waker();
263
264 QUEUE.schedule_wake(Instant::from_secs(100), &waker2);
265
266 assert_eq!(queue_len(), 2);
267 }
268
269 #[test]
270 #[serial]
271 fn test_trigger() {
272 setup();
273
274 let (flag, waker) = test_waker();
275
276 QUEUE.schedule_wake(Instant::from_secs(100), &waker);
277
278 assert!(!flag.awoken.load(Ordering::Relaxed));
279
280 MockDriver::get().advance(Duration::from_secs(99));
281
282 assert!(!flag.awoken.load(Ordering::Relaxed));
283
284 assert_eq!(queue_len(), 1);
285
286 MockDriver::get().advance(Duration::from_secs(1));
287
288 assert!(flag.awoken.load(Ordering::Relaxed));
289
290 assert_eq!(queue_len(), 0);
291 }
292
293 #[test]
294 #[serial]
295 fn test_immediate_trigger() {
296 setup();
297
298 let (flag, waker) = test_waker();
299
300 QUEUE.schedule_wake(Instant::from_secs(100), &waker);
301
302 MockDriver::get().advance(Duration::from_secs(50));
303
304 let (flag2, waker2) = test_waker();
305
306 QUEUE.schedule_wake(Instant::from_secs(40), &waker2);
307
308 assert!(!flag.awoken.load(Ordering::Relaxed));
309 assert!(flag2.awoken.load(Ordering::Relaxed));
310 assert_eq!(queue_len(), 1);
311 }
312
313 #[test]
314 #[serial]
315 fn test_queue_overflow() {
316 setup();
317
318 for i in 1..super::QUEUE_SIZE {
319 let (flag, waker) = test_waker();
320
321 QUEUE.schedule_wake(Instant::from_secs(310), &waker);
322
323 assert_eq!(queue_len(), i);
324 assert!(!flag.awoken.load(Ordering::Relaxed));
325 }
326
327 let (flag, waker) = test_waker();
328
329 QUEUE.schedule_wake(Instant::from_secs(300), &waker);
330
331 assert_eq!(queue_len(), super::QUEUE_SIZE);
332 assert!(!flag.awoken.load(Ordering::Relaxed));
333
334 let (flag2, waker2) = test_waker();
335
336 QUEUE.schedule_wake(Instant::from_secs(305), &waker2);
337
338 assert_eq!(queue_len(), super::QUEUE_SIZE);
339 assert!(flag.awoken.load(Ordering::Relaxed));
340
341 let (_flag3, waker3) = test_waker();
342 QUEUE.schedule_wake(Instant::from_secs(320), &waker3);
343 assert_eq!(queue_len(), super::QUEUE_SIZE);
344 assert!(flag2.awoken.load(Ordering::Relaxed));
345 }
346}
generated by cgit (git 2.34.1) at 2026-04-05 21:38:49 +0000