aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--embassy-executor/src/executor/raw/mod.rs48
-rw-r--r--embassy-executor/src/executor/raw/run_queue.rs6
-rw-r--r--embassy-executor/src/executor/raw/waker.rs4
3 files changed, 26 insertions, 32 deletions
diff --git a/embassy-executor/src/executor/raw/mod.rs b/embassy-executor/src/executor/raw/mod.rs
index 87317bc02..fb4cc6288 100644
--- a/embassy-executor/src/executor/raw/mod.rs
+++ b/embassy-executor/src/executor/raw/mod.rs
@@ -70,24 +70,6 @@ impl TaskHeader {
70 timer_queue_item: timer_queue::TimerQueueItem::new(), 70 timer_queue_item: timer_queue::TimerQueueItem::new(),
71 } 71 }
72 } 72 }
73
74 pub(crate) unsafe fn enqueue(&self) {
75 critical_section::with(|cs| {
76 let state = self.state.load(Ordering::Relaxed);
77
78 // If already scheduled, or if not started,
79 if (state & STATE_RUN_QUEUED != 0) || (state & STATE_SPAWNED == 0) {
80 return;
81 }
82
83 // Mark it as scheduled
84 self.state.store(state | STATE_RUN_QUEUED, Ordering::Relaxed);
85
86 // We have just marked the task as scheduled, so enqueue it.
87 let executor = &*self.executor.get();
88 executor.enqueue(cs, self as *const TaskHeader as *mut TaskHeader);
89 })
90 }
91} 73}
92 74
93/// Raw storage in which a task can be spawned. 75/// Raw storage in which a task can be spawned.
@@ -155,7 +137,7 @@ impl<F: Future + 'static> TaskStorage<F> {
155 // Initialize the task 137 // Initialize the task
156 self.raw.poll_fn.write(Self::poll); 138 self.raw.poll_fn.write(Self::poll);
157 self.future.write(future()); 139 self.future.write(future());
158 NonNull::new_unchecked(&self.raw as *const TaskHeader as *mut TaskHeader) 140 NonNull::new_unchecked(self as *const TaskStorage<F> as *const TaskHeader as *mut TaskHeader)
159 } 141 }
160 142
161 unsafe fn poll(p: NonNull<TaskHeader>) { 143 unsafe fn poll(p: NonNull<TaskHeader>) {
@@ -323,7 +305,7 @@ impl Executor {
323 /// - `task` must be set up to run in this executor. 305 /// - `task` must be set up to run in this executor.
324 /// - `task` must NOT be already enqueued (in this executor or another one). 306 /// - `task` must NOT be already enqueued (in this executor or another one).
325 #[inline(always)] 307 #[inline(always)]
326 unsafe fn enqueue(&self, cs: CriticalSection, task: *mut TaskHeader) { 308 unsafe fn enqueue(&self, cs: CriticalSection, task: NonNull<TaskHeader>) {
327 if self.run_queue.enqueue(cs, task) { 309 if self.run_queue.enqueue(cs, task) {
328 (self.signal_fn)(self.signal_ctx) 310 (self.signal_fn)(self.signal_ctx)
329 } 311 }
@@ -339,11 +321,10 @@ impl Executor {
339 /// In this case, the task's Future must be Send. This is because this is effectively 321 /// In this case, the task's Future must be Send. This is because this is effectively
340 /// sending the task to the executor thread. 322 /// sending the task to the executor thread.
341 pub(super) unsafe fn spawn(&'static self, task: NonNull<TaskHeader>) { 323 pub(super) unsafe fn spawn(&'static self, task: NonNull<TaskHeader>) {
342 let task = task.as_ref(); 324 task.as_ref().executor.set(self);
343 task.executor.set(self);
344 325
345 critical_section::with(|cs| { 326 critical_section::with(|cs| {
346 self.enqueue(cs, task as *const _ as _); 327 self.enqueue(cs, task);
347 }) 328 })
348 } 329 }
349 330
@@ -366,9 +347,7 @@ impl Executor {
366 /// no `poll()` already running. 347 /// no `poll()` already running.
367 pub unsafe fn poll(&'static self) { 348 pub unsafe fn poll(&'static self) {
368 #[cfg(feature = "time")] 349 #[cfg(feature = "time")]
369 self.timer_queue.dequeue_expired(Instant::now(), |p| { 350 self.timer_queue.dequeue_expired(Instant::now(), |task| wake_task(task));
370 p.as_ref().enqueue();
371 });
372 351
373 self.run_queue.dequeue_all(|p| { 352 self.run_queue.dequeue_all(|p| {
374 let task = p.as_ref(); 353 let task = p.as_ref();
@@ -421,7 +400,22 @@ impl Executor {
421/// 400///
422/// `task` must be a valid task pointer obtained from [`task_from_waker`]. 401/// `task` must be a valid task pointer obtained from [`task_from_waker`].
423pub unsafe fn wake_task(task: NonNull<TaskHeader>) { 402pub unsafe fn wake_task(task: NonNull<TaskHeader>) {
424 task.as_ref().enqueue(); 403 critical_section::with(|cs| {
404 let header = task.as_ref();
405 let state = header.state.load(Ordering::Relaxed);
406
407 // If already scheduled, or if not started,
408 if (state & STATE_RUN_QUEUED != 0) || (state & STATE_SPAWNED == 0) {
409 return;
410 }
411
412 // Mark it as scheduled
413 header.state.store(state | STATE_RUN_QUEUED, Ordering::Relaxed);
414
415 // We have just marked the task as scheduled, so enqueue it.
416 let executor = &*header.executor.get();
417 executor.enqueue(cs, task);
418 })
425} 419}
426 420
427#[cfg(feature = "time")] 421#[cfg(feature = "time")]
diff --git a/embassy-executor/src/executor/raw/run_queue.rs b/embassy-executor/src/executor/raw/run_queue.rs
index 31615da7e..ed8c82a5c 100644
--- a/embassy-executor/src/executor/raw/run_queue.rs
+++ b/embassy-executor/src/executor/raw/run_queue.rs
@@ -46,10 +46,10 @@ impl RunQueue {
46 /// 46 ///
47 /// `item` must NOT be already enqueued in any queue. 47 /// `item` must NOT be already enqueued in any queue.
48 #[inline(always)] 48 #[inline(always)]
49 pub(crate) unsafe fn enqueue(&self, _cs: CriticalSection, task: *mut TaskHeader) -> bool { 49 pub(crate) unsafe fn enqueue(&self, _cs: CriticalSection, task: NonNull<TaskHeader>) -> bool {
50 let prev = self.head.load(Ordering::Relaxed); 50 let prev = self.head.load(Ordering::Relaxed);
51 (*task).run_queue_item.next.store(prev, Ordering::Relaxed); 51 task.as_ref().run_queue_item.next.store(prev, Ordering::Relaxed);
52 self.head.store(task, Ordering::Relaxed); 52 self.head.store(task.as_ptr(), Ordering::Relaxed);
53 prev.is_null() 53 prev.is_null()
54 } 54 }
55 55
diff --git a/embassy-executor/src/executor/raw/waker.rs b/embassy-executor/src/executor/raw/waker.rs
index f6ae332fa..6b9c03a62 100644
--- a/embassy-executor/src/executor/raw/waker.rs
+++ b/embassy-executor/src/executor/raw/waker.rs
@@ -2,7 +2,7 @@ use core::mem;
2use core::ptr::NonNull; 2use core::ptr::NonNull;
3use core::task::{RawWaker, RawWakerVTable, Waker}; 3use core::task::{RawWaker, RawWakerVTable, Waker};
4 4
5use super::TaskHeader; 5use super::{wake_task, TaskHeader};
6 6
7const VTABLE: RawWakerVTable = RawWakerVTable::new(clone, wake, wake, drop); 7const VTABLE: RawWakerVTable = RawWakerVTable::new(clone, wake, wake, drop);
8 8
@@ -11,7 +11,7 @@ unsafe fn clone(p: *const ()) -> RawWaker {
11} 11}
12 12
13unsafe fn wake(p: *const ()) { 13unsafe fn wake(p: *const ()) {
14 (*(p as *mut TaskHeader)).enqueue() 14 wake_task(NonNull::new_unchecked(p as *mut TaskHeader))
15} 15}
16 16
17unsafe fn drop(_: *const ()) { 17unsafe fn drop(_: *const ()) {