aboutsummaryrefslogtreecommitdiff
path: root/embassy-executor
diff options
context:
space:
mode:
Diffstat (limited to 'embassy-executor')
-rw-r--r--embassy-executor/src/raw/mod.rs76
-rw-r--r--embassy-executor/src/raw/trace.rs190
-rw-r--r--embassy-executor/src/spawner.rs3
3 files changed, 163 insertions, 106 deletions
diff --git a/embassy-executor/src/raw/mod.rs b/embassy-executor/src/raw/mod.rs
index 075d8a254..b4adfe01b 100644
--- a/embassy-executor/src/raw/mod.rs
+++ b/embassy-executor/src/raw/mod.rs
@@ -93,6 +93,78 @@ pub(crate) struct TaskHeader {
93 pub(crate) name: Option<&'static str>, 93 pub(crate) name: Option<&'static str>,
94 #[cfg(feature = "trace")] 94 #[cfg(feature = "trace")]
95 pub(crate) id: u32, 95 pub(crate) id: u32,
96 #[cfg(feature = "trace")]
97 all_tasks_next: AtomicPtr<TaskHeader>,
98}
99
100/// A thread-safe tracker for all tasks in the system
101///
102/// This struct uses an intrusive linked list approach to track all tasks
103/// without additional memory allocations. It maintains a global list of
104/// tasks that can be traversed to find all currently existing tasks.
105#[cfg(feature = "trace")]
106pub struct TaskTracker {
107 head: AtomicPtr<TaskHeader>,
108}
109
110#[cfg(feature = "trace")]
111impl TaskTracker {
112 /// Creates a new empty task tracker
113 ///
114 /// Initializes a tracker with no tasks in its list.
115 pub const fn new() -> Self {
116 Self {
117 head: AtomicPtr::new(core::ptr::null_mut()),
118 }
119 }
120
121 /// Adds a task to the tracker
122 ///
123 /// This method inserts a task at the head of the intrusive linked list.
124 /// The operation is thread-safe and lock-free, using atomic operations
125 /// to ensure consistency even when called from different contexts.
126 ///
127 /// # Arguments
128 /// * `task` - The task reference to add to the tracker
129 pub fn add(&self, task: TaskRef) {
130 let task_ptr = task.as_ptr() as *mut TaskHeader;
131
132 loop {
133 let current_head = self.head.load(Ordering::Acquire);
134 unsafe {
135 (*task_ptr).all_tasks_next.store(current_head, Ordering::Relaxed);
136 }
137
138 if self
139 .head
140 .compare_exchange(current_head, task_ptr, Ordering::Release, Ordering::Relaxed)
141 .is_ok()
142 {
143 break;
144 }
145 }
146 }
147
148 /// Performs an operation on each task in the tracker
149 ///
150 /// This method traverses the entire list of tasks and calls the provided
151 /// function for each task. This allows inspecting or processing all tasks
152 /// in the system without modifying the tracker's structure.
153 ///
154 /// # Arguments
155 /// * `f` - A function to call for each task in the tracker
156 pub fn for_each<F>(&self, mut f: F)
157 where
158 F: FnMut(TaskRef),
159 {
160 let mut current = self.head.load(Ordering::Acquire);
161 while !current.is_null() {
162 let task = unsafe { TaskRef::from_ptr(current) };
163 f(task);
164
165 current = unsafe { (*current).all_tasks_next.load(Ordering::Acquire) };
166 }
167 }
96} 168}
97 169
98/// This is essentially a `&'static TaskStorage<F>` where the type of the future has been erased. 170/// This is essentially a `&'static TaskStorage<F>` where the type of the future has been erased.
@@ -173,7 +245,7 @@ impl TaskRef {
173 #[cfg(feature = "trace")] 245 #[cfg(feature = "trace")]
174 pub fn id(&self) -> u32 { 246 pub fn id(&self) -> u32 {
175 self.header().id 247 self.header().id
176 } 248 }
177 249
178 /// Set the ID for a task 250 /// Set the ID for a task
179 #[cfg(feature = "trace")] 251 #[cfg(feature = "trace")]
@@ -228,6 +300,8 @@ impl<F: Future + 'static> TaskStorage<F> {
228 name: None, 300 name: None,
229 #[cfg(feature = "trace")] 301 #[cfg(feature = "trace")]
230 id: 0, 302 id: 0,
303 #[cfg(feature = "trace")]
304 all_tasks_next: AtomicPtr::new(core::ptr::null_mut()),
231 }, 305 },
232 future: UninitCell::uninit(), 306 future: UninitCell::uninit(),
233 } 307 }
diff --git a/embassy-executor/src/raw/trace.rs b/embassy-executor/src/raw/trace.rs
index 28be79cee..81c8a0024 100644
--- a/embassy-executor/src/raw/trace.rs
+++ b/embassy-executor/src/raw/trace.rs
@@ -81,107 +81,19 @@
81 81
82#![allow(unused)] 82#![allow(unused)]
83 83
84use crate::raw::{SyncExecutor, TaskHeader, TaskRef}; 84use crate::raw::{SyncExecutor, TaskHeader, TaskRef, TaskTracker};
85 85
86use core::cell::UnsafeCell; 86use core::cell::UnsafeCell;
87use core::sync::atomic::{AtomicUsize, Ordering}; 87use core::sync::atomic::{AtomicUsize, Ordering};
88use rtos_trace::TaskInfo; 88use rtos_trace::TaskInfo;
89 89
90const MAX_TASKS: usize = 1000; 90/// Global task tracker instance
91
92/// Represents a task being tracked in the task registry.
93///
94/// Contains the task's unique identifier and optional name.
95#[derive(Clone)]
96pub struct TrackedTask {
97 task_id: u32,
98}
99
100/// A thread-safe registry for tracking tasks in the system.
101///
102/// This registry maintains a list of active tasks with their IDs and optional names.
103/// It supports registering, unregistering, and querying information about tasks.
104/// The registry has a fixed capacity of `MAX_TASKS`.
105pub struct TaskRegistry {
106 tasks: [UnsafeCell<Option<TrackedTask>>; MAX_TASKS],
107 count: AtomicUsize,
108}
109
110impl TaskRegistry {
111 /// Creates a new empty task registry.
112 ///
113 /// This initializes a registry that can track up to `MAX_TASKS` tasks.
114 pub const fn new() -> Self {
115 const EMPTY: UnsafeCell<Option<TrackedTask>> = UnsafeCell::new(None);
116 Self {
117 tasks: [EMPTY; MAX_TASKS],
118 count: AtomicUsize::new(0),
119 }
120 }
121
122 /// Registers a new task in the registry.
123 ///
124 /// # Arguments
125 /// * `task_id` - Unique identifier for the task
126 /// * `name` - Optional name for the task
127 ///
128 /// # Note
129 /// If the registry is full, the task will not be registered.
130 pub fn register(&self, task_id: u32) {
131 let count = self.count.load(Ordering::Relaxed);
132 if count < MAX_TASKS {
133 for i in 0..MAX_TASKS {
134 unsafe {
135 let slot = &self.tasks[i];
136 let slot_ref = &mut *slot.get();
137 if slot_ref.is_none() {
138 *slot_ref = Some(TrackedTask { task_id });
139 self.count.fetch_add(1, Ordering::Relaxed);
140 break;
141 }
142 }
143 }
144 }
145 }
146
147 /// Removes a task from the registry.
148 ///
149 /// # Arguments
150 /// * `task_id` - Unique identifier of the task to remove
151 pub fn unregister(&self, task_id: u32) {
152 for i in 0..MAX_TASKS {
153 unsafe {
154 let slot = &self.tasks[i];
155 let slot_ref = &mut *slot.get();
156 if let Some(task) = slot_ref {
157 if task.task_id == task_id {
158 *slot_ref = None;
159 self.count.fetch_sub(1, Ordering::Relaxed);
160 break;
161 }
162 }
163 }
164 }
165 }
166
167 /// Returns an iterator over all registered tasks.
168 ///
169 /// This allows accessing information about all tasks currently in the registry.
170 pub fn get_all_tasks(&self) -> impl Iterator<Item = TrackedTask> + '_ {
171 (0..MAX_TASKS).filter_map(move |i| unsafe {
172 let slot = &self.tasks[i];
173 (*slot.get()).clone()
174 })
175 }
176}
177
178unsafe impl Sync for TaskRegistry {}
179unsafe impl Send for TaskRegistry {}
180
181/// Global task registry instance used for tracking all tasks in the system.
182/// 91///
183/// This provides a centralized registry accessible from anywhere in the application. 92/// This static provides access to the global task tracker which maintains
184pub static TASK_REGISTRY: TaskRegistry = TaskRegistry::new(); 93/// a list of all tasks in the system. It's automatically updated by the
94/// task lifecycle hooks in the trace module.
95#[cfg(feature = "trace")]
96pub static TASK_TRACKER: TaskTracker = TaskTracker::new();
185 97
186#[cfg(not(feature = "rtos-trace"))] 98#[cfg(not(feature = "rtos-trace"))]
187extern "Rust" { 99extern "Rust" {
@@ -262,6 +174,9 @@ pub(crate) fn task_new(executor: &SyncExecutor, task: &TaskRef) {
262 174
263 #[cfg(feature = "rtos-trace")] 175 #[cfg(feature = "rtos-trace")]
264 rtos_trace::trace::task_new(task.as_ptr() as u32); 176 rtos_trace::trace::task_new(task.as_ptr() as u32);
177
178 #[cfg(feature = "rtos-trace")]
179 TASK_TRACKER.add(*task);
265} 180}
266 181
267#[inline] 182#[inline]
@@ -272,8 +187,6 @@ pub(crate) fn task_end(executor: *const SyncExecutor, task: &TaskRef) {
272 unsafe { 187 unsafe {
273 _embassy_trace_task_end(executor as u32, task.as_ptr() as u32) 188 _embassy_trace_task_end(executor as u32, task.as_ptr() as u32)
274 } 189 }
275
276 TASK_REGISTRY.unregister(task_id);
277} 190}
278 191
279#[inline] 192#[inline]
@@ -316,20 +229,93 @@ pub(crate) fn executor_idle(executor: &SyncExecutor) {
316 rtos_trace::trace::system_idle(); 229 rtos_trace::trace::system_idle();
317} 230}
318 231
232/// Returns an iterator over all active tasks in the system
233///
234/// This function provides a convenient way to iterate over all tasks
235/// that are currently tracked in the system. The returned iterator
236/// yields each task in the global task tracker.
237///
238/// # Returns
239/// An iterator that yields `TaskRef` items for each task
240#[cfg(feature = "trace")]
241pub fn get_all_active_tasks() -> impl Iterator<Item = TaskRef> + 'static {
242 struct TaskIterator<'a> {
243 tracker: &'a TaskTracker,
244 current: *mut TaskHeader,
245 }
246
247 impl<'a> Iterator for TaskIterator<'a> {
248 type Item = TaskRef;
249
250 fn next(&mut self) -> Option<Self::Item> {
251 if self.current.is_null() {
252 return None;
253 }
254
255 let task = unsafe { TaskRef::from_ptr(self.current) };
256 self.current = unsafe { (*self.current).all_tasks_next.load(Ordering::Acquire) };
257
258 Some(task)
259 }
260 }
261
262 TaskIterator {
263 tracker: &TASK_TRACKER,
264 current: TASK_TRACKER.head.load(Ordering::Acquire),
265 }
266}
267
268/// Get all active tasks, filtered by a predicate function
269#[cfg(feature = "trace")]
270pub fn filter_active_tasks<F>(predicate: F) -> impl Iterator<Item = TaskRef> + 'static
271where
272 F: Fn(&TaskRef) -> bool + 'static,
273{
274 get_all_active_tasks().filter(move |task| predicate(task))
275}
276
277/// Count the number of active tasks
278#[cfg(feature = "trace")]
279pub fn count_active_tasks() -> usize {
280 let mut count = 0;
281 TASK_TRACKER.for_each(|_| count += 1);
282 count
283}
284
285/// Perform an action on each active task
286#[cfg(feature = "trace")]
287pub fn with_all_active_tasks<F>(f: F)
288where
289 F: FnMut(TaskRef),
290{
291 TASK_TRACKER.for_each(f);
292}
293
294/// Get tasks by name
295#[cfg(feature = "trace")]
296pub fn get_tasks_by_name(name: &'static str) -> impl Iterator<Item = TaskRef> + 'static {
297 filter_active_tasks(move |task| task.name() == Some(name))
298}
299
300/// Get tasks by ID
301#[cfg(feature = "trace")]
302pub fn get_task_by_id(id: u32) -> Option<TaskRef> {
303 filter_active_tasks(move |task| task.id() == id).next()
304}
305
319#[cfg(feature = "rtos-trace")] 306#[cfg(feature = "rtos-trace")]
320impl rtos_trace::RtosTraceOSCallbacks for crate::raw::SyncExecutor { 307impl rtos_trace::RtosTraceOSCallbacks for crate::raw::SyncExecutor {
321 fn task_list() { 308 fn task_list() {
322 for task in TASK_REGISTRY.get_all_tasks() { 309 with_all_active_tasks(|task| {
323 let task_ref = unsafe { TaskRef::from_ptr(task.task_id as *const TaskHeader) }; 310 let name = task.name().unwrap_or("unnamed task\0");
324 let name = task_ref.name().unwrap_or("unnamed\0");
325 let info = rtos_trace::TaskInfo { 311 let info = rtos_trace::TaskInfo {
326 name, 312 name,
327 priority: 0, 313 priority: 0,
328 stack_base: 0, 314 stack_base: 0,
329 stack_size: 0, 315 stack_size: 0,
330 }; 316 };
331 rtos_trace::trace::task_send_info(task.task_id, info); 317 rtos_trace::trace::task_send_info(task.as_id(), info);
332 } 318 });
333 } 319 }
334 fn time() -> u64 { 320 fn time() -> u64 {
335 const fn gcd(a: u64, b: u64) -> u64 { 321 const fn gcd(a: u64, b: u64) -> u64 {
diff --git a/embassy-executor/src/spawner.rs b/embassy-executor/src/spawner.rs
index 7f907346d..5e42f01bf 100644
--- a/embassy-executor/src/spawner.rs
+++ b/embassy-executor/src/spawner.rs
@@ -5,8 +5,6 @@ use core::sync::atomic::Ordering;
5use core::task::Poll; 5use core::task::Poll;
6 6
7use super::raw; 7use super::raw;
8#[cfg(feature = "rtos-trace")]
9use super::raw::trace::TASK_REGISTRY;
10 8
11/// Token to spawn a newly-created task in an executor. 9/// Token to spawn a newly-created task in an executor.
12/// 10///
@@ -173,7 +171,6 @@ impl Spawner {
173 Some(task) => { 171 Some(task) => {
174 task.set_name(Some(name)); 172 task.set_name(Some(name));
175 let task_id = task.as_ptr() as u32; 173 let task_id = task.as_ptr() as u32;
176 TASK_REGISTRY.register(task_id);
177 task.set_id(task_id); 174 task.set_id(task_id);
178 175
179 unsafe { self.executor.spawn(task) }; 176 unsafe { self.executor.spawn(task) };