feat(dag): implement GEDF

applications/tests/test_dag/src/lib.rs

@@ -7,7 +7,7 @@ use awkernel_async_lib::scheduler::SchedulerType;
 use awkernel_lib::delay::wait_microsec;
 use core::time::Duration;
 
-const LOG_ENABLE: bool = false;
+const LOG_ENABLE: bool = true;
 
 pub async fn run() {
     wait_microsec(1000000);
@@ -30,7 +30,7 @@ pub async fn run() {
             (number,)
         },
         vec![Cow::from("topic0")],
-        SchedulerType::PrioritizedFIFO(30),
+        SchedulerType::GEDF(5),
         period,
     )
     .await;
@@ -47,7 +47,7 @@ pub async fn run() {
         },
         vec![Cow::from("topic0")],
         vec![Cow::from("topic1"), Cow::from("topic2")],
-        SchedulerType::PrioritizedFIFO(0),
+        SchedulerType::GEDF(10),
     )
     .await;
 
@@ -62,7 +62,7 @@ pub async fn run() {
         },
         vec![Cow::from("topic1")],
         vec![Cow::from("topic3")],
-        SchedulerType::PrioritizedFIFO(0),
+        SchedulerType::GEDF(10),
     )
     .await;
 
@@ -77,7 +77,7 @@ pub async fn run() {
         },
         vec![Cow::from("topic2")],
         vec![Cow::from("topic4")],
-        SchedulerType::PrioritizedFIFO(0),
+        SchedulerType::GEDF(10),
     )
     .await;
 
@@ -90,7 +90,7 @@ pub async fn run() {
             }
         },
         vec![Cow::from("topic3"), Cow::from("topic4")],
-        SchedulerType::PrioritizedFIFO(0),
+        SchedulerType::GEDF(10),
         Duration::from_secs(1),
     )
     .await;

awkernel_async_lib/src/dag.rs

@@ -63,6 +63,7 @@ use crate::{
         visit::{EdgeRef, IntoNodeReferences, NodeRef},
     },
     scheduler::SchedulerType,
+    task::DagInfo,
     time_interval::interval,
     Attribute, MultipleReceiver, MultipleSender, VectorToPublishers, VectorToSubscribers,
 };
@@ -168,6 +169,7 @@ impl core::fmt::Display for DagError {
 pub struct Dag {
     id: u32,
     graph: Mutex<graph::Graph<NodeInfo, EdgeInfo>>,
+    absolute_deadline: Mutex<Option<u64>>,
 
     #[cfg(feature = "perf")]
     response_info: Mutex<ResponseInfo>,
@@ -190,6 +192,25 @@ impl Dag {
         graph.externals(Direction::Outgoing).collect()
     }
 
+    pub fn is_source_node(&self, node_index: NodeIndex) -> bool {
+        let source_nodes = self.get_source_nodes();
+        source_nodes.contains(&node_index)
+    }
+
+    // Returns the relative deadline of the first sink node, if it exists.
+    pub fn get_sink_relative_deadline(&self) -> Option<Duration> {
+        let sink_nodes: Vec<NodeIndex> = self.get_sink_nodes();
+
+        if let Some(sink_node_index) = sink_nodes.first() {
+            self.graph
+                .lock(&mut MCSNode::new())
+                .node_weight(*sink_node_index)?
+                .relative_deadline
+        } else {
+            None
+        }
+    }
+
     fn set_relative_deadline(&self, node_idx: NodeIndex, deadline: Duration) {
         let mut node = MCSNode::new();
         let mut graph = self.graph.lock(&mut node);
@@ -284,6 +305,10 @@ impl Dag {
         self.check_subscribe_mismatch::<Args>(&subscribe_topic_names, node_idx);
         self.check_publish_mismatch::<Ret>(&publish_topic_names, node_idx);
 
+        // To prevent errors caused by ownership moves
+        let dag_id = self.id;
+        let node_id = node_idx.index() as u32;
+
         let mut node = MCSNode::new();
         let mut pending_tasks = PENDING_TASKS.lock(&mut node);
         pending_tasks
@@ -297,6 +322,7 @@ impl Dag {
                         subscribe_topic_names,
                         publish_topic_names,
                         sched_type,
+                        DagInfo { dag_id, node_id },
                     )
                     .await
                 })
@@ -330,6 +356,10 @@ impl Dag {
             }
         };
 
+        // To prevent errors caused by ownership moves
+        let dag_id = self.id;
+        let node_id = node_idx.index() as u32;
+
         let mut node = MCSNode::new();
         let mut source_pending_tasks = SOURCE_PENDING_TASKS.lock(&mut node);
         source_pending_tasks.insert(
@@ -343,6 +373,7 @@ impl Dag {
                         sched_type,
                         period,
                         measure_f,
+                        DagInfo { dag_id, node_id },
                     )
                     .await
                 })
@@ -384,6 +415,10 @@ impl Dag {
             }
         };
 
+        // To prevent errors caused by ownership moves
+        let dag_id = self.id;
+        let node_id = node_idx.index() as u32;
+
         let mut node = MCSNode::new();
         let mut pending_tasks = PENDING_TASKS.lock(&mut node);
         pending_tasks
@@ -396,6 +431,7 @@ impl Dag {
                         measure_f,
                         subscribe_topic_names,
                         sched_type,
+                        DagInfo { dag_id, node_id },
                     )
                     .await
                 })
@@ -437,6 +473,20 @@ impl Dag {
                 .push(node_idx.index());
         }
     }
+
+    #[inline(always)]
+    pub fn set_absolute_deadline(&self, deadline: u64) {
+        let mut node = MCSNode::new();
+        let mut absolute_deadline = self.absolute_deadline.lock(&mut node);
+        *absolute_deadline = Some(deadline);
+    }
+
+    #[inline(always)]
+    pub fn get_absolute_deadline(&self) -> Option<u64> {
+        let mut node = MCSNode::new();
+        let absolute_deadline = self.absolute_deadline.lock(&mut node);
+        *absolute_deadline
+    }
 }
 
 struct PendingTask {
@@ -463,6 +513,10 @@ struct NodeInfo {
     relative_deadline: Option<Duration>,
 }
 
+pub fn to_node_index(index: u32) -> NodeIndex {
+    NodeIndex::new(index as usize)
+}
+
 struct EdgeInfo {
     topic_name: Cow<'static, str>,
 }
@@ -489,6 +543,7 @@ impl Dags {
                 let dag = Arc::new(Dag {
                     id,
                     graph: Mutex::new(graph::Graph::new()),
+                    absolute_deadline: Mutex::new(None),
 
                     #[cfg(feature = "perf")]
                     response_info: Mutex::new(ResponseInfo::new()),
@@ -525,6 +580,21 @@ pub fn get_dag(id: u32) -> Option<Arc<Dag>> {
     dags.id_to_dag.get(&id).cloned()
 }
 
+#[inline(always)]
+pub fn get_dag_absolute_deadline(dag_id: u32) -> Option<u64> {
+    get_dag(dag_id)?.get_absolute_deadline()
+}
+
+#[inline(always)]
+pub fn set_dag_absolute_deadline(dag_id: u32, deadline: u64) -> bool {
+    if let Some(dag) = get_dag(dag_id) {
+        dag.set_absolute_deadline(deadline);
+        true
+    } else {
+        false
+    }
+}
+
 pub async fn finish_create_dags(dags: &[Arc<Dag>]) -> Result<(), Vec<DagError>> {
     match validate_all_rules(dags) {
         Ok(()) => {
@@ -833,6 +903,7 @@ async fn spawn_reactor<F, Args, Ret>(
     subscribe_topic_names: Vec<Cow<'static, str>>,
     publish_topic_names: Vec<Cow<'static, str>>,
     sched_type: SchedulerType,
+    dag_info: DagInfo,
 ) -> u32
 where
     F: Fn(
@@ -862,7 +933,7 @@ where
         }
     };
 
-    crate::task::spawn(reactor_name, future, sched_type)
+    crate::task::spawn_with_dag_info(reactor_name, future, sched_type, dag_info)
 }
 
 async fn spawn_periodic_reactor<F, Ret>(
@@ -872,6 +943,7 @@ async fn spawn_periodic_reactor<F, Ret>(
     sched_type: SchedulerType,
     period: Duration,
     _release_measure: Option<MeasureF>,
+    dag_info: DagInfo,
 ) -> u32
 where
     F: Fn() -> <Ret::Publishers as MultipleSender>::Item + Send + 'static,
@@ -909,7 +981,7 @@ where
         }
     };
 
-    let task_id = crate::task::spawn(reactor_name, future, sched_type);
+    let task_id = crate::task::spawn_with_dag_info(reactor_name, future, sched_type, dag_info);
 
     #[cfg(feature = "perf")]
     release_measure();
@@ -922,6 +994,7 @@ async fn spawn_sink_reactor<F, Args>(
     f: F,
     subscribe_topic_names: Vec<Cow<'static, str>>,
     sched_type: SchedulerType,
+    dag_info: DagInfo,
 ) -> u32
 where
     F: Fn(<Args::Subscribers as MultipleReceiver>::Item) + Send + 'static,
@@ -933,11 +1006,10 @@ where
             Args::create_subscribers(subscribe_topic_names, Attribute::default());
 
         loop {
-            let args: <<Args as VectorToSubscribers>::Subscribers as MultipleReceiver>::Item =
-                subscribers.recv_all().await;
+            let args: <Args::Subscribers as MultipleReceiver>::Item = subscribers.recv_all().await;
             f(args);
         }
     };
 
-    crate::task::spawn(reactor_name, future, sched_type)
+    crate::task::spawn_with_dag_info(reactor_name, future, sched_type, dag_info)
 }

awkernel_async_lib/src/scheduler/gedf.rs

@@ -4,9 +4,10 @@ use core::cmp::max;
 
 use super::{Scheduler, SchedulerType, Task};
 use crate::{
+    dag::{get_dag, get_dag_absolute_deadline, set_dag_absolute_deadline, to_node_index},
     scheduler::{get_priority, peek_preemption_pending, push_preemption_pending},
     task::{
-        get_task, get_tasks_running, set_current_task, set_need_preemption, State,
+        get_task, get_tasks_running, set_current_task, set_need_preemption, DagInfo, State,
         MAX_TASK_PRIORITY,
     },
 };
@@ -71,10 +72,18 @@ impl Scheduler for GEDFScheduler {
         let (wake_time, absolute_deadline) = {
             let mut node_inner = MCSNode::new();
             let mut info = task.info.lock(&mut node_inner);
+            let dag_info = info.get_dag_info();
             match info.scheduler_type {
                 SchedulerType::GEDF(relative_deadline) => {
                     let wake_time = awkernel_lib::delay::uptime();
-                    let absolute_deadline = wake_time + relative_deadline;
+                    let absolute_deadline = if let Some(ref dag_info) = dag_info {
+                        calculate_and_update_dag_deadline(dag_info, wake_time)
+                    } else {
+                        // If dag_info is not present, the task is treated as a regular task, and
+                        // the absolute_deadline is calculated using the scheduler's relative_deadline.
+                        wake_time + relative_deadline
+                    };
+
                     task.priority
                         .update_priority_info(self.priority, MAX_TASK_PRIORITY - absolute_deadline);
                     info.update_absolute_deadline(absolute_deadline);
@@ -184,3 +193,35 @@ impl GEDFScheduler {
         false
     }
 }
+
+fn get_dag_sink_relative_deadline_ms(dag_id: u32) -> u64 {
+    let dag = get_dag(dag_id).unwrap_or_else(|| panic!("GEDF scheduler: DAG {dag_id} not found"));
+    dag.get_sink_relative_deadline()
+        .map(|deadline| deadline.as_millis() as u64)
+        .unwrap_or_else(|| panic!("GEDF scheduler: DAG {dag_id} has no sink relative deadline set"))
+}
+
+fn calculate_and_set_dag_deadline(dag_id: u32, wake_time: u64) -> u64 {
+    let relative_deadline_ms = get_dag_sink_relative_deadline_ms(dag_id);
+    let dag_absolute_deadline = wake_time + relative_deadline_ms;
+    set_dag_absolute_deadline(dag_id, dag_absolute_deadline);
+    dag_absolute_deadline
+}
+
+pub fn calculate_and_update_dag_deadline(dag_info: &DagInfo, wake_time: u64) -> u64 {
+    let dag_id = dag_info.dag_id;
+    let node_id = dag_info.node_id;
+
+    if let Some(absolute_deadline) = get_dag_absolute_deadline(dag_id) {
+        let dag =
+            get_dag(dag_id).unwrap_or_else(|| panic!("GEDF scheduler: DAG {dag_id} not found"));
+        let current_node_index = to_node_index(node_id);
+        if !dag.is_source_node(current_node_index) {
+            return absolute_deadline;
+        }
+
+        return calculate_and_set_dag_deadline(dag_id, wake_time);
+    }
+
+    calculate_and_set_dag_deadline(dag_id, wake_time)
+}