edge_executor/
queue.rs

1use core::cell::RefCell;
2
3use alloc::collections::vec_deque::VecDeque;
4
5use async_task::Runnable;
6use embassy_sync::blocking_mutex::{
7    raw::{CriticalSectionRawMutex, RawMutex},
8    Mutex,
9};
10
11extern crate alloc;
12
13/// A trait abstracting the async executor queue.
14/// The queue MUST be thread-safe (and could be ISR-safe if the executor is used in an embedded context).
15///
16/// Pushing and popping items from the queue must NOT block the current thread / the ISR interrupt.
17///
18/// Allocating memory in the push and pop operations is only allowed if the underlying operating system
19/// allows allocations from an ISR context, and if the queue is ISR-safe in the first place.
20pub trait ExecutorQueue: Sized {
21    /// Creates a new queue.
22    fn new() -> Self;
23
24    /// Pushes a runnable into the queue.
25    fn push(&self, runnable: Runnable);
26
27    /// Pop an item from the queue.
28    ///
29    /// Return `Some(runnable)` if a runnable was successfully popped, or `None` if the queue is empty.
30    fn pop(&self) -> Option<Runnable>;
31}
32
33/// An unbounded queue implementation using `VecDeque` from the `alloc` crate.
34///
35/// Note that this queue does allocate memory on push, and deallocates memory on pop,
36/// so it may not be suitable for all use cases
37/// (e.g. embedded contexts without an allocator, or contexts where allocations are not allowed in ISRs).
38pub struct UnboundQueue<M: RawMutex = CriticalSectionRawMutex>(
39    Mutex<M, RefCell<VecDeque<Runnable>>>,
40);
41
42impl<M: RawMutex> Default for UnboundQueue<M> {
43    fn default() -> Self {
44        Self::new()
45    }
46}
47
48impl<M: RawMutex> UnboundQueue<M> {
49    /// Creates a new empty queue.
50    pub const fn new() -> Self {
51        Self(Mutex::new(RefCell::new(VecDeque::new())))
52    }
53
54    /// Create a new queue with the specified capacity.
55    pub fn with_capacity(capacity: usize) -> Self {
56        Self(Mutex::new(RefCell::new(VecDeque::with_capacity(capacity))))
57    }
58}
59
60impl<M: RawMutex> ExecutorQueue for UnboundQueue<M>
61where
62    M: RawMutex,
63{
64    fn new() -> Self {
65        Self::new()
66    }
67
68    fn push(&self, runnable: Runnable) {
69        self.0.lock(|rc| {
70            let mut queue = rc.borrow_mut();
71
72            queue.push_back(runnable);
73        })
74    }
75
76    fn pop(&self) -> Option<Runnable> {
77        self.0.lock(|rc| {
78            let mut queue = rc.borrow_mut();
79
80            queue.pop_front()
81        })
82    }
83}
84
85/// A fixed-capacity queue implementation using `heapless::Deque`.
86pub struct BoundQueue<const C: usize = 64, M: RawMutex = CriticalSectionRawMutex>(
87    Mutex<M, RefCell<heapless::Deque<Runnable, C>>>,
88);
89
90impl<const C: usize, M: RawMutex> Default for BoundQueue<C, M> {
91    fn default() -> Self {
92        Self::new()
93    }
94}
95
96impl<const C: usize, M: RawMutex> BoundQueue<C, M> {
97    /// Creates a new empty queue.
98    pub const fn new() -> Self {
99        Self(Mutex::new(RefCell::new(heapless::Deque::new())))
100    }
101}
102
103impl<const C: usize, M: RawMutex> ExecutorQueue for BoundQueue<C, M> {
104    fn new() -> Self {
105        Self::new()
106    }
107
108    fn push(&self, runnable: Runnable) {
109        self.0.lock(|rc| {
110            let mut queue = rc.borrow_mut();
111
112            queue
113                .push_back(runnable)
114                .expect("BoundQueue capacity exceeded");
115        })
116    }
117
118    fn pop(&self) -> Option<Runnable> {
119        self.0.lock(|rc| {
120            let mut queue = rc.borrow_mut();
121
122            queue.pop_front()
123        })
124    }
125}
edge_executor/queue.rs

edge_executor/
queue.rs
