aboutsummaryrefslogtreecommitdiff
path: root/embassy-executor/src/executor/raw/mod.rs
diff options
context:
space:
mode:
Diffstat (limited to 'embassy-executor/src/executor/raw/mod.rs')
-rw-r--r--embassy-executor/src/executor/raw/mod.rs48
1 files changed, 21 insertions, 27 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")]