Skip to content

Commit 3bf6490

Browse files
committed
zephyr: embassy: Use a semaphore for the executor
Fixes a potential race condition in the suspend/resume sequence. Signed-off-by: Aaron Olowin <[email protected]>
1 parent 349164d commit 3bf6490

File tree

1 file changed

+8
-30
lines changed

1 file changed

+8
-30
lines changed

zephyr/src/embassy/executor.rs

Lines changed: 8 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -1,29 +1,24 @@
11
//! An embassy executor tailored for Zephyr
22
3-
use core::{marker::PhantomData, sync::atomic::Ordering};
3+
use core::marker::PhantomData;
44

5+
use crate::sys::sync::Semaphore;
6+
use crate::time::Forever;
57
use embassy_executor::{raw, Spawner};
6-
use zephyr_sys::{k_current_get, k_thread_resume, k_thread_suspend, k_tid_t};
7-
8-
use crate::sync::atomic::AtomicBool;
98

109
/// Zephyr-thread based executor.
1110
pub struct Executor {
1211
inner: Option<raw::Executor>,
13-
id: k_tid_t,
14-
pend: AtomicBool,
12+
poll_needed: Semaphore,
1513
not_send: PhantomData<*mut ()>,
1614
}
1715

1816
impl Executor {
1917
/// Create a new Executor.
2018
pub fn new() -> Self {
21-
let id = unsafe { k_current_get() };
22-
2319
Self {
2420
inner: None,
25-
pend: AtomicBool::new(false),
26-
id,
21+
poll_needed: Semaphore::new(0, 1).unwrap(),
2722
not_send: PhantomData,
2823
}
2924
}
@@ -36,17 +31,13 @@ impl Executor {
3631
init(inner.spawner());
3732

3833
loop {
34+
let _ = self.poll_needed.take(Forever);
3935
unsafe {
4036
// The raw executor's poll only runs things that were queued _before_ this poll
4137
// itself is actually run. This means, specifically, that if the polled execution
4238
// causes this, or other threads to enqueue, this will return without running them.
43-
// `__pender` _will_ be called, but it isn't "sticky" like `wfe/sev` are. To
44-
// simulate this, we will use the 'pend' atomic to count
39+
// `__pender` _will_ be called, so the next time around the semaphore will be taken.
4540
inner.poll();
46-
if !self.pend.swap(false, Ordering::SeqCst) {
47-
// printkln!("_suspend");
48-
k_thread_suspend(k_current_get());
49-
}
5041
}
5142
}
5243
}
@@ -61,20 +52,7 @@ impl Default for Executor {
6152
#[export_name = "__pender"]
6253
fn __pender(context: *mut ()) {
6354
unsafe {
64-
let myself = k_current_get();
65-
6655
let this = context as *const Executor;
67-
let other = (*this).id;
68-
69-
// If the other is a different thread, resume it.
70-
if other != myself {
71-
// printkln!("_resume");
72-
k_thread_resume(other);
73-
}
74-
// Otherwise, we need to make sure our own next suspend doesn't happen.
75-
// We need to also prevent a suspend from happening in the case where the only running
76-
// thread causes other work to become pending. The resume below will do nothing, as we
77-
// are just running.
78-
(*this).pend.store(true, Ordering::SeqCst);
56+
(*this).poll_needed.give();
7957
}
8058
}

0 commit comments

Comments
 (0)