Merge pull request #635 from tier4/fix_priority_inversion_gedf

fix(GEDFScheduler): solve inter-scheduler priority inversion

Author: kobayu858

awkernel_async_lib/src/scheduler/gedf.rs

@@ -1,18 +1,17 @@
 //! A GEDF scheduler.
 
+use core::cmp::max;
+
 use super::{Scheduler, SchedulerType, Task};
 use crate::{
-    scheduler::get_priority,
+    scheduler::{get_priority, peek_preemption_pending, push_preemption_pending},
     task::{
-        get_absolute_deadline_by_task_id, get_tasks_running, set_need_preemption, State,
+        get_task, get_tasks_running, set_current_task, set_need_preemption, State,
         MAX_TASK_PRIORITY,
     },
 };
-use alloc::{collections::BinaryHeap, sync::Arc};
-use awkernel_lib::{
-    cpu::num_cpu,
-    sync::mutex::{MCSNode, Mutex},
-};
+use alloc::{collections::BinaryHeap, sync::Arc, vec::Vec};
+use awkernel_lib::sync::mutex::{MCSNode, Mutex};
 
 pub struct GEDFScheduler {
     data: Mutex<Option<GEDFData>>, // Run queue.
@@ -63,30 +62,36 @@ impl GEDFData {
 impl Scheduler for GEDFScheduler {
     fn wake_task(&self, task: Arc<Task>) {
         let mut node = MCSNode::new();
+        // The reason for acquiring this lock before invoke_preemption() is to prevent priority inversion from occurring
+        // when invoke_preemption() is executed between the time the next task is determined and the RUNNING is updated
+        // within the scheduler's get_next().
         let mut data = self.data.lock(&mut node);
-        let data = data.get_or_insert_with(GEDFData::new);
-
-        let mut node = MCSNode::new();
-        let mut info = task.info.lock(&mut node);
-
-        let SchedulerType::GEDF(relative_deadline) = info.scheduler_type else {
-            unreachable!();
+        let internal_data = data.get_or_insert_with(GEDFData::new);
+
+        let (wake_time, absolute_deadline) = {
+            let mut node_inner = MCSNode::new();
+            let mut info = task.info.lock(&mut node_inner);
+            match info.scheduler_type {
+                SchedulerType::GEDF(relative_deadline) => {
+                    let wake_time = awkernel_lib::delay::uptime();
+                    let absolute_deadline = wake_time + relative_deadline;
+                    task.priority
+                        .update_priority_info(self.priority, MAX_TASK_PRIORITY - absolute_deadline);
+                    info.update_absolute_deadline(absolute_deadline);
+
+                    (wake_time, absolute_deadline)
+                }
+                _ => unreachable!(),
+            }
         };
 
-        let wake_time = awkernel_lib::delay::uptime();
-        let absolute_deadline = wake_time + relative_deadline;
-
-        task.priority
-            .update_priority_info(self.priority, MAX_TASK_PRIORITY - absolute_deadline);
-        info.update_absolute_deadline(absolute_deadline);
-
-        data.queue.push(GEDFTask {
-            task: task.clone(),
-            absolute_deadline,
-            wake_time,
-        });
-
-        self.invoke_preemption(absolute_deadline);
+        if !self.invoke_preemption(task.clone()) {
+            internal_data.queue.push(GEDFTask {
+                task: task.clone(),
+                absolute_deadline,
+                wake_time,
+            });
+        }
     }
 
     fn get_next(&self) -> Option<Arc<Task>> {
@@ -116,6 +121,7 @@ impl Scheduler for GEDFScheduler {
                     task_info.need_preemption = false;
                 }
                 task_info.state = State::Running;
+                set_current_task(awkernel_lib::cpu::cpu_id(), task.task.id);
             }
 
             return Some(task.task);
@@ -137,30 +143,42 @@ pub static SCHEDULER: GEDFScheduler = GEDFScheduler {
 };
 
 impl GEDFScheduler {
-    fn invoke_preemption(&self, absolute_deadline: u64) {
-        // Get running tasks and filter out tasks with task_id == 0.
-        let mut tasks = get_tasks_running();
-        tasks.retain(|task| task.task_id != 0);
-
-        // If the number of running tasks is less than the number of non-primary CPUs, preempt is not required.
-        let num_non_primary_cpus = num_cpu() - 1;
-        if tasks.len() < num_non_primary_cpus {
-            return;
+    fn invoke_preemption(&self, task: Arc<Task>) -> bool {
+        let tasks_running = get_tasks_running()
+            .into_iter()
+            .filter(|rt| rt.task_id != 0) // Filter out idle CPUs
+            .collect::<Vec<_>>();
+
+        // If the task has already been running, preempt is not required.
+        if tasks_running.is_empty() || tasks_running.iter().any(|rt| rt.task_id == task.id) {
+            return false;
         }
 
-        let task_with_max_deadline = tasks
+        let preemption_target = tasks_running
             .iter()
-            .filter_map(|task| {
-                get_absolute_deadline_by_task_id(task.task_id).map(|deadline| (task, deadline))
+            .filter_map(|rt| {
+                get_task(rt.task_id).map(|t| {
+                    let highest_pending = peek_preemption_pending(rt.cpu_id).unwrap_or(t.clone());
+                    (max(t, highest_pending), rt.cpu_id)
+                })
             })
-            .max_by_key(|&(_, deadline)| deadline);
+            .min()
+            .unwrap();
 
-        if let Some((task, max_absolute_deadline)) = task_with_max_deadline {
-            if max_absolute_deadline > absolute_deadline {
-                let preempt_irq = awkernel_lib::interrupt::get_preempt_irq();
-                set_need_preemption(task.task_id, task.cpu_id);
-                awkernel_lib::interrupt::send_ipi(preempt_irq, task.cpu_id as u32);
-            }
+        let (target_task, target_cpu) = preemption_target;
+        if task > target_task {
+            push_preemption_pending(target_cpu, task);
+            let preempt_irq = awkernel_lib::interrupt::get_preempt_irq();
+            set_need_preemption(target_task.id, target_cpu);
+            awkernel_lib::interrupt::send_ipi(preempt_irq, target_cpu as u32);
+
+            // NOTE(atsushi421): Currently, preemption is requested regardless of the number of idle CPUs.
+            // While this implementation easily prevents priority inversion, it may also cause unnecessary preemption.
+            // Therefore, a more sophisticated implementation will be considered in the future.
+
+            return true;
         }
+
+        false
     }
 }

awkernel_async_lib/src/task.rs

@@ -942,18 +942,6 @@ pub fn get_scheduler_type_by_task_id(task_id: u32) -> Option<SchedulerType> {
     })
 }
 
-#[inline(always)]
-pub fn get_absolute_deadline_by_task_id(task_id: u32) -> Option<u64> {
-    let mut node = MCSNode::new();
-    let tasks = TASKS.lock(&mut node);
-
-    tasks.id_to_task.get(&task_id).and_then(|task| {
-        let mut node = MCSNode::new();
-        let info = task.info.lock(&mut node);
-        info.get_absolute_deadline()
-    })
-}
-
 #[inline(always)]
 pub fn set_need_preemption(task_id: u32, cpu_id: usize) {
     let mut node = MCSNode::new();