mirror of
https://github.com/embassy-rs/embassy.git
synced 2025-09-27 12:20:37 +00:00
There can be only one (run queue)
This commit is contained in:
parent
67ed473973
commit
7af8f35a50
@ -76,13 +76,16 @@ js-sys = { version = "0.3", optional = true }
|
||||
# arch-avr dependencies
|
||||
avr-device = { version = "0.7.0", optional = true }
|
||||
|
||||
# Note: this is ONLY a dependency when the target has atomics, this is
|
||||
# used for `run_queue_atomics`. We need to be conditional because
|
||||
# cordyceps *requires* the use of atomics, so we pull it in when
|
||||
# `run_queue_atomics` would be enabled, and NOT when `run_queue_critical_section`
|
||||
# would be enabled.
|
||||
[target.'cfg(target_has_atomic="ptr")'.dependencies.cordyceps]
|
||||
version = "0.3.3"
|
||||
|
||||
[dependencies.cordyceps]
|
||||
# version = "0.3.3"
|
||||
# todo: update after https://github.com/hawkw/mycelium/pull/537 is merged
|
||||
git = "https://github.com/hawkw/mycelium"
|
||||
rev = "86c428eecfd37ee24dd81f14c4a9f5c8ecefcf17"
|
||||
|
||||
# Note: this is ONLY a dependency when the target does NOT have atomics
|
||||
[target.'cfg(not(target_has_atomic="ptr"))'.dependencies.mutex]
|
||||
version = "1.0"
|
||||
|
||||
[dev-dependencies]
|
||||
critical-section = { version = "1.1", features = ["std"] }
|
||||
@ -135,6 +138,5 @@ rtos-trace = ["_any_trace", "metadata-name", "dep:rtos-trace", "dep:embassy-time
|
||||
_any_trace = []
|
||||
|
||||
## Enable "Earliest Deadline First" Scheduler, using soft-realtime "deadlines" to prioritize
|
||||
## tasks based on the remaining time before their deadline. Adds some overhead. Requires
|
||||
## hardware atomic support
|
||||
## tasks based on the remaining time before their deadline. Adds some overhead.
|
||||
edf-scheduler = ["dep:embassy-time-driver"]
|
||||
|
@ -1,9 +1,6 @@
|
||||
use core::future::{poll_fn, Future};
|
||||
use core::task::Poll;
|
||||
|
||||
#[cfg(not(target_has_atomic = "ptr"))]
|
||||
compile_error!("The `edf-scheduler` feature is currently only supported on targets with atomics.");
|
||||
|
||||
/// A type for interacting with the deadline of the current task
|
||||
///
|
||||
/// Requires the `edf-scheduler` feature
|
||||
|
@ -7,8 +7,6 @@
|
||||
//! Using this module requires respecting subtle safety contracts. If you can, prefer using the safe
|
||||
//! [executor wrappers](crate::Executor) and the [`embassy_executor::task`](embassy_executor_macros::task) macro, which are fully safe.
|
||||
|
||||
#[cfg_attr(target_has_atomic = "ptr", path = "run_queue_atomics.rs")]
|
||||
#[cfg_attr(not(target_has_atomic = "ptr"), path = "run_queue_critical_section.rs")]
|
||||
mod run_queue;
|
||||
|
||||
#[cfg_attr(all(cortex_m, target_has_atomic = "32"), path = "state_atomics_arm.rs")]
|
||||
|
@ -1,9 +1,15 @@
|
||||
use core::ptr::{addr_of_mut, NonNull};
|
||||
|
||||
use cordyceps::sorted_list::Links;
|
||||
use cordyceps::Linked;
|
||||
#[cfg(feature = "edf-scheduler")]
|
||||
use cordyceps::SortedList;
|
||||
use cordyceps::{Linked, TransferStack};
|
||||
|
||||
#[cfg(target_has_atomic = "ptr")]
|
||||
type TransferStack<T> = cordyceps::TransferStack<T>;
|
||||
|
||||
#[cfg(not(target_has_atomic = "ptr"))]
|
||||
type TransferStack<T> = cordyceps::TransferStack<mutex::raw_impls::cs::CriticalSectionRawMutex, T>;
|
||||
|
||||
use super::{TaskHeader, TaskRef};
|
||||
|
||||
@ -77,7 +83,7 @@ impl RunQueue {
|
||||
pub(crate) fn dequeue_all(&self, on_task: impl Fn(TaskRef)) {
|
||||
let taken = self.stack.take_all();
|
||||
for taskref in taken {
|
||||
taskref.header().state.run_dequeue();
|
||||
run_dequeue(&taskref);
|
||||
on_task(taskref);
|
||||
}
|
||||
}
|
||||
@ -122,8 +128,24 @@ impl RunQueue {
|
||||
};
|
||||
|
||||
// We got one task, mark it as dequeued, and process the task.
|
||||
taskref.header().state.run_dequeue();
|
||||
run_dequeue(&taskref);
|
||||
on_task(taskref);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// atomic state does not require a cs...
|
||||
#[cfg(target_has_atomic = "ptr")]
|
||||
#[inline(always)]
|
||||
fn run_dequeue(taskref: &TaskRef) {
|
||||
taskref.header().state.run_dequeue();
|
||||
}
|
||||
|
||||
/// ...while non-atomic state does
|
||||
#[cfg(not(target_has_atomic = "ptr"))]
|
||||
#[inline(always)]
|
||||
fn run_dequeue(taskref: &TaskRef) {
|
||||
critical_section::with(|cs| {
|
||||
taskref.header().state.run_dequeue(cs);
|
||||
})
|
||||
}
|
@ -1,74 +0,0 @@
|
||||
use core::cell::Cell;
|
||||
|
||||
use critical_section::{CriticalSection, Mutex};
|
||||
|
||||
use super::TaskRef;
|
||||
|
||||
pub(crate) struct RunQueueItem {
|
||||
next: Mutex<Cell<Option<TaskRef>>>,
|
||||
}
|
||||
|
||||
impl RunQueueItem {
|
||||
pub const fn new() -> Self {
|
||||
Self {
|
||||
next: Mutex::new(Cell::new(None)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Atomic task queue using a very, very simple lock-free linked-list queue:
|
||||
///
|
||||
/// To enqueue a task, task.next is set to the old head, and head is atomically set to task.
|
||||
///
|
||||
/// Dequeuing is done in batches: the queue is emptied by atomically replacing head with
|
||||
/// null. Then the batch is iterated following the next pointers until null is reached.
|
||||
///
|
||||
/// Note that batches will be iterated in the reverse order as they were enqueued. This is OK
|
||||
/// for our purposes: it can't create fairness problems since the next batch won't run until the
|
||||
/// current batch is completely processed, so even if a task enqueues itself instantly (for example
|
||||
/// by waking its own waker) can't prevent other tasks from running.
|
||||
pub(crate) struct RunQueue {
|
||||
head: Mutex<Cell<Option<TaskRef>>>,
|
||||
}
|
||||
|
||||
impl RunQueue {
|
||||
pub const fn new() -> Self {
|
||||
Self {
|
||||
head: Mutex::new(Cell::new(None)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Enqueues an item. Returns true if the queue was empty.
|
||||
///
|
||||
/// # Safety
|
||||
///
|
||||
/// `item` must NOT be already enqueued in any queue.
|
||||
#[inline(always)]
|
||||
pub(crate) unsafe fn enqueue(&self, task: TaskRef, cs: CriticalSection<'_>) -> bool {
|
||||
let prev = self.head.borrow(cs).replace(Some(task));
|
||||
task.header().run_queue_item.next.borrow(cs).set(prev);
|
||||
|
||||
prev.is_none()
|
||||
}
|
||||
|
||||
/// Empty the queue, then call `on_task` for each task that was in the queue.
|
||||
/// NOTE: It is OK for `on_task` to enqueue more tasks. In this case they're left in the queue
|
||||
/// and will be processed by the *next* call to `dequeue_all`, *not* the current one.
|
||||
pub(crate) fn dequeue_all(&self, on_task: impl Fn(TaskRef)) {
|
||||
// Atomically empty the queue.
|
||||
let mut next = critical_section::with(|cs| self.head.borrow(cs).take());
|
||||
|
||||
// Iterate the linked list of tasks that were previously in the queue.
|
||||
while let Some(task) = next {
|
||||
// If the task re-enqueues itself, the `next` pointer will get overwritten.
|
||||
// Therefore, first read the next pointer, and only then process the task.
|
||||
|
||||
critical_section::with(|cs| {
|
||||
next = task.header().run_queue_item.next.borrow(cs).get();
|
||||
task.header().state.run_dequeue(cs);
|
||||
});
|
||||
|
||||
on_task(task);
|
||||
}
|
||||
}
|
||||
}
|
Loading…
x
Reference in New Issue
Block a user