From 46209cbff0c5e4c13899fba82e2f835c8c980618 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rom=C3=A1n=20C=C3=A1rdenas=20Rodr=C3=ADguez?= Date: Mon, 19 Oct 2026 17:31:49 +0900 Subject: [PATCH 1/4] feat(dag): use release times instead of absolute deadlines --- awkernel_async_lib/src/dag.rs | 164 +++++++++++------------ awkernel_async_lib/src/scheduler/gedf.rs | 20 +-- 2 files changed, 91 insertions(+), 93 deletions(-) diff --git a/awkernel_async_lib/src/dag.rs b/awkernel_async_lib/src/dag.rs index 7ea9b7f86..4d8ee27f4 100644 --- a/awkernel_async_lib/src/dag.rs +++ b/awkernel_async_lib/src/dag.rs @@ -119,8 +119,10 @@ impl_tuple_size!(T1); impl_tuple_size!(T1, T2); impl_tuple_size!(T1, T2, T3); -#[derive(Clone)] +#[derive(Clone, Debug)] pub enum DagError { + NodeNotFound(u32, usize), + MissingSinkDeadline(u32), NotWeaklyConnected(u32), ContainsCycle(u32), MissingPendingTasks(u32), @@ -140,6 +142,8 @@ pub enum DagError { impl core::fmt::Display for DagError { fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { match self { + DagError::NodeNotFound(dag_id, node_id) => write!(f, "DAG#{dag_id} Node#{node_id} not found"), + DagError::MissingSinkDeadline(dag_id) => write!(f, "DAG#{dag_id} has no sink relative deadline"), DagError::NotWeaklyConnected(dag_id) => write!(f, "DAG#{dag_id} is not weakly connected"), DagError::ContainsCycle(dag_id) => write!(f, "DAG#{dag_id} contains a cycle"), DagError::MissingPendingTasks(dag_id) => write!(f, "DAG#{dag_id} has missing pending tasks"), @@ -173,13 +177,34 @@ impl core::fmt::Display for DagError { } } +type DagGraph = graph::Graph; + +impl DagGraph { + #[inline] + fn get_sink_relative_deadline(&self) -> Option { + let sink_node_index = self.externals(Direction::Outgoing).next()?; + self.node_weight(sink_node_index).unwrap().relative_deadline + } + + #[inline] + fn set_relative_deadline(&mut self, node_idx: NodeIndex, deadline: Duration) -> bool { + match self.node_weight_mut(node_idx) { + Some(info) => { + info.relative_deadline = Some(deadline); + true + } + None => false, + } + } +} + pub struct Dag { id: u32, - graph: Mutex>, + graph: Mutex, - /// Absolute deadline of each released job that a node has not finished, keyed by job number. + /// Release time of each released job that a node has not finished, keyed by job number. /// A job is one run of the DAG. It starts when the source node wakes for a new release. - absolute_deadlines: Mutex>, + release_times: Mutex>, #[cfg(feature = "perf")] response_info: Mutex, @@ -203,17 +228,25 @@ impl Dag { } // Returns the relative deadline of the first sink node, if it exists. - pub fn get_sink_relative_deadline(&self) -> Option { + pub fn get_sink_relative_deadline(&self) -> Result { let mut node = MCSNode::new(); let graph = self.graph.lock(&mut node); - let sink_node_index = graph.externals(Direction::Outgoing).next()?; - graph.node_weight(sink_node_index)?.relative_deadline + graph + .get_sink_relative_deadline() + .ok_or(DagError::MissingSinkDeadline(self.id)) } - fn set_relative_deadline(&self, node_idx: NodeIndex, deadline: Duration) { + fn set_relative_deadline( + &self, + node_idx: NodeIndex, + deadline: Duration, + ) -> Result<(), DagError> { let mut node = MCSNode::new(); let mut graph = self.graph.lock(&mut node); - graph.node_weight_mut(node_idx).unwrap().relative_deadline = Some(deadline); + match graph.set_relative_deadline(node_idx, deadline) { + true => Ok(()), + false => Err(DagError::NodeNotFound(self.id, node_idx.index())), + } } fn add_node_with_topic_edges( @@ -396,7 +429,8 @@ impl Dag { ::Item: TupleSize, { let node_idx = self.add_node_with_topic_edges(&subscribe_topic_names, &Vec::new()); - self.set_relative_deadline(node_idx, relative_deadline); + self.set_relative_deadline(node_idx, relative_deadline) + .unwrap(); self.check_subscribe_mismatch::(&subscribe_topic_names, node_idx); let measure_f = { @@ -475,17 +509,17 @@ impl Dag { } } - /// Returns the absolute deadline of the job that the node at `node_idx` processes next. + /// Returns the release time of the job that the node at `node_idx` processes next. /// If that job is not released and `release` is true, the source node releases it with - /// the value of `release_deadline`. Any other wake gets that value without a release. - pub(crate) fn get_or_release_absolute_deadline( + /// the value of `release_time`. Any other wake gets that value without a release. + pub(crate) fn get_or_release_time( &self, node_idx: NodeIndex, release: bool, - release_deadline: impl FnOnce() -> u64, + release_time: impl FnOnce() -> u64, ) -> u64 { let mut node = MCSNode::new(); - let mut absolute_deadlines = self.absolute_deadlines.lock(&mut node); + let mut release_times = self.release_times.lock(&mut node); let (job, release) = { let mut graph_node = MCSNode::new(); @@ -497,20 +531,19 @@ impl Dag { let node_info = graph.node_weight_mut(node_idx).unwrap(); let job = node_info.num_finished; let release = release && is_source; - node_info.deadline_job = - (release || absolute_deadlines.contains_key(&job)).then_some(job); + node_info.deadline_job = (release || release_times.contains_key(&job)).then_some(job); (job, release) }; - if let Some(&deadline) = absolute_deadlines.get(&job) { - return deadline; + if let Some(&released) = release_times.get(&job) { + return released; } - let release_deadline = release_deadline(); + let release_time = release_time(); if release { - absolute_deadlines.insert(job, release_deadline); + release_times.insert(job, release_time); } - release_deadline + release_time } /// Returns true if the job that the node at `node_idx` processes is released, but the @@ -518,19 +551,19 @@ impl Dag { /// in the queue and returns without `wake_task`. A yield then gives the node the deadline. fn has_stale_deadline(&self, node_idx: NodeIndex) -> bool { let mut node = MCSNode::new(); - let absolute_deadlines = self.absolute_deadlines.lock(&mut node); + let release_times = self.release_times.lock(&mut node); let mut graph_node = MCSNode::new(); let graph = self.graph.lock(&mut graph_node); let node_info = graph.node_weight(node_idx).unwrap(); node_info.deadline_job != Some(node_info.num_finished) - && absolute_deadlines.contains_key(&node_info.num_finished) + && release_times.contains_key(&node_info.num_finished) } /// Records that the node at `node_idx` finished its current job. fn finish_job(&self, node_idx: NodeIndex) { let mut node = MCSNode::new(); - let mut absolute_deadlines = self.absolute_deadlines.lock(&mut node); + let mut release_times = self.release_times.lock(&mut node); let mut graph_node = MCSNode::new(); let mut graph = self.graph.lock(&mut graph_node); @@ -543,14 +576,14 @@ impl Dag { .map(|node_info| node_info.num_finished) .min() .unwrap(); - absolute_deadlines.retain(|&job, _| job >= oldest_unfinished); + release_times.retain(|&job, _| job >= oldest_unfinished); // A node whose task ended, for example by a panic, stops counting. Flow control keeps // every running node at most `queue_size + 1` jobs behind its upstream node, so an // entry beyond this cap belongs only to such a node. let max_unfinished = graph.node_count() * (Attribute::default().queue_size + 1) + 1; - while absolute_deadlines.len() > max_unfinished { - absolute_deadlines.pop_first(); + while release_times.len() > max_unfinished { + release_times.pop_first(); } } } @@ -611,7 +644,7 @@ impl Dags { let dag = Arc::new(Dag { id, graph: Mutex::new(graph::Graph::new()), - absolute_deadlines: Mutex::new(BTreeMap::new()), + release_times: Mutex::new(BTreeMap::new()), #[cfg(feature = "perf")] response_info: Mutex::new(ResponseInfo::new()), @@ -1175,17 +1208,11 @@ mod tests { let sink = dag.add_node_with_topic_edges(&["deadline_test_topic".into()], &[]); // A non-source node does not release a job. - assert_eq!(dag.get_or_release_absolute_deadline(sink, true, || 5), 5); + assert_eq!(dag.get_or_release_time(sink, true, || 5), 5); // The source node releases job 0. A later wake in job 0 keeps the deadline. - assert_eq!( - dag.get_or_release_absolute_deadline(source, true, || 10), - 10 - ); - assert_eq!( - dag.get_or_release_absolute_deadline(source, true, || 11), - 10 - ); + assert_eq!(dag.get_or_release_time(source, true, || 10), 10); + assert_eq!(dag.get_or_release_time(source, true, || 11), 10); // The input of job 0 can reach the sink node without a wake. assert!(dag.has_stale_deadline(sink)); @@ -1193,28 +1220,19 @@ mod tests { dag.finish_job(source); // A preempted source node resumes before the next period and does not release job 1. - assert_eq!( - dag.get_or_release_absolute_deadline(source, false, || 15), - 15 - ); - assert_eq!( - dag.get_or_release_absolute_deadline(source, true, || 20), - 20 - ); + assert_eq!(dag.get_or_release_time(source, false, || 15), 15); + assert_eq!(dag.get_or_release_time(source, true, || 20), 20); // The sink node wakes for job 0 after the release of job 1. - assert_eq!(dag.get_or_release_absolute_deadline(sink, true, || 21), 10); + assert_eq!(dag.get_or_release_time(sink, true, || 21), 10); assert!(!dag.has_stale_deadline(sink)); // The source node released job 1. dag.finish_job(sink); assert!(dag.has_stale_deadline(sink)); - assert_eq!(dag.get_or_release_absolute_deadline(sink, true, || 22), 20); + assert_eq!(dag.get_or_release_time(sink, true, || 22), 20); assert!(!dag.has_stale_deadline(sink)); - assert!(!dag - .absolute_deadlines - .lock(&mut MCSNode::new()) - .contains_key(&0)); + assert!(!dag.release_times.lock(&mut MCSNode::new()).contains_key(&0)); } #[test] @@ -1226,42 +1244,22 @@ mod tests { let sink = dag.add_node_with_topic_edges(&["retention_test_b".into()], &[]); // The source node releases jobs 0, 1, and 2. - assert_eq!( - dag.get_or_release_absolute_deadline(source, true, || 10), - 10 - ); + assert_eq!(dag.get_or_release_time(source, true, || 10), 10); dag.finish_job(source); - assert_eq!( - dag.get_or_release_absolute_deadline(source, true, || 20), - 20 - ); + assert_eq!(dag.get_or_release_time(source, true, || 20), 20); dag.finish_job(source); - assert_eq!( - dag.get_or_release_absolute_deadline(source, true, || 30), - 30 - ); + assert_eq!(dag.get_or_release_time(source, true, || 30), 30); // The sink node finishes job 0 before the middle node records its finish. dag.finish_job(sink); - assert_eq!( - dag.get_or_release_absolute_deadline(middle, true, || 31), - 10 - ); + assert_eq!(dag.get_or_release_time(middle, true, || 31), 10); dag.finish_job(middle); - assert_eq!( - dag.get_or_release_absolute_deadline(middle, true, || 32), - 20 - ); - assert_eq!(dag.get_or_release_absolute_deadline(sink, true, || 33), 20); + assert_eq!(dag.get_or_release_time(middle, true, || 32), 20); + assert_eq!(dag.get_or_release_time(sink, true, || 33), 20); let mut node = MCSNode::new(); - let jobs: Vec = dag - .absolute_deadlines - .lock(&mut node) - .keys() - .copied() - .collect(); + let jobs: Vec = dag.release_times.lock(&mut node).keys().copied().collect(); assert_eq!(jobs, [1, 2]); } @@ -1273,17 +1271,17 @@ mod tests { // The sink node stopped, for example by a panic, and the source node keeps releasing. for deadline in 0..100 { - dag.get_or_release_absolute_deadline(source, true, || deadline); + dag.get_or_release_time(source, true, || deadline); dag.finish_job(source); } let mut node = MCSNode::new(); - let absolute_deadlines = dag.absolute_deadlines.lock(&mut node); + let release_times = dag.release_times.lock(&mut node); // The cap for two nodes. assert_eq!( - absolute_deadlines.len(), + release_times.len(), 2 * (Attribute::default().queue_size + 1) + 1 ); - assert_eq!(absolute_deadlines.last_key_value(), Some((&99, &99))); + assert_eq!(release_times.last_key_value(), Some((&99, &99))); } } diff --git a/awkernel_async_lib/src/scheduler/gedf.rs b/awkernel_async_lib/src/scheduler/gedf.rs index 2666d4318..65c3cc2d1 100644 --- a/awkernel_async_lib/src/scheduler/gedf.rs +++ b/awkernel_async_lib/src/scheduler/gedf.rs @@ -217,14 +217,14 @@ pub fn calculate_and_update_dag_deadline(dag_info: &DagInfo, wake_time: u64, sta // source node finishes a job, it waits only for that tick, because it has no subscriber. let release = state != State::Preempted; - dag.get_or_release_absolute_deadline(to_node_index(dag_info.node_id), release, || { - // Microseconds, the unit of `awkernel_lib::delay::uptime()`. - let relative_deadline_us = dag - .get_sink_relative_deadline() - .map(|deadline| deadline.as_micros() as u64) - .unwrap_or_else(|| { - panic!("GEDF scheduler: DAG {dag_id} has no sink relative deadline set") - }); - wake_time + relative_deadline_us - }) + // Microseconds, the unit of `awkernel_lib::delay::uptime()`. + let relative_deadline_us = dag + .get_sink_relative_deadline() + .map(|deadline| deadline.as_micros() as u64) + .unwrap_or_else(|_| { + panic!("GEDF scheduler: DAG {dag_id} has no sink relative deadline set") + }); + + dag.get_or_release_time(to_node_index(dag_info.node_id), release, || wake_time) + + relative_deadline_us } From 7052a917c8ab5298fff766b48902cb0e94b6a0b4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rom=C3=A1n=20C=C3=A1rdenas=20Rodr=C3=ADguez?= Date: Mon, 19 Oct 2026 18:50:06 +0900 Subject: [PATCH 2/4] feat(dag): add per-node WCET; compute scheduling feasibility --- applications/tests/test_dag/src/lib.rs | 36 ++-- awkernel_async_lib/src/dag.rs | 264 +++++++++++++++++++++++-- 2 files changed, 266 insertions(+), 34 deletions(-) diff --git a/applications/tests/test_dag/src/lib.rs b/applications/tests/test_dag/src/lib.rs index a7330457f..ad6e66b01 100644 --- a/applications/tests/test_dag/src/lib.rs +++ b/applications/tests/test_dag/src/lib.rs @@ -12,28 +12,32 @@ const LOG_ENABLE: bool = true; pub async fn run() { wait_microsec(1000000); - let period = Duration::from_nanos(1000000000); + let period = Duration::from_nanos(1000000000); // 1 second let exe_time = (period.as_micros() as f64 * 0.1) as u64; log::debug!("period is {} [ns]", period.as_nanos()); let dag = create_dag(); - dag.register_periodic_reactor::<_, (i32,)>( - "reactor_source_node".into(), - move || -> (i32,) { - wait_microsec(exe_time); - let number: i32 = 1; - if LOG_ENABLE { - log::debug!("value={number} in reactor_source_node"); - } - (number,) - }, - vec![Cow::from("topic0")], - SchedulerType::GEDF(5), - period, - ) - .await; + let source = dag + .register_periodic_reactor::<_, (i32,)>( + "reactor_source_node".into(), + move || -> (i32,) { + wait_microsec(exe_time); + let number: i32 = 1; + if LOG_ENABLE { + log::debug!("value={number} in reactor_source_node"); + } + (number,) + }, + vec![Cow::from("topic0")], + SchedulerType::GEDF(5), + period, + ) + .await; + // Nodes without an estimated WCET default to zero. + dag.set_node_wcet(source, Duration::from_micros(exe_time)) + .unwrap(); dag.register_reactor::<_, (i32,), (i32, i32)>( "reactor_node1".into(), diff --git a/awkernel_async_lib/src/dag.rs b/awkernel_async_lib/src/dag.rs index 4d8ee27f4..5807db46c 100644 --- a/awkernel_async_lib/src/dag.rs +++ b/awkernel_async_lib/src/dag.rs @@ -12,6 +12,7 @@ //! - `Dag`: The central struct representing an individual task graph. //! - `create_dag()`: The entry point for creating a new DAG. //! - `register_xx_reactor()`: Methods to register different types of reactors. +//! - `set_node_wcet()`: Method to set the estimated Worst-Case Execution Time (WCET) for a node. //! - `finish_create_dags()`: The function to validate the defined DAGs and start their execution. //! - `DagError`: Defines potential errors that can occur during DAG validation. //! @@ -35,6 +36,19 @@ //! - **No Dangling Edges**: Every published topic must be subscribed to by another node, and every subscribed topic must have a publisher. //! - **Globally Unique Topics (Inter-DAG)**: A topic name must be unique across all active DAGs in the system. This prevents different data pipelines from unintentionally interfering with each other. //! +//! # Estimated WCET and Node Deadlines +//! +//! Each node has an estimated WCET, which is zero unless set with `set_node_wcet()`. +//! The `register_xx_reactor()` methods return the `NodeIndex` of the new node for this purpose. +//! The `finish_create_dags()` method computes the latest finish of each node as an offset from +//! the job release, and checks that the WCETs fit. The latest finish of the sink is its relative deadline `D`, +//! and that of any other node is the minimum over its successors of `latest finish - WCET`. +//! +//! The schedulers do not use the latest finish yet. GEDF and ClusteredEDF ignore the WCETs and give every node of a job the deadline `release time + D`. +//! +//! - **Feasible WCETs**: The critical path, which is the sum of the WCETs along the longest path, must not exceed `D`. +//! Otherwise `finish_create_dags` returns `DagError::InfeasibleWcet`. +//! //! # Example Usage //! //! See the `applications/tests/test_dag` example for practical usage. @@ -54,19 +68,16 @@ mod visit; mod performance; use crate::{ - dag::{ - graph::{ - algo::{connected_components, is_cyclic_directed}, - direction::Direction, - NodeIndex, - }, - visit::{EdgeRef, IntoNodeReferences, NodeRef}, - }, - scheduler::SchedulerType, - task::DagInfo, - time_interval::interval, - Attribute, MultipleReceiver, MultipleSender, VectorToPublishers, VectorToSubscribers, + scheduler::SchedulerType, task::DagInfo, time_interval::interval, Attribute, MultipleReceiver, + MultipleSender, VectorToPublishers, VectorToSubscribers, +}; +use graph::{ + algo::{connected_components, is_cyclic_directed}, + direction::Direction, }; +use visit::{EdgeRef, IntoNodeReferences, NodeRef}; + +pub use graph::NodeIndex; #[cfg(feature = "period-index-propagation")] use crate::task::perf::{ @@ -136,6 +147,8 @@ pub enum DagError { DuplicatePublish(u32, usize), TopicHasMultiplePublishers(u32, Cow<'static, str>), InterDagTopicConflict(Cow<'static, str>, Vec), + WcetOverflow(u32), + InfeasibleWcet(u32, usize), } #[rustfmt::skip] @@ -173,6 +186,10 @@ impl core::fmt::Display for DagError { DagError::InterDagTopicConflict(topic_name, dag_ids) => { write!(f, "Topic '{topic_name}' is used in multiple DAGs. Conflicting DAG IDs: {dag_ids:?}") } + DagError::WcetOverflow(dag_id) => write!(f, "DAG#{dag_id}: estimated WCET computation overflowed"), + DagError::InfeasibleWcet(dag_id, node_id) => { + write!(f, "DAG#{dag_id} Node#{node_id}: estimated WCETs exceed the sink relative deadline") + } } } } @@ -180,6 +197,11 @@ impl core::fmt::Display for DagError { type DagGraph = graph::Graph; impl DagGraph { + #[inline] + fn get_node_wcet(&self, node_idx: NodeIndex) -> Option { + self.node_weight(node_idx).map(|info| info.wcet) + } + #[inline] fn get_sink_relative_deadline(&self) -> Option { let sink_node_index = self.externals(Direction::Outgoing).next()?; @@ -196,6 +218,42 @@ impl DagGraph { None => false, } } + + /// Returns the nodes in topological order: every node comes after all its predecessors. + /// It uses Kahn's algorithm and runs in O(|V| + |E|) time. + /// + /// On a graph with a cycle, the nodes of the cycle, and the nodes reachable only through + /// it, are missing from the result. Check `is_cyclic_directed` first, or compare the length + /// with `node_count()`. + fn topological_order(&self) -> Vec { + let mut in_degree: BTreeMap = self + .node_indices() + .map(|idx| { + ( + idx, + self.neighbors_directed(idx, Direction::Incoming).count(), + ) + }) + .collect(); + let mut ready: Vec = in_degree + .iter() + .filter(|(_, degree)| **degree == 0) + .map(|(idx, _)| *idx) + .collect(); + let mut order: Vec = Vec::with_capacity(in_degree.len()); + while let Some(idx) = ready.pop() { + order.push(idx); + for succ in self.neighbors_directed(idx, Direction::Outgoing) { + if let Some(degree) = in_degree.get_mut(&succ) { + *degree = degree.saturating_sub(1); + if *degree == 0 { + ready.push(succ); + } + } + } + } + order + } } pub struct Dag { @@ -204,6 +262,7 @@ pub struct Dag { /// Release time of each released job that a node has not finished, keyed by job number. /// A job is one run of the DAG. It starts when the source node wakes for a new release. + /// The absolute deadline of a node is the release time plus its `relative_deadline`. release_times: Mutex>, #[cfg(feature = "perf")] @@ -249,6 +308,28 @@ impl Dag { } } + pub fn get_node_wcet(&self, node_idx: NodeIndex) -> Result { + let mut node = MCSNode::new(); + let graph = self.graph.lock(&mut node); + graph + .get_node_wcet(node_idx) + .ok_or(DagError::NodeNotFound(self.id, node_idx.index())) + } + + /// Sets the estimated WCET of a node. It is zero unless set. + /// Call it before `finish_create_dags`, which derives the node deadlines from it. + pub fn set_node_wcet(&self, node_idx: NodeIndex, wcet: Duration) -> Result<(), DagError> { + let mut node = MCSNode::new(); + let mut graph = self.graph.lock(&mut node); + match graph.node_weight_mut(node_idx) { + Some(info) => { + info.wcet = wcet; + Ok(()) + } + None => Err(DagError::NodeNotFound(self.id, node_idx.index())), + } + } + fn add_node_with_topic_edges( &self, subscribe_topic_names: &[Cow<'static, str>], @@ -259,6 +340,7 @@ impl Dag { subscribe_topic_names: subscribe_topic_names.to_vec(), publish_topic_names: publish_topic_names.to_vec(), relative_deadline: None, + wcet: Duration::ZERO, // Temporary WCET num_finished: 0, deadline_job: None, }; @@ -322,7 +404,8 @@ impl Dag { subscribe_topic_names: Vec>, publish_topic_names: Vec>, sched_type: SchedulerType, - ) where + ) -> NodeIndex + where F: Fn( ::Item, ) -> ::Item @@ -361,6 +444,8 @@ impl Dag { .await }) })); + + node_idx } pub async fn register_periodic_reactor( @@ -370,7 +455,8 @@ impl Dag { publish_topic_names: Vec>, sched_type: SchedulerType, period: Duration, - ) where + ) -> NodeIndex + where F: Fn() -> ::Item + Send + 'static, Ret: VectorToPublishers, Ret::Publishers: Send, @@ -413,6 +499,8 @@ impl Dag { }) }), ); + + node_idx } pub async fn register_sink_reactor( @@ -422,7 +510,8 @@ impl Dag { subscribe_topic_names: Vec>, sched_type: SchedulerType, relative_deadline: Duration, - ) where + ) -> NodeIndex + where F: Fn(::Item) + Send + 'static, Args: VectorToSubscribers, Args::Subscribers: Send, @@ -471,6 +560,8 @@ impl Dag { .await }) })); + + node_idx } fn check_subscribe_mismatch( @@ -609,9 +700,10 @@ struct NodeInfo { task_id: u32, subscribe_topic_names: Vec>, publish_topic_names: Vec>, - relative_deadline: Option, - num_finished: u64, // Number of DAG jobs that this node finished. - deadline_job: Option, // Job whose deadline the last wake of this node got. + relative_deadline: Option, // Sink node's relative deadline; None for other nodes until computed. + wcet: Duration, // Declared estimate; zero unless set. + num_finished: u64, // Number of DAG jobs that this node finished. + deadline_job: Option, // Job whose deadline the last wake of this node got. } pub fn to_node_index(index: u32) -> NodeIndex { @@ -725,6 +817,8 @@ fn validate_all_rules(dags: &[Arc]) -> Result<(), Vec> { errors.extend(pubsub_duplicate_errors); } else if let Err(duplicate_error) = validate_single_publisher_per_topic(dag) { errors.extend(duplicate_error); + } else if let Err(timing_error) = compute_timing(dag) { + errors.push(timing_error); } } @@ -950,6 +1044,82 @@ fn validate_single_publisher_per_topic(dag: &Dag) -> Result<(), Vec> { } } +/// Computes the relative deadline of every node, as an offset from the job release, from the +/// estimated WCETs and the sink relative deadline `D`. The earliest starts are computed only to +/// check feasibility. +/// +/// - `ES(source) = 0`, and `ES(v) = max over predecessors p of (ES(p) + C(p))`. +/// - `deadline(sink) = D`, and `deadline(v) = min over successors s of (deadline(s) - C(s))`. +/// +/// Fails if some node cannot finish by its latest finish, that is, if the critical path +/// exceeds `D`. It requires an acyclic DAG, which `validate_dag` ensures. +fn compute_timing(dag: &Dag) -> Result<(), DagError> { + let dag_id = dag.id; + let mut node = MCSNode::new(); + let mut graph = dag.graph.lock(&mut node); + + let sink_deadline = graph + .get_sink_relative_deadline() + .ok_or(DagError::MissingSinkDeadline(dag_id))?; + + let order = graph.topological_order(); + + // Forward pass. + let mut earliest_start: BTreeMap = BTreeMap::new(); + for &idx in &order { + let mut start = Duration::ZERO; + for pred in graph.neighbors_directed(idx, Direction::Incoming) { + let pred_start = earliest_start.get(&pred).copied().unwrap_or(Duration::ZERO); + let pred_finish = pred_start + .checked_add(graph.get_node_wcet(pred).unwrap_or(Duration::ZERO)) + .ok_or(DagError::WcetOverflow(dag_id))?; + start = start.max(pred_finish); + } + earliest_start.insert(idx, start); + } + + // Backward pass. + let mut latest_finish: BTreeMap = BTreeMap::new(); + for &idx in order.iter().rev() { + let mut finish = sink_deadline; + let mut has_successor = false; + for succ in graph.neighbors_directed(idx, Direction::Outgoing) { + let succ_finish = latest_finish.get(&succ).copied().unwrap_or(sink_deadline); + let succ_start = succ_finish + .checked_sub(graph.get_node_wcet(succ).unwrap_or(Duration::ZERO)) + .ok_or(DagError::InfeasibleWcet(dag_id, succ.index()))?; + finish = if has_successor { + finish.min(succ_start) + } else { + succ_start + }; + has_successor = true; + } + latest_finish.insert(idx, finish); + } + + // Every node must be able to finish by its latest finish. + for &idx in &order { + let start = earliest_start.get(&idx).copied().unwrap_or(Duration::ZERO); + let finish = latest_finish.get(&idx).copied().unwrap_or(sink_deadline); + let earliest_finish = start + .checked_add(graph.get_node_wcet(idx).unwrap_or(Duration::ZERO)) + .ok_or(DagError::WcetOverflow(dag_id))?; + if earliest_finish > finish { + return Err(DagError::InfeasibleWcet(dag_id, idx.index())); + } + } + + for &idx in &order { + if let Some(info) = graph.node_weight_mut(idx) { + info.relative_deadline = + Some(latest_finish.get(&idx).copied().unwrap_or(sink_deadline)); + } + } + + Ok(()) +} + async fn spawn_dag(dag: &Arc) { let dag_id = dag.id; let pending_tasks = { @@ -1284,4 +1454,62 @@ mod tests { ); assert_eq!(release_times.last_key_value(), Some((&99, &99))); } + + fn latest_finish_of(dag: &Dag, idx: NodeIndex) -> Duration { + let mut node = MCSNode::new(); + let graph = dag.graph.lock(&mut node); + graph.node_weight(idx).unwrap().relative_deadline.unwrap() + } + + // source -> a -> sink, and source -> b -> sink. + fn diamond(prefix: &str) -> (Arc, [NodeIndex; 4]) { + let dag = create_dag(); + let t = |s: &str| -> Cow<'static, str> { alloc::format!("{prefix}_{s}").into() }; + let source = dag.add_node_with_topic_edges(&[], &[t("sa"), t("sb")]); + let a = dag.add_node_with_topic_edges(&[t("sa")], &[t("as")]); + let b = dag.add_node_with_topic_edges(&[t("sb")], &[t("bs")]); + let sink = dag.add_node_with_topic_edges(&[t("as"), t("bs")], &[]); + dag.set_relative_deadline(sink, Duration::from_millis(100)) + .unwrap(); + (dag, [source, a, b, sink]) + } + + #[test] + fn zero_wcets_give_the_sink_deadline_to_every_node() { + let (dag, nodes) = diamond("timing_zero"); + assert!(compute_timing(&dag).is_ok()); + for idx in nodes { + assert_eq!(latest_finish_of(&dag, idx), Duration::from_millis(100)); + } + } + + #[test] + fn latest_finish_follows_the_downstream_wcets() { + let (dag, [source, a, b, sink]) = diamond("timing_diamond"); + dag.set_node_wcet(source, Duration::from_millis(10)) + .unwrap(); + dag.set_node_wcet(a, Duration::from_millis(30)).unwrap(); + dag.set_node_wcet(b, Duration::from_millis(5)).unwrap(); + dag.set_node_wcet(sink, Duration::from_millis(20)).unwrap(); + assert!(compute_timing(&dag).is_ok()); + + assert_eq!(latest_finish_of(&dag, sink), Duration::from_millis(100)); + assert_eq!(latest_finish_of(&dag, a), Duration::from_millis(80)); + assert_eq!(latest_finish_of(&dag, b), Duration::from_millis(80)); + // The slower branch `a` bounds the source. + assert_eq!(latest_finish_of(&dag, source), Duration::from_millis(50)); + } + + #[test] + fn critical_path_over_the_deadline_is_rejected() { + let (dag, [source, a, _b, sink]) = diamond("timing_infeasible"); + dag.set_node_wcet(source, Duration::from_millis(50)) + .unwrap(); + dag.set_node_wcet(a, Duration::from_millis(40)).unwrap(); + dag.set_node_wcet(sink, Duration::from_millis(20)).unwrap(); + assert!(matches!( + compute_timing(&dag), + Err(DagError::InfeasibleWcet(..)) + )); + } } From 4d3e67cde98002b38ab8cc23f6e40c6bf184eb7e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rom=C3=A1n=20C=C3=A1rdenas=20Rodr=C3=ADguez?= Date: Mon, 19 Oct 2026 18:56:08 +0900 Subject: [PATCH 3/4] feat(dag): add REDF scheduler for DAGs; generic dag_edf for reusing code --- awkernel_async_lib/src/dag.rs | 118 ++++++--- awkernel_async_lib/src/scheduler.rs | 17 +- awkernel_async_lib/src/scheduler/dag_edf.rs | 263 ++++++++++++++++++++ awkernel_async_lib/src/scheduler/gedf.rs | 238 +++--------------- awkernel_async_lib/src/scheduler/redf.rs | 33 +++ 5 files changed, 431 insertions(+), 238 deletions(-) create mode 100644 awkernel_async_lib/src/scheduler/dag_edf.rs create mode 100644 awkernel_async_lib/src/scheduler/redf.rs diff --git a/awkernel_async_lib/src/dag.rs b/awkernel_async_lib/src/dag.rs index 5807db46c..c0cac116d 100644 --- a/awkernel_async_lib/src/dag.rs +++ b/awkernel_async_lib/src/dag.rs @@ -44,7 +44,8 @@ //! the job release, and checks that the WCETs fit. The latest finish of the sink is its relative deadline `D`, //! and that of any other node is the minimum over its successors of `latest finish - WCET`. //! -//! The schedulers do not use the latest finish yet. GEDF and ClusteredEDF ignore the WCETs and give every node of a job the deadline `release time + D`. +//! - **`SchedulerType::REDF`** gives a node the deadline `release time + latest finish`. +//! - **GEDF and ClusteredEDF** ignore the WCETs and give every node of a job the deadline `release time + D`. //! //! - **Feasible WCETs**: The critical path, which is the sum of the WCETs along the longest path, must not exceed `D`. //! Otherwise `finish_create_dags` returns `DagError::InfeasibleWcet`. @@ -600,19 +601,52 @@ impl Dag { } } + /// Returns the absolute deadline of the job that the node at `node_idx` processes next. + /// It is the release time of the job plus the `relative_deadline` of the node. + /// See `get_or_release_job` for the release. + /// + /// # Panics + /// + /// Panics if the node has no relative deadline, which `compute_timing` sets. + pub(crate) fn get_or_release_absolute_deadline( + &self, + node_idx: NodeIndex, + release: bool, + release_time: impl FnOnce() -> u64, + ) -> u64 { + let (release_time, relative_deadline) = + self.get_or_release_job(node_idx, release, release_time); + // `compute_timing` sets it for every node when the DAG is built. + let relative_deadline = relative_deadline + .unwrap_or_else(|| panic!("DAG {}: node has no relative deadline", self.id)); + release_time.saturating_add(relative_deadline) + } + /// Returns the release time of the job that the node at `node_idx` processes next. - /// If that job is not released and `release` is true, the source node releases it with - /// the value of `release_time`. Any other wake gets that value without a release. - pub(crate) fn get_or_release_time( + /// See `get_or_release_job` for the release. + pub(crate) fn get_or_release_release_time( &self, node_idx: NodeIndex, release: bool, release_time: impl FnOnce() -> u64, ) -> u64 { + self.get_or_release_job(node_idx, release, release_time).0 + } + + /// Returns the release time of the job that the node at `node_idx` processes next and the + /// `relative_deadline` of the node in microseconds, if it is set. + /// If that job is not released and `release` is true, the source node releases it with + /// the value of `release_time`. Any other wake uses that value without a release. + fn get_or_release_job( + &self, + node_idx: NodeIndex, + release: bool, + release_time: impl FnOnce() -> u64, + ) -> (u64, Option) { let mut node = MCSNode::new(); let mut release_times = self.release_times.lock(&mut node); - let (job, release) = { + let (job, release, relative_deadline) = { let mut graph_node = MCSNode::new(); let mut graph = self.graph.lock(&mut graph_node); let is_source = graph @@ -623,18 +657,25 @@ impl Dag { let job = node_info.num_finished; let release = release && is_source; node_info.deadline_job = (release || release_times.contains_key(&job)).then_some(job); - (job, release) + // Microseconds, the unit of `awkernel_lib::delay::uptime()`. + let relative_deadline = node_info + .relative_deadline + .map(|deadline| u64::try_from(deadline.as_micros()).unwrap_or(u64::MAX)); + (job, release, relative_deadline) }; - if let Some(&released) = release_times.get(&job) { - return released; - } + let release_time = match release_times.get(&job) { + Some(&released) => released, + None => { + let new_release_time = release_time(); + if release { + release_times.insert(job, new_release_time); + } + new_release_time + } + }; - let release_time = release_time(); - if release { - release_times.insert(job, release_time); - } - release_time + (release_time, relative_deadline) } /// Returns true if the job that the node at `node_idx` processes is released, but the @@ -1378,11 +1419,11 @@ mod tests { let sink = dag.add_node_with_topic_edges(&["deadline_test_topic".into()], &[]); // A non-source node does not release a job. - assert_eq!(dag.get_or_release_time(sink, true, || 5), 5); + assert_eq!(dag.get_or_release_release_time(sink, true, || 5), 5); // The source node releases job 0. A later wake in job 0 keeps the deadline. - assert_eq!(dag.get_or_release_time(source, true, || 10), 10); - assert_eq!(dag.get_or_release_time(source, true, || 11), 10); + assert_eq!(dag.get_or_release_release_time(source, true, || 10), 10); + assert_eq!(dag.get_or_release_release_time(source, true, || 11), 10); // The input of job 0 can reach the sink node without a wake. assert!(dag.has_stale_deadline(sink)); @@ -1390,17 +1431,17 @@ mod tests { dag.finish_job(source); // A preempted source node resumes before the next period and does not release job 1. - assert_eq!(dag.get_or_release_time(source, false, || 15), 15); - assert_eq!(dag.get_or_release_time(source, true, || 20), 20); + assert_eq!(dag.get_or_release_release_time(source, false, || 15), 15); + assert_eq!(dag.get_or_release_release_time(source, true, || 20), 20); // The sink node wakes for job 0 after the release of job 1. - assert_eq!(dag.get_or_release_time(sink, true, || 21), 10); + assert_eq!(dag.get_or_release_release_time(sink, true, || 21), 10); assert!(!dag.has_stale_deadline(sink)); // The source node released job 1. dag.finish_job(sink); assert!(dag.has_stale_deadline(sink)); - assert_eq!(dag.get_or_release_time(sink, true, || 22), 20); + assert_eq!(dag.get_or_release_release_time(sink, true, || 22), 20); assert!(!dag.has_stale_deadline(sink)); assert!(!dag.release_times.lock(&mut MCSNode::new()).contains_key(&0)); } @@ -1414,19 +1455,19 @@ mod tests { let sink = dag.add_node_with_topic_edges(&["retention_test_b".into()], &[]); // The source node releases jobs 0, 1, and 2. - assert_eq!(dag.get_or_release_time(source, true, || 10), 10); + assert_eq!(dag.get_or_release_release_time(source, true, || 10), 10); dag.finish_job(source); - assert_eq!(dag.get_or_release_time(source, true, || 20), 20); + assert_eq!(dag.get_or_release_release_time(source, true, || 20), 20); dag.finish_job(source); - assert_eq!(dag.get_or_release_time(source, true, || 30), 30); + assert_eq!(dag.get_or_release_release_time(source, true, || 30), 30); // The sink node finishes job 0 before the middle node records its finish. dag.finish_job(sink); - assert_eq!(dag.get_or_release_time(middle, true, || 31), 10); + assert_eq!(dag.get_or_release_release_time(middle, true, || 31), 10); dag.finish_job(middle); - assert_eq!(dag.get_or_release_time(middle, true, || 32), 20); - assert_eq!(dag.get_or_release_time(sink, true, || 33), 20); + assert_eq!(dag.get_or_release_release_time(middle, true, || 32), 20); + assert_eq!(dag.get_or_release_release_time(sink, true, || 33), 20); let mut node = MCSNode::new(); let jobs: Vec = dag.release_times.lock(&mut node).keys().copied().collect(); @@ -1441,7 +1482,7 @@ mod tests { // The sink node stopped, for example by a panic, and the source node keeps releasing. for deadline in 0..100 { - dag.get_or_release_time(source, true, || deadline); + dag.get_or_release_release_time(source, true, || deadline); dag.finish_job(source); } @@ -1512,4 +1553,25 @@ mod tests { Err(DagError::InfeasibleWcet(..)) )); } + + #[test] + fn node_deadline_is_release_plus_relative_deadline() { + let (dag, [source, a, _b, sink]) = diamond("timing_deadline"); + + dag.set_node_wcet(sink, Duration::from_millis(20)).unwrap(); // Latest start: 100 - 20 = 80 + dag.set_node_wcet(a, Duration::from_millis(30)).unwrap(); // Latest start: 80 - 30 = 50 + dag.set_node_wcet(source, Duration::from_millis(10)) // Latest start: 50 - 10 = 40 + .unwrap(); + + assert!(compute_timing(&dag).is_ok()); + + assert_eq!( + dag.get_or_release_absolute_deadline(source, true, || 1_000), // job release time is 1_000 + 1_000 + 50_000 // release time + latest finish of sink (50) + ); + assert_eq!( + dag.get_or_release_absolute_deadline(sink, true, || 2_000), + 1_000 + 100_000 // release time + latest finish of sink (100) + ); + } } diff --git a/awkernel_async_lib/src/scheduler.rs b/awkernel_async_lib/src/scheduler.rs index 4d82e6b85..5bd953db9 100644 --- a/awkernel_async_lib/src/scheduler.rs +++ b/awkernel_async_lib/src/scheduler.rs @@ -20,10 +20,12 @@ use awkernel_lib::{ use alloc::boxed::Box; mod clustered_edf; +mod dag_edf; pub mod gedf; pub(super) mod panicked; mod prioritized_fifo; mod prioritized_rr; +mod redf; static SLEEPING: Mutex = Mutex::new(SleepingTasks::new()); @@ -85,6 +87,12 @@ pub enum SchedulerType { /// warning. ClusteredEDF(u64, CpuSet), GEDF(u64), // relative deadline + /// Relative Earliest Deadline First: `relative_deadline`. + /// + /// Like GEDF, but a task of a DAG node gets the deadline of its job release plus the + /// latest finish of the node, which is derived from the estimated WCETs of the DAG. + /// `relative_deadline` is used only by tasks that do not belong to a DAG. + REDF(u64), PrioritizedFIFO(u8), PrioritizedRR(u8), Panicked, @@ -95,6 +103,7 @@ impl SchedulerType { matches!( (self, other), (SchedulerType::GEDF(_), SchedulerType::GEDF(_)) + | (SchedulerType::REDF(_), SchedulerType::REDF(_)) | ( SchedulerType::ClusteredEDF(_, _), SchedulerType::ClusteredEDF(_, _) @@ -140,14 +149,17 @@ impl SchedulerType { /// - The second highest priority. /// - GEDF scheduler. /// - The third highest priority. -/// - Prioritized FIFO scheduler. +/// - REDF scheduler. /// - The fourth highest priority. +/// - Prioritized FIFO scheduler. +/// - The fifth highest priority. /// - Prioritized Round-Robin scheduler. /// - The lowest priority. /// - Panicked scheduler. -static PRIORITY_LIST: [SchedulerType; 5] = [ +static PRIORITY_LIST: [SchedulerType; 6] = [ SchedulerType::ClusteredEDF(0, CpuSet::empty()), SchedulerType::GEDF(0), + SchedulerType::REDF(0), SchedulerType::PrioritizedFIFO(0), SchedulerType::PrioritizedRR(0), SchedulerType::Panicked, @@ -283,6 +295,7 @@ pub(crate) fn get_scheduler(sched_type: &SchedulerType) -> &'static dyn Schedule SchedulerType::PrioritizedFIFO(_) => &prioritized_fifo::SCHEDULER, SchedulerType::PrioritizedRR(_) => &prioritized_rr::SCHEDULER, SchedulerType::GEDF(_) => &gedf::SCHEDULER, + SchedulerType::REDF(_) => &redf::SCHEDULER, SchedulerType::ClusteredEDF(_, _) => &clustered_edf::SCHEDULER, SchedulerType::Panicked => &panicked::SCHEDULER, } diff --git a/awkernel_async_lib/src/scheduler/dag_edf.rs b/awkernel_async_lib/src/scheduler/dag_edf.rs new file mode 100644 index 000000000..fbc54613a --- /dev/null +++ b/awkernel_async_lib/src/scheduler/dag_edf.rs @@ -0,0 +1,263 @@ +//! A generic EDF scheduler for tasks that may belong to a DAG. +//! +//! [`DagEdfScheduler`] holds everything the DAG-aware EDF schedulers share: the run queue ordered +//! by absolute deadline, the wake and dispatch logic, and the preemption. A concrete scheduler, +//! such as GEDF or REDF, is a [`DagEdfPolicy`] that says only how the deadline is computed. +//! +//! To add a scheduler of this family: +//! 1. Add a variant to `SchedulerType`, and register it in `PRIORITY_LIST`, `equals()` and +//! `get_scheduler()` in `scheduler.rs`. +//! 2. Implement [`DagEdfPolicy`] on a marker type, and define +//! `pub static SCHEDULER: DagEdfScheduler = DagEdfScheduler::new();`. + +use core::{cmp::max, marker::PhantomData}; + +use super::{Scheduler, SchedulerType, Task}; +use crate::{ + dag::{get_dag, to_node_index, Dag, NodeIndex}, + scheduler::GLOBAL_WAKE_GET_MUTEX, + scheduler::{ + get_priority, peek_preemption_pending, push_preemption_pending, PREEMPTION_ENABLED, + }, + task::{ + get_task, get_tasks_running, set_current_task, set_need_preemption, DagInfo, State, + MAX_TASK_PRIORITY, + }, +}; +use alloc::{collections::BinaryHeap, sync::Arc, vec::Vec}; +use awkernel_lib::sync::mutex::{MCSNode, Mutex}; + +/// The scheduler-specific part of a [`DagEdfScheduler`]. +pub trait DagEdfPolicy: 'static { + /// The scheduler type with a dummy payload. It selects the priority of the scheduler. + const SCHEDULER_TYPE: SchedulerType; + + /// The name of the scheduler for panic messages. + const NAME: &'static str; + + /// Returns the relative deadline in `scheduler_type`, which is the deadline of a task that + /// does not belong to a DAG, or `None` if `scheduler_type` is not of this scheduler. + fn relative_deadline(scheduler_type: &SchedulerType) -> Option; + + /// Returns the absolute deadline of the job that the node `node_idx` of `dag` processes + /// next. `release` is true if this wake may release that job. The policy gets the release + /// time of the job with `Dag::get_or_release_release_time` or + /// `Dag::get_or_release_absolute_deadline`, and passes `wake_time` as the release time. + fn dag_deadline(dag: &Dag, node_idx: NodeIndex, release: bool, wake_time: u64) -> u64; +} + +/// Returns the absolute deadline of the DAG job that the node of `dag_info` processes next, +/// according to the policy `P`. +/// +/// A wake of the source node in any `state` except `State::Preempted` may release that job. +/// +/// # Panics +/// +/// Panics if the DAG does not exist, or if the policy panics. +pub fn calculate_dag_deadline( + dag_info: &DagInfo, + wake_time: u64, + state: State, +) -> u64 { + let dag_id = dag_info.dag_id; + let dag = + get_dag(dag_id).unwrap_or_else(|| panic!("{} scheduler: DAG {dag_id} not found", P::NAME)); + + // The wake of a preempted task resumes it and does not release a job. Any other wake + // of the source node at an unreleased job must come from its `Interval` tick. After the + // source node finishes a job, it waits only for that tick, because it has no subscriber. + let release = state != State::Preempted; + + P::dag_deadline(&dag, to_node_index(dag_info.node_id), release, wake_time) +} + +pub struct DagEdfScheduler { + data: Mutex>, // Run queue. + priority: u8, + policy: PhantomData P>, +} + +struct DagEdfTask { + task: Arc, + absolute_deadline: u64, + wake_time: u64, +} + +impl PartialOrd for DagEdfTask { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +impl PartialEq for DagEdfTask { + fn eq(&self, other: &Self) -> bool { + self.absolute_deadline == other.absolute_deadline && self.wake_time == other.wake_time + } +} + +impl Ord for DagEdfTask { + fn cmp(&self, other: &Self) -> core::cmp::Ordering { + match other.absolute_deadline.cmp(&self.absolute_deadline) { + core::cmp::Ordering::Equal => other.wake_time.cmp(&self.wake_time), + other => other, + } + } +} + +impl Eq for DagEdfTask {} + +struct DagEdfData { + queue: BinaryHeap, +} + +impl DagEdfData { + fn new() -> Self { + Self { + queue: BinaryHeap::new(), + } + } +} + +impl DagEdfScheduler

{ + pub const fn new() -> Self { + Self { + data: Mutex::new(None), + priority: get_priority(&P::SCHEDULER_TYPE), + policy: PhantomData, + } + } + + fn invoke_preemption(&self, task: Arc) -> bool { + if !PREEMPTION_ENABLED { + return false; + } + + let tasks_running = get_tasks_running() + .into_iter() + .filter(|rt| rt.task_id != 0) // Filter out idle CPUs + .collect::>(); + + // 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 preemption_target = tasks_running + .iter() + .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) + }) + }) + .min() + .unwrap(); + + let (target_task, target_cpu) = preemption_target; + if task.preempts(&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 + } +} + +impl Default for DagEdfScheduler

{ + fn default() -> Self { + Self::new() + } +} + +impl Scheduler for DagEdfScheduler

{ + fn wake_task(&self, task: Arc) { + 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(); + let Some(relative_deadline) = P::relative_deadline(&info.scheduler_type) else { + unreachable!() + }; + + let wake_time = awkernel_lib::delay::uptime(); + let absolute_deadline = if let Some(ref dag_info) = dag_info { + calculate_dag_deadline::

(dag_info, wake_time, info.state) + } 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); + + (wake_time, absolute_deadline) + }; + + let mut node = MCSNode::new(); + let _guard = GLOBAL_WAKE_GET_MUTEX.lock(&mut node); + if !self.invoke_preemption(task.clone()) { + let mut node_inner = MCSNode::new(); + let mut data = self.data.lock(&mut node_inner); + let internal_data = data.get_or_insert_with(DagEdfData::new); + internal_data.queue.push(DagEdfTask { + task: task.clone(), + absolute_deadline, + wake_time, + }); + } + } + + fn get_next(&self, execution_ensured: bool) -> Option> { + let mut node = MCSNode::new(); + let mut data = self.data.lock(&mut node); + + #[allow(clippy::question_mark)] + let data = match data.as_mut() { + Some(data) => data, + None => return None, + }; + + loop { + // Pop a task from the run queue. + let task = data.queue.pop()?; + + // Make the state of the task Running. + { + let mut node = MCSNode::new(); + let mut task_info = task.task.info.lock(&mut node); + + if matches!(task_info.state, State::Terminated | State::Panicked) { + continue; + } + + if task_info.state == State::Preempted { + task_info.need_preemption = false; + } + if execution_ensured { + task_info.state = State::Running; + set_current_task(awkernel_lib::cpu::cpu_id(), task.task.id); + } + } + + return Some(task.task); + } + } + + fn scheduler_name(&self) -> SchedulerType { + P::SCHEDULER_TYPE + } + + fn priority(&self) -> u8 { + self.priority + } +} diff --git a/awkernel_async_lib/src/scheduler/gedf.rs b/awkernel_async_lib/src/scheduler/gedf.rs index 65c3cc2d1..58e78e90c 100644 --- a/awkernel_async_lib/src/scheduler/gedf.rs +++ b/awkernel_async_lib/src/scheduler/gedf.rs @@ -1,230 +1,52 @@ //! A GEDF scheduler. +//! +//! Every node of a DAG job gets the deadline `release time + sink relative deadline`. -use core::cmp::max; - -use super::{Scheduler, SchedulerType, Task}; +use super::{ + dag_edf::{calculate_dag_deadline, DagEdfPolicy, DagEdfScheduler}, + SchedulerType, +}; use crate::{ - dag::{get_dag, to_node_index}, - scheduler::GLOBAL_WAKE_GET_MUTEX, - scheduler::{ - get_priority, peek_preemption_pending, push_preemption_pending, PREEMPTION_ENABLED, - }, - task::{ - get_task, get_tasks_running, set_current_task, set_need_preemption, DagInfo, State, - MAX_TASK_PRIORITY, - }, + dag::{Dag, NodeIndex}, + task::{DagInfo, State}, }; -use alloc::{collections::BinaryHeap, sync::Arc, vec::Vec}; -use awkernel_lib::sync::mutex::{MCSNode, Mutex}; -pub struct GEDFScheduler { - data: Mutex>, // Run queue. - priority: u8, -} +pub struct GedfPolicy; -struct GEDFTask { - task: Arc, - absolute_deadline: u64, - wake_time: u64, -} +pub type GEDFScheduler = DagEdfScheduler; -impl PartialOrd for GEDFTask { - fn partial_cmp(&self, other: &Self) -> Option { - Some(self.cmp(other)) - } -} +pub static SCHEDULER: GEDFScheduler = DagEdfScheduler::new(); -impl PartialEq for GEDFTask { - fn eq(&self, other: &Self) -> bool { - self.absolute_deadline == other.absolute_deadline && self.wake_time == other.wake_time - } -} +impl DagEdfPolicy for GedfPolicy { + const SCHEDULER_TYPE: SchedulerType = SchedulerType::GEDF(0); + const NAME: &'static str = "GEDF"; -impl Ord for GEDFTask { - fn cmp(&self, other: &Self) -> core::cmp::Ordering { - match other.absolute_deadline.cmp(&self.absolute_deadline) { - core::cmp::Ordering::Equal => other.wake_time.cmp(&self.wake_time), - other => other, + fn relative_deadline(scheduler_type: &SchedulerType) -> Option { + match scheduler_type { + SchedulerType::GEDF(relative_deadline) => Some(*relative_deadline), + _ => None, } } -} -impl Eq for GEDFTask {} + fn dag_deadline(dag: &Dag, node_idx: NodeIndex, release: bool, wake_time: u64) -> u64 { + let release_time = dag.get_or_release_release_time(node_idx, release, || wake_time); -struct GEDFData { - queue: BinaryHeap, -} + // GEDF only uses the sink node's relative deadline + let sink_deadline_us = dag + .get_sink_relative_deadline() + .map(|deadline| deadline.as_micros() as u64) + .unwrap_or_else(|_| panic!("GEDF scheduler: DAG has no sink relative deadline set")); -impl GEDFData { - fn new() -> Self { - Self { - queue: BinaryHeap::new(), - } + release_time.saturating_add(sink_deadline_us) } } -impl Scheduler for GEDFScheduler { - fn wake_task(&self, task: Arc) { - 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 = if let Some(ref dag_info) = dag_info { - calculate_and_update_dag_deadline(dag_info, wake_time, info.state) - } 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); - - (wake_time, absolute_deadline) - } - _ => unreachable!(), - } - }; - - let mut node = MCSNode::new(); - let _guard = GLOBAL_WAKE_GET_MUTEX.lock(&mut node); - if !self.invoke_preemption(task.clone()) { - let mut node_inner = MCSNode::new(); - let mut data = self.data.lock(&mut node_inner); - let internal_data = data.get_or_insert_with(GEDFData::new); - internal_data.queue.push(GEDFTask { - task: task.clone(), - absolute_deadline, - wake_time, - }); - } - } - - fn get_next(&self, execution_ensured: bool) -> Option> { - let mut node = MCSNode::new(); - let mut data = self.data.lock(&mut node); - - #[allow(clippy::question_mark)] - let data = match data.as_mut() { - Some(data) => data, - None => return None, - }; - - loop { - // Pop a task from the run queue. - let task = data.queue.pop()?; - - // Make the state of the task Running. - { - let mut node = MCSNode::new(); - let mut task_info = task.task.info.lock(&mut node); - - if matches!(task_info.state, State::Terminated | State::Panicked) { - continue; - } - - if task_info.state == State::Preempted { - task_info.need_preemption = false; - } - if execution_ensured { - task_info.state = State::Running; - set_current_task(awkernel_lib::cpu::cpu_id(), task.task.id); - } - } - - return Some(task.task); - } - } - - fn scheduler_name(&self) -> SchedulerType { - SchedulerType::GEDF(0) - } - - fn priority(&self) -> u8 { - self.priority - } -} - -pub static SCHEDULER: GEDFScheduler = GEDFScheduler { - data: Mutex::new(None), - priority: get_priority(&SchedulerType::GEDF(0)), -}; - -impl GEDFScheduler { - fn invoke_preemption(&self, task: Arc) -> bool { - if !PREEMPTION_ENABLED { - return false; - } - - let tasks_running = get_tasks_running() - .into_iter() - .filter(|rt| rt.task_id != 0) // Filter out idle CPUs - .collect::>(); - - // 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 preemption_target = tasks_running - .iter() - .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) - }) - }) - .min() - .unwrap(); - - let (target_task, target_cpu) = preemption_target; - if task.preempts(&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 - } -} - -/// Returns the absolute deadline of the DAG job that the node of `dag_info` processes next. -/// A wake of the source node in any `state` except `State::Preempted` releases that job -/// with `wake_time` plus the sink relative deadline, if it is not released yet. Any other wake -/// before the release gets that value without a release. +/// Returns the absolute deadline of the DAG job that the node of `dag_info` processes next, +/// with the GEDF rule. /// /// # Panics /// -/// Panics if the DAG does not exist, has no node `dag_info.node_id`, or has no sink relative -/// deadline. +/// Panics if the DAG does not exist or has no sink relative deadline. pub fn calculate_and_update_dag_deadline(dag_info: &DagInfo, wake_time: u64, state: State) -> u64 { - let dag_id = dag_info.dag_id; - let dag = get_dag(dag_id).unwrap_or_else(|| panic!("GEDF scheduler: DAG {dag_id} not found")); - - // The wake of a preempted task resumes it and does not release a job. Any other wake - // of the source node at an unreleased job must come from its `Interval` tick. After the - // source node finishes a job, it waits only for that tick, because it has no subscriber. - let release = state != State::Preempted; - - // Microseconds, the unit of `awkernel_lib::delay::uptime()`. - let relative_deadline_us = dag - .get_sink_relative_deadline() - .map(|deadline| deadline.as_micros() as u64) - .unwrap_or_else(|_| { - panic!("GEDF scheduler: DAG {dag_id} has no sink relative deadline set") - }); - - dag.get_or_release_time(to_node_index(dag_info.node_id), release, || wake_time) - + relative_deadline_us + calculate_dag_deadline::(dag_info, wake_time, state) } diff --git a/awkernel_async_lib/src/scheduler/redf.rs b/awkernel_async_lib/src/scheduler/redf.rs new file mode 100644 index 000000000..a5ee753f1 --- /dev/null +++ b/awkernel_async_lib/src/scheduler/redf.rs @@ -0,0 +1,33 @@ +//! A REDF (Relative Earliest Deadline First) scheduler. +//! +//! Each node of a DAG job gets its own deadline: `release time + node relative deadline`. +//! The relative deadline of a node is its latest finish time, computed from the sink deadline +//! and the estimated WCETs when the DAG is built. + +use super::{ + dag_edf::{DagEdfPolicy, DagEdfScheduler}, + SchedulerType, +}; +use crate::dag::{Dag, NodeIndex}; + +pub struct RedfPolicy; + +pub type REDFScheduler = DagEdfScheduler; + +pub static SCHEDULER: REDFScheduler = DagEdfScheduler::new(); + +impl DagEdfPolicy for RedfPolicy { + const SCHEDULER_TYPE: SchedulerType = SchedulerType::REDF(0); + const NAME: &'static str = "REDF"; + + fn relative_deadline(scheduler_type: &SchedulerType) -> Option { + match scheduler_type { + SchedulerType::REDF(relative_deadline) => Some(*relative_deadline), + _ => None, + } + } + // REDF uses per-node relative deadlines + fn dag_deadline(dag: &Dag, node_idx: NodeIndex, release: bool, wake_time: u64) -> u64 { + dag.get_or_release_absolute_deadline(node_idx, release, || wake_time) + } +} From 6be889d567d43039795f220d17767c84d329ffac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rom=C3=A1n=20C=C3=A1rdenas=20Rodr=C3=ADguez?= Date: Tue, 20 Oct 2026 15:47:00 +0900 Subject: [PATCH 4/4] feat(dag): use Timing struct; include slack and critical path analysis --- awkernel_async_lib/src/dag.rs | 257 ++++++++++++-------- awkernel_async_lib/src/scheduler/dag_edf.rs | 4 +- awkernel_async_lib/src/scheduler/gedf.rs | 5 +- awkernel_async_lib/src/scheduler/redf.rs | 11 +- 4 files changed, 172 insertions(+), 105 deletions(-) diff --git a/awkernel_async_lib/src/dag.rs b/awkernel_async_lib/src/dag.rs index c0cac116d..91c99411d 100644 --- a/awkernel_async_lib/src/dag.rs +++ b/awkernel_async_lib/src/dag.rs @@ -200,7 +200,19 @@ type DagGraph = graph::Graph; impl DagGraph { #[inline] fn get_node_wcet(&self, node_idx: NodeIndex) -> Option { - self.node_weight(node_idx).map(|info| info.wcet) + self.node_weight(node_idx) + .map(|info| info.timing.estimated_wcet) + } + + #[inline] + fn set_node_wcet(&mut self, node_idx: NodeIndex, wcet: Duration) -> bool { + match self.node_weight_mut(node_idx) { + Some(info) => { + info.timing.estimated_wcet = wcet; + true + } + None => false, + } } #[inline] @@ -322,12 +334,9 @@ impl Dag { pub fn set_node_wcet(&self, node_idx: NodeIndex, wcet: Duration) -> Result<(), DagError> { let mut node = MCSNode::new(); let mut graph = self.graph.lock(&mut node); - match graph.node_weight_mut(node_idx) { - Some(info) => { - info.wcet = wcet; - Ok(()) - } - None => Err(DagError::NodeNotFound(self.id, node_idx.index())), + match graph.set_node_wcet(node_idx, wcet) { + true => Ok(()), + false => Err(DagError::NodeNotFound(self.id, node_idx.index())), } } @@ -341,7 +350,7 @@ impl Dag { subscribe_topic_names: subscribe_topic_names.to_vec(), publish_topic_names: publish_topic_names.to_vec(), relative_deadline: None, - wcet: Duration::ZERO, // Temporary WCET + timing: NodeTiming::default(), num_finished: 0, deadline_job: None, }; @@ -601,52 +610,25 @@ impl Dag { } } - /// Returns the absolute deadline of the job that the node at `node_idx` processes next. - /// It is the release time of the job plus the `relative_deadline` of the node. - /// See `get_or_release_job` for the release. - /// - /// # Panics - /// - /// Panics if the node has no relative deadline, which `compute_timing` sets. - pub(crate) fn get_or_release_absolute_deadline( - &self, - node_idx: NodeIndex, - release: bool, - release_time: impl FnOnce() -> u64, - ) -> u64 { - let (release_time, relative_deadline) = - self.get_or_release_job(node_idx, release, release_time); - // `compute_timing` sets it for every node when the DAG is built. - let relative_deadline = relative_deadline - .unwrap_or_else(|| panic!("DAG {}: node has no relative deadline", self.id)); - release_time.saturating_add(relative_deadline) - } - - /// Returns the release time of the job that the node at `node_idx` processes next. - /// See `get_or_release_job` for the release. - pub(crate) fn get_or_release_release_time( - &self, - node_idx: NodeIndex, - release: bool, - release_time: impl FnOnce() -> u64, - ) -> u64 { - self.get_or_release_job(node_idx, release, release_time).0 - } - - /// Returns the release time of the job that the node at `node_idx` processes next and the - /// `relative_deadline` of the node in microseconds, if it is set. + /// Returns the release time of the job that the node at `node_idx` processes next, and the + /// result of `read_timing`, which gets the timing parameters of the node. The caller takes + /// only the parameters it needs, with no copy of the others. /// If that job is not released and `release` is true, the source node releases it with /// the value of `release_time`. Any other wake uses that value without a release. - fn get_or_release_job( + /// + /// `read_timing` runs while the DAG locks are held, so it must be short and must not use + /// the DAG. + pub(crate) fn get_or_release_job( &self, node_idx: NodeIndex, release: bool, release_time: impl FnOnce() -> u64, - ) -> (u64, Option) { + read_timing: impl FnOnce(&NodeTiming) -> R, + ) -> (u64, R) { let mut node = MCSNode::new(); let mut release_times = self.release_times.lock(&mut node); - let (job, release, relative_deadline) = { + let (job, release, read) = { let mut graph_node = MCSNode::new(); let mut graph = self.graph.lock(&mut graph_node); let is_source = graph @@ -657,11 +639,7 @@ impl Dag { let job = node_info.num_finished; let release = release && is_source; node_info.deadline_job = (release || release_times.contains_key(&job)).then_some(job); - // Microseconds, the unit of `awkernel_lib::delay::uptime()`. - let relative_deadline = node_info - .relative_deadline - .map(|deadline| u64::try_from(deadline.as_micros()).unwrap_or(u64::MAX)); - (job, release, relative_deadline) + (job, release, read_timing(&node_info.timing)) }; let release_time = match release_times.get(&job) { @@ -675,7 +653,7 @@ impl Dag { } }; - (release_time, relative_deadline) + (release_time, read) } /// Returns true if the job that the node at `node_idx` processes is released, but the @@ -737,14 +715,42 @@ impl PendingTask { } } +/// The timing parameters of a node. All are offsets from the release of the job. +#[derive(Clone, Copy, Debug, Default)] +pub struct NodeTiming { + /// Estimated worst-case execution time of the node. It is zero unless set with `set_node_wcet()`. + pub estimated_wcet: Duration, + /// Estimated earliest start of the node, computed from the predecessors. + pub estimated_earliest_start: Duration, + /// Estimated latest finish of the node, computed from the successors. + pub estimated_latest_finish: Duration, + /// True if the node is on a critical path, that is, a longest path of the DAG by WCET. + /// Such a node has the least slack, which is `D` minus the length of that path. + pub is_critical: bool, +} + +impl NodeTiming { + pub fn estimated_earliest_finish(&self) -> Duration { + self.estimated_earliest_start + self.estimated_wcet + } + + pub fn estimated_latest_start(&self) -> Duration { + self.estimated_latest_finish - self.estimated_wcet + } + + pub fn slack(&self) -> Duration { + self.estimated_latest_finish - self.estimated_earliest_finish() + } +} + struct NodeInfo { task_id: u32, subscribe_topic_names: Vec>, publish_topic_names: Vec>, - relative_deadline: Option, // Sink node's relative deadline; None for other nodes until computed. - wcet: Duration, // Declared estimate; zero unless set. - num_finished: u64, // Number of DAG jobs that this node finished. - deadline_job: Option, // Job whose deadline the last wake of this node got. + relative_deadline: Option, // Sink node's relative deadline; None for other nodes + timing: NodeTiming, // The WCET is declared, and `compute_timing` computes the rest. + num_finished: u64, // Number of DAG jobs that this node finished. + deadline_job: Option, // Job whose deadline the last wake of this node got. } pub fn to_node_index(index: u32) -> NodeIndex { @@ -1087,7 +1093,7 @@ fn validate_single_publisher_per_topic(dag: &Dag) -> Result<(), Vec> { /// Computes the relative deadline of every node, as an offset from the job release, from the /// estimated WCETs and the sink relative deadline `D`. The earliest starts are computed only to -/// check feasibility. +/// check feasibility and to mark the nodes on a critical path. /// /// - `ES(source) = 0`, and `ES(v) = max over predecessors p of (ES(p) + C(p))`. /// - `deadline(sink) = D`, and `deadline(v) = min over successors s of (deadline(s) - C(s))`. @@ -1105,7 +1111,7 @@ fn compute_timing(dag: &Dag) -> Result<(), DagError> { let order = graph.topological_order(); - // Forward pass. + // Forward pass to compute per-node earliest start times. let mut earliest_start: BTreeMap = BTreeMap::new(); for &idx in &order { let mut start = Duration::ZERO; @@ -1119,7 +1125,7 @@ fn compute_timing(dag: &Dag) -> Result<(), DagError> { earliest_start.insert(idx, start); } - // Backward pass. + // Backward pass to compute per-node latest finish times. let mut latest_finish: BTreeMap = BTreeMap::new(); for &idx in order.iter().rev() { let mut finish = sink_deadline; @@ -1139,23 +1145,34 @@ fn compute_timing(dag: &Dag) -> Result<(), DagError> { latest_finish.insert(idx, finish); } - // Every node must be able to finish by its latest finish. + // Check feasibility, and find the slack of every node. + let mut slacks: BTreeMap = BTreeMap::new(); for &idx in &order { - let start = earliest_start.get(&idx).copied().unwrap_or(Duration::ZERO); - let finish = latest_finish.get(&idx).copied().unwrap_or(sink_deadline); - let earliest_finish = start - .checked_add(graph.get_node_wcet(idx).unwrap_or(Duration::ZERO)) + let wcet = graph.get_node_wcet(idx).unwrap_or(Duration::ZERO); + let earliest_start = earliest_start.get(&idx).copied().unwrap_or(Duration::ZERO); + let latest_finish = latest_finish.get(&idx).copied().unwrap_or(sink_deadline); + + let earliest_finish = earliest_start + .checked_add(wcet) .ok_or(DagError::WcetOverflow(dag_id))?; - if earliest_finish > finish { - return Err(DagError::InfeasibleWcet(dag_id, idx.index())); - } + let slack = latest_finish + .checked_sub(earliest_finish) + .ok_or(DagError::InfeasibleWcet(dag_id, idx.index()))?; + slacks.insert(idx, slack); } + // A node is critical if it has the least slack, which only the nodes on a longest path have. + let min_slack = slacks.values().min().copied().unwrap_or(Duration::ZERO); + for &idx in &order { - if let Some(info) = graph.node_weight_mut(idx) { - info.relative_deadline = - Some(latest_finish.get(&idx).copied().unwrap_or(sink_deadline)); - } + let info = graph + .node_weight_mut(idx) + .ok_or(DagError::NodeNotFound(dag_id, idx.index()))?; + info.timing.estimated_earliest_start = + earliest_start.get(&idx).copied().unwrap_or(Duration::ZERO); + info.timing.estimated_latest_finish = + latest_finish.get(&idx).copied().unwrap_or(sink_deadline); + info.timing.is_critical = slacks.get(&idx) == Some(&min_slack); } Ok(()) @@ -1419,11 +1436,11 @@ mod tests { let sink = dag.add_node_with_topic_edges(&["deadline_test_topic".into()], &[]); // A non-source node does not release a job. - assert_eq!(dag.get_or_release_release_time(sink, true, || 5), 5); + assert_eq!(dag.get_or_release_job(sink, true, || 5, |_| ()).0, 5); // The source node releases job 0. A later wake in job 0 keeps the deadline. - assert_eq!(dag.get_or_release_release_time(source, true, || 10), 10); - assert_eq!(dag.get_or_release_release_time(source, true, || 11), 10); + assert_eq!(dag.get_or_release_job(source, true, || 10, |_| ()).0, 10); + assert_eq!(dag.get_or_release_job(source, true, || 11, |_| ()).0, 10); // The input of job 0 can reach the sink node without a wake. assert!(dag.has_stale_deadline(sink)); @@ -1431,17 +1448,17 @@ mod tests { dag.finish_job(source); // A preempted source node resumes before the next period and does not release job 1. - assert_eq!(dag.get_or_release_release_time(source, false, || 15), 15); - assert_eq!(dag.get_or_release_release_time(source, true, || 20), 20); + assert_eq!(dag.get_or_release_job(source, false, || 15, |_| ()).0, 15); + assert_eq!(dag.get_or_release_job(source, true, || 20, |_| ()).0, 20); // The sink node wakes for job 0 after the release of job 1. - assert_eq!(dag.get_or_release_release_time(sink, true, || 21), 10); + assert_eq!(dag.get_or_release_job(sink, true, || 21, |_| ()).0, 10); assert!(!dag.has_stale_deadline(sink)); // The source node released job 1. dag.finish_job(sink); assert!(dag.has_stale_deadline(sink)); - assert_eq!(dag.get_or_release_release_time(sink, true, || 22), 20); + assert_eq!(dag.get_or_release_job(sink, true, || 22, |_| ()).0, 20); assert!(!dag.has_stale_deadline(sink)); assert!(!dag.release_times.lock(&mut MCSNode::new()).contains_key(&0)); } @@ -1455,19 +1472,19 @@ mod tests { let sink = dag.add_node_with_topic_edges(&["retention_test_b".into()], &[]); // The source node releases jobs 0, 1, and 2. - assert_eq!(dag.get_or_release_release_time(source, true, || 10), 10); + assert_eq!(dag.get_or_release_job(source, true, || 10, |_| ()).0, 10); dag.finish_job(source); - assert_eq!(dag.get_or_release_release_time(source, true, || 20), 20); + assert_eq!(dag.get_or_release_job(source, true, || 20, |_| ()).0, 20); dag.finish_job(source); - assert_eq!(dag.get_or_release_release_time(source, true, || 30), 30); + assert_eq!(dag.get_or_release_job(source, true, || 30, |_| ()).0, 30); // The sink node finishes job 0 before the middle node records its finish. dag.finish_job(sink); - assert_eq!(dag.get_or_release_release_time(middle, true, || 31), 10); + assert_eq!(dag.get_or_release_job(middle, true, || 31, |_| ()).0, 10); dag.finish_job(middle); - assert_eq!(dag.get_or_release_release_time(middle, true, || 32), 20); - assert_eq!(dag.get_or_release_release_time(sink, true, || 33), 20); + assert_eq!(dag.get_or_release_job(middle, true, || 32, |_| ()).0, 20); + assert_eq!(dag.get_or_release_job(sink, true, || 33, |_| ()).0, 20); let mut node = MCSNode::new(); let jobs: Vec = dag.release_times.lock(&mut node).keys().copied().collect(); @@ -1482,7 +1499,7 @@ mod tests { // The sink node stopped, for example by a panic, and the source node keeps releasing. for deadline in 0..100 { - dag.get_or_release_release_time(source, true, || deadline); + dag.get_or_release_job(source, true, || deadline, |_| ()).0; dag.finish_job(source); } @@ -1499,7 +1516,11 @@ mod tests { fn latest_finish_of(dag: &Dag, idx: NodeIndex) -> Duration { let mut node = MCSNode::new(); let graph = dag.graph.lock(&mut node); - graph.node_weight(idx).unwrap().relative_deadline.unwrap() + graph + .node_weight(idx) + .unwrap() + .timing + .estimated_latest_finish } // source -> a -> sink, and source -> b -> sink. @@ -1555,23 +1576,65 @@ mod tests { } #[test] - fn node_deadline_is_release_plus_relative_deadline() { - let (dag, [source, a, _b, sink]) = diamond("timing_deadline"); - - dag.set_node_wcet(sink, Duration::from_millis(20)).unwrap(); // Latest start: 100 - 20 = 80 - dag.set_node_wcet(a, Duration::from_millis(30)).unwrap(); // Latest start: 80 - 30 = 50 - dag.set_node_wcet(source, Duration::from_millis(10)) // Latest start: 50 - 10 = 40 + fn nodes_on_the_longest_path_are_critical() { + let (dag, [source, a, b, sink]) = diamond("timing_critical"); + dag.set_node_wcet(source, Duration::from_millis(10)) .unwrap(); + dag.set_node_wcet(a, Duration::from_millis(30)).unwrap(); + dag.set_node_wcet(b, Duration::from_millis(5)).unwrap(); + dag.set_node_wcet(sink, Duration::from_millis(20)).unwrap(); + assert!(compute_timing(&dag).is_ok()); + + // The path source -> a -> sink is 60 ms, shorter than D, and still the longest. + let critical = |idx| { + dag.get_or_release_job(idx, false, || 0, |timing| timing.is_critical) + .1 + }; + assert!(critical(source)); + assert!(critical(a)); + assert!(!critical(b)); + assert!(critical(sink)); + } + #[test] + fn released_job_returns_release_time_and_node_timing() { + let (dag, [source, a, _b, sink]) = diamond("timing_job"); + dag.set_node_wcet(sink, Duration::from_millis(20)).unwrap(); + dag.set_node_wcet(a, Duration::from_millis(30)).unwrap(); + dag.set_node_wcet(source, Duration::from_millis(10)) + .unwrap(); assert!(compute_timing(&dag).is_ok()); - assert_eq!( - dag.get_or_release_absolute_deadline(source, true, || 1_000), // job release time is 1_000 - 1_000 + 50_000 // release time + latest finish of sink (50) + let (release_time, (wcet, earliest_start, latest_finish)) = dag.get_or_release_job( + source, + true, + || 1_000, + |timing| { + ( + timing.estimated_wcet, + timing.estimated_earliest_start, + timing.estimated_latest_finish, + ) + }, ); - assert_eq!( - dag.get_or_release_absolute_deadline(sink, true, || 2_000), - 1_000 + 100_000 // release time + latest finish of sink (100) + assert_eq!(release_time, 1_000); + assert_eq!(wcet, Duration::from_millis(10)); + assert_eq!(earliest_start, Duration::ZERO); + assert_eq!(latest_finish, Duration::from_millis(50)); + + let (release_time, (earliest_start, latest_finish)) = dag.get_or_release_job( + sink, + true, + || 2_000, + |timing| { + ( + timing.estimated_earliest_start, + timing.estimated_latest_finish, + ) + }, ); + assert_eq!(release_time, 1_000); // The job of the source node. + assert_eq!(earliest_start, Duration::from_millis(40)); + assert_eq!(latest_finish, Duration::from_millis(100)); } } diff --git a/awkernel_async_lib/src/scheduler/dag_edf.rs b/awkernel_async_lib/src/scheduler/dag_edf.rs index fbc54613a..489a31fc0 100644 --- a/awkernel_async_lib/src/scheduler/dag_edf.rs +++ b/awkernel_async_lib/src/scheduler/dag_edf.rs @@ -41,8 +41,8 @@ pub trait DagEdfPolicy: 'static { /// Returns the absolute deadline of the job that the node `node_idx` of `dag` processes /// next. `release` is true if this wake may release that job. The policy gets the release - /// time of the job with `Dag::get_or_release_release_time` or - /// `Dag::get_or_release_absolute_deadline`, and passes `wake_time` as the release time. + /// time of the job and the timing parameters it needs with `Dag::get_or_release_job`, + /// and passes `wake_time` as the release time. fn dag_deadline(dag: &Dag, node_idx: NodeIndex, release: bool, wake_time: u64) -> u64; } diff --git a/awkernel_async_lib/src/scheduler/gedf.rs b/awkernel_async_lib/src/scheduler/gedf.rs index 58e78e90c..d021b234d 100644 --- a/awkernel_async_lib/src/scheduler/gedf.rs +++ b/awkernel_async_lib/src/scheduler/gedf.rs @@ -29,14 +29,11 @@ impl DagEdfPolicy for GedfPolicy { } fn dag_deadline(dag: &Dag, node_idx: NodeIndex, release: bool, wake_time: u64) -> u64 { - let release_time = dag.get_or_release_release_time(node_idx, release, || wake_time); - - // GEDF only uses the sink node's relative deadline + let (release_time, _) = dag.get_or_release_job(node_idx, release, || wake_time, |_| ()); let sink_deadline_us = dag .get_sink_relative_deadline() .map(|deadline| deadline.as_micros() as u64) .unwrap_or_else(|_| panic!("GEDF scheduler: DAG has no sink relative deadline set")); - release_time.saturating_add(sink_deadline_us) } } diff --git a/awkernel_async_lib/src/scheduler/redf.rs b/awkernel_async_lib/src/scheduler/redf.rs index a5ee753f1..f73061a29 100644 --- a/awkernel_async_lib/src/scheduler/redf.rs +++ b/awkernel_async_lib/src/scheduler/redf.rs @@ -26,8 +26,15 @@ impl DagEdfPolicy for RedfPolicy { _ => None, } } - // REDF uses per-node relative deadlines + + // REDF uses the estimated latest finish of each node. fn dag_deadline(dag: &Dag, node_idx: NodeIndex, release: bool, wake_time: u64) -> u64 { - dag.get_or_release_absolute_deadline(node_idx, release, || wake_time) + let (release_time, latest_finish) = dag.get_or_release_job( + node_idx, + release, + || wake_time, + |timing| timing.estimated_latest_finish.as_micros() as u64, + ); + release_time.saturating_add(latest_finish) } }