Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Allow polling exited tasks #3622

Merged
merged 7 commits into from
Dec 17, 2024
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 36 additions & 33 deletions embassy-executor/src/raw/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -50,33 +50,32 @@ use super::SpawnToken;
/// A task's complete life cycle is as follows:
///
/// ```text
/// ┌────────────┐ ┌────────────────────────┐
/// ┌─►│Not spawned │◄─6┤Not spawned|Run enqueued│
/// │ ├7─►│ │
/// └─────┬──────┘ └──────▲─────────────────┘
/// 1 │
/// │ ┌────────────┘
/// 5
/// ┌─────▼────┴─────────┐
/// │Spawned|Run enqueued│
/// │
/// └─────┬▲─────────────┘
/// 2│
/// │3
/// ┌─────▼┴─────┐
/// └─4┤ Spawned │
/// │ │
/// └────────────┘
/// ┌────────────┐ ┌────────────────────────┐
/// │Not spawned │◄─5┤Not spawned|Run enqueued│
/// │ ├6─►│ │
/// └─────┬──────┘ └──────▲─────────────────┘
/// 1 │
/// │ ┌────────────┘
/// │ 4
/// ┌─────▼────┴─────────┐
/// │Spawned|Run enqueued│
/// │ │
/// └─────┬▲─────────────┘
/// 2│
/// │3
/// ┌─────▼┴─────┐
/// Spawned │
/// │ │
/// └────────────┘
/// ```
///
/// Transitions:
/// - 1: Task is spawned - `AvailableTask::claim -> Executor::spawn`
/// - 2: During poll - `RunQueue::dequeue_all -> State::run_dequeue`
/// - 3: Task wakes itself, waker wakes task - `Waker::wake -> wake_task -> State::run_enqueue`
/// - 4: Task exits - `TaskStorage::poll -> Poll::Ready`
/// - 5: A run-queued task exits - `TaskStorage::poll -> Poll::Ready`
/// - 6: Task is dequeued and then ignored via `State::run_dequeue`
/// - 7: A task is waken when it is not spawned - `wake_task -> State::run_enqueue`
/// - 3: Task wakes itself, waker wakes task, or task exits - `Waker::wake -> wake_task -> State::run_enqueue`
/// - 4: A run-queued task exits - `TaskStorage::poll -> Poll::Ready`
/// - 5: Task is dequeued. The task's future is not polled, because exiting the task replaces its `poll_fn`.
/// - 6: A task is waken when it is not spawned - `wake_task -> State::run_enqueue`
pub(crate) struct TaskHeader {
pub(crate) state: State,
pub(crate) run_queue_item: RunQueueItem,
Expand Down Expand Up @@ -202,16 +201,29 @@ impl<F: Future + 'static> TaskStorage<F> {
}
}

unsafe fn poll_to_despawn(p: TaskRef) {
// The task's future has already been dropped, we just mark it as `!SPAWNED`.
let this = &*p.as_ptr().cast::<TaskStorage<F>>();
this.raw.state.despawn();
}

unsafe fn poll(p: TaskRef) {
let this = &*(p.as_ptr() as *const TaskStorage<F>);
let this = &*p.as_ptr().cast::<TaskStorage<F>>();

let future = Pin::new_unchecked(this.future.as_mut());
let waker = waker::from_task(p);
let mut cx = Context::from_waker(&waker);
match future.poll(&mut cx) {
Poll::Ready(_) => {
waker.wake_by_ref();
bugadani marked this conversation as resolved.
Show resolved Hide resolved

// As the future has finished and this function will not be called
// again, we can safely drop the future here.
this.future.drop_in_place();
this.raw.state.despawn();

// We replace the poll_fn with a despawn function, so that the task is cleaned up
// when the executor polls it next.
this.raw.poll_fn.set(Some(Self::poll_to_despawn));
}
Poll::Pending => {}
}
Expand Down Expand Up @@ -411,15 +423,6 @@ impl SyncExecutor {
self.run_queue.dequeue_all(|p| {
let task = p.header();

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);

Expand Down
1 change: 1 addition & 0 deletions embassy-executor/src/raw/run_queue_atomics.rs
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@ impl RunQueue {
// safety: there are no concurrent accesses to `next`
next = unsafe { task.header().run_queue_item.next.get() };

task.header().state.run_dequeue();
on_task(task);
}
}
Expand Down
7 changes: 4 additions & 3 deletions embassy-executor/src/raw/run_queue_critical_section.rs
Original file line number Diff line number Diff line change
Expand Up @@ -63,9 +63,10 @@ impl RunQueue {
// If the task re-enqueues itself, the `next` pointer will get overwritten.
// Therefore, first read the next pointer, and only then process the task.

// safety: we know if the task is enqueued, no one else will touch the `next` pointer.
let cs = unsafe { CriticalSection::new() };
next = task.header().run_queue_item.next.borrow(cs).get();
critical_section::with(|cs| {
next = task.header().run_queue_item.next.borrow(cs).get();
task.header().state.run_dequeue(cs);
});

on_task(task);
}
Expand Down
5 changes: 2 additions & 3 deletions embassy-executor/src/raw/state_atomics.rs
Original file line number Diff line number Diff line change
Expand Up @@ -52,8 +52,7 @@ impl State {

/// Unmark the task as run-queued. Return whether the task is spawned.
#[inline(always)]
pub fn run_dequeue(&self) -> bool {
let state = self.state.fetch_and(!STATE_RUN_QUEUED, Ordering::AcqRel);
state & STATE_SPAWNED != 0
pub fn run_dequeue(&self) {
self.state.fetch_and(!STATE_RUN_QUEUED, Ordering::AcqRel);
}
}
4 changes: 1 addition & 3 deletions embassy-executor/src/raw/state_atomics_arm.rs
Original file line number Diff line number Diff line change
Expand Up @@ -75,11 +75,9 @@ impl State {

/// Unmark the task as run-queued. Return whether the task is spawned.
#[inline(always)]
pub fn run_dequeue(&self) -> bool {
pub fn run_dequeue(&self) {
compiler_fence(Ordering::Release);

let r = self.spawned.load(Ordering::Relaxed);
self.run_queued.store(false, Ordering::Relaxed);
r
}
}
8 changes: 2 additions & 6 deletions embassy-executor/src/raw/state_critical_section.rs
Original file line number Diff line number Diff line change
Expand Up @@ -67,11 +67,7 @@ impl State {

/// Unmark the task as run-queued. Return whether the task is spawned.
#[inline(always)]
pub fn run_dequeue(&self) -> bool {
self.update(|s| {
let ok = *s & STATE_SPAWNED != 0;
*s &= !STATE_RUN_QUEUED;
ok
})
pub fn run_dequeue(&self, cs: CriticalSection<'_>) {
self.update_with_cs(cs, |s| *s &= !STATE_RUN_QUEUED)
}
}
3 changes: 3 additions & 0 deletions embassy-executor/tests/test.rs
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ fn executor_task() {
&[
"pend", // spawning a task pends the executor
"poll task1", // poll only once.
"pend", // task is done, wakes itself to exit
]
)
}
Expand Down Expand Up @@ -179,6 +180,7 @@ fn waking_after_completion_does_not_poll() {
"pend", // manual wake, single pend for two wakes
"pend", // respawning a task pends the executor
"poll task1", //
"pend", // task is done, wakes itself to exit
]
)
}
Expand Down Expand Up @@ -266,6 +268,7 @@ fn waking_with_old_waker_after_respawn() {
"yield_now", //
"pend", // manual wake, gets cleared by poll
"poll task1", //
"pend", // task is done, wakes itself to exit
]
);
}
Expand Down