cu29_runtime/planner/
pgs.rs

1//! Profile-guided scheduling inputs: the application contract and the
2//! execution profile extracted from a recorded log.
3
4use crate::config::CuConfig;
5use crate::config::Flavor;
6use crate::config::MAX_RT_PRIORITY;
7use crate::config::SchedulingPolicy;
8use alloc::collections::BTreeMap;
9use alloc::collections::BTreeSet;
10use alloc::collections::VecDeque;
11use alloc::format;
12use alloc::string::String;
13use alloc::string::ToString;
14use alloc::vec::Vec;
15use cu29_traits::CuError;
16use cu29_traits::CuResult;
17use serde::Deserialize;
18use serde::Serialize;
19
20const PROFILE_VERSION: u32 = 1;
21
22/// What the application asks of a schedule: the chains it measures, the CPUs
23/// it may use, and the capacity it may explore. Read only by planning tools.
24#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
25#[serde(deny_unknown_fields)]
26pub struct CuContract {
27    /// Latency chains, each measured inside one CopperList.
28    pub chains: Vec<CuChain>,
29    /// Expected delivery rate of sources, for the rate criterion.
30    #[serde(default)]
31    pub sources: Vec<CuSourceRate>,
32    /// CPUs the plan may pin workers to.
33    pub cpus: Vec<usize>,
34    /// Largest number of CopperLists a candidate may keep in flight.
35    pub max_in_flight: u32,
36    /// Fraction of each deadline and worker window reserved for timing
37    /// variation. A value of `0.2` keeps twenty percent free.
38    #[serde(default = "default_headroom")]
39    pub headroom: f64,
40    /// Scheduling policy of every worker thread a candidate creates.
41    #[serde(default)]
42    pub worker_policy: SchedulingPolicy,
43}
44
45/// A latency chain from a source task to a sink task, measured per
46/// CopperList as the sink's process end minus the source's time of validity
47/// (its process start when no time of validity is set).
48#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
49#[serde(deny_unknown_fields)]
50pub struct CuChain {
51    pub id: String,
52    pub source: String,
53    pub sink: String,
54    pub deadline_ms: u32,
55}
56
57/// The period a source is expected to deliver at.
58#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
59#[serde(deny_unknown_fields)]
60pub struct CuSourceRate {
61    pub task: String,
62    pub period_ms: u32,
63}
64
65fn default_headroom() -> f64 {
66    0.2
67}
68
69impl CuContract {
70    /// Checks the contract against one mission of `config`: every chain and
71    /// source names a task of the mission, every sink is reachable from its
72    /// source, and the CPU set and capacity are usable.
73    pub fn validate(&self, config: &CuConfig, mission: Option<&str>) -> CuResult<()> {
74        let graph = config.get_graph(mission)?;
75        let tasks: BTreeSet<String> = graph
76            .get_all_nodes()
77            .into_iter()
78            .filter(|(_, node)| node.get_flavor() == Flavor::Task)
79            .map(|(_, node)| node.get_id())
80            .collect();
81        let mut ids = BTreeSet::new();
82        for chain in &self.chains {
83            if chain.id.is_empty() || !ids.insert(&chain.id) {
84                return Err(CuError::from(format!(
85                    "Chain id '{}' is empty or used twice",
86                    chain.id
87                )));
88            }
89            for task in [&chain.source, &chain.sink] {
90                if !tasks.contains(task) {
91                    return Err(CuError::from(format!(
92                        "Chain '{}' names unknown task '{task}'",
93                        chain.id
94                    )));
95                }
96            }
97            if chain.deadline_ms == 0 {
98                return Err(CuError::from(format!(
99                    "Chain '{}' needs a positive deadline",
100                    chain.id
101                )));
102            }
103            let mut seen = BTreeSet::new();
104            let mut queue = VecDeque::from([chain.source.as_str()]);
105            while let Some(node) = queue.pop_front() {
106                if !seen.insert(node) {
107                    continue;
108                }
109                for edge in graph.edges().filter(|edge| edge.src == node) {
110                    queue.push_back(edge.dst.as_str());
111                }
112            }
113            if !seen.contains(chain.sink.as_str()) {
114                return Err(CuError::from(format!(
115                    "Chain '{}': task '{}' is not reachable from '{}'",
116                    chain.id, chain.sink, chain.source
117                )));
118            }
119        }
120        let mut source_tasks = BTreeSet::new();
121        for source in &self.sources {
122            if !source_tasks.insert(source.task.as_str()) {
123                return Err(CuError::from(format!(
124                    "Source rate for task '{}' is listed twice",
125                    source.task
126                )));
127            }
128            if !tasks.contains(&source.task) {
129                return Err(CuError::from(format!(
130                    "Source rate names unknown task '{}'",
131                    source.task
132                )));
133            }
134            if source.period_ms == 0 {
135                return Err(CuError::from(format!(
136                    "Source '{}' needs a positive period",
137                    source.task
138                )));
139            }
140        }
141        if self.cpus.is_empty() {
142            return Err(CuError::from("The contract needs at least one CPU"));
143        }
144        if self.cpus.iter().collect::<BTreeSet<_>>().len() != self.cpus.len() {
145            return Err(CuError::from("The contract lists a CPU twice"));
146        }
147        if self.max_in_flight == 0 {
148            return Err(CuError::from("max_in_flight must be positive"));
149        }
150        if !(0.0..1.0).contains(&self.headroom) {
151            return Err(CuError::from(
152                "headroom must be a fraction of the deadline in [0, 1)",
153            ));
154        }
155        match self.worker_policy {
156            SchedulingPolicy::Fifo { priority } | SchedulingPolicy::RoundRobin { priority }
157                if priority > MAX_RT_PRIORITY.saturating_sub(2) =>
158            {
159                return Err(CuError::from(format!(
160                    "worker_policy priority {priority} leaves no room for the high-priority worker and dispatcher"
161                )));
162            }
163            _ => {}
164        }
165        Ok(())
166    }
167
168    pub fn deserialize_ron(text: &str) -> CuResult<Self> {
169        ron::from_str(text).map_err(|e| CuError::new_with_cause("Could not parse contract", e))
170    }
171
172    pub fn serialize_ron(&self) -> CuResult<String> {
173        ron::ser::to_string_pretty(self, ron::ser::PrettyConfig::default())
174            .map_err(|e| CuError::new_with_cause("Could not serialize contract", e))
175    }
176
177    #[cfg(feature = "std")]
178    pub fn read(path: &std::path::Path) -> CuResult<Self> {
179        let text = std::fs::read_to_string(path)
180            .map_err(|e| CuError::new_with_cause("Could not read contract", e))?;
181        Self::deserialize_ron(&text)
182    }
183
184    #[cfg(feature = "std")]
185    pub fn write(&self, path: &std::path::Path) -> CuResult<()> {
186        std::fs::write(path, self.serialize_ron()?)
187            .map_err(|e| CuError::new_with_cause("Could not write contract", e))
188    }
189}
190
191/// Costs and chain latencies measured on one recorded run, keyed by the plan
192/// step keys of the mission the run executed.
193#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
194#[serde(deny_unknown_fields)]
195pub struct CuProfile {
196    pub version: u32,
197    /// Identity of the graph the run executed; a profile applies only to
198    /// configs with the same signature.
199    pub config_signature: String,
200    pub mission: String,
201    /// CopperLists in the measured window.
202    pub copperlists: u64,
203    /// Wall time from the first recorded process start to the last end.
204    pub window_ns: u64,
205    /// Wall span of each CopperList's process steps.
206    pub copperlist_span: CuCostStats,
207    pub operations: BTreeMap<String, CuOperationProfile>,
208    pub chains: BTreeMap<String, CuChainProfile>,
209    pub sources: BTreeMap<String, CuSourceProfile>,
210}
211
212/// One operation's cost, split by whether it produced an output payload.
213#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
214#[serde(deny_unknown_fields)]
215pub struct CuOperationProfile {
216    pub fired: CuCostStats,
217    pub skipped: CuCostStats,
218    /// Fired invocations per second over the window.
219    pub firing_rate_hz: f64,
220    /// Which CopperLists the operation fires in, when that repeats.
221    #[serde(default, skip_serializing_if = "Option::is_none")]
222    pub firing: Option<CuFiringPattern>,
223}
224
225/// An operation firing in the CopperLists whose id modulo `period` is one
226/// of `phases`, as observed over at least two periods.
227#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
228#[serde(deny_unknown_fields)]
229pub struct CuFiringPattern {
230    pub period: u32,
231    pub phases: Vec<u32>,
232}
233
234impl CuFiringPattern {
235    /// The smallest period under which `fired` (indexed by CopperList id
236    /// from `first_id`) repeats, checked over every pair of ids one period
237    /// apart; `None` below two periods of evidence or above `max_period`.
238    pub fn detect(first_id: u64, fired: &[bool], max_period: u32) -> Option<Self> {
239        (1..=max_period as usize)
240            .filter(|&period| fired.len() >= 2 * period)
241            .find(|&period| (period..fired.len()).all(|i| fired[i] == fired[i - period]))
242            .map(|period| Self {
243                period: period as u32,
244                phases: (0..period)
245                    .filter(|&i| fired[i])
246                    .map(|i| ((first_id + i as u64) % period as u64) as u32)
247                    .collect(),
248            })
249    }
250
251    /// The fraction of CopperLists with `id % modulus == offset` the
252    /// operation fires in.
253    pub fn probability(&self, modulus: u32, offset: u32) -> f64 {
254        let g = gcd(self.period, modulus);
255        let hits = self
256            .phases
257            .iter()
258            .filter(|&&phase| phase % g == offset % g)
259            .count() as f64;
260        hits / f64::from(self.period / g)
261    }
262}
263
264fn gcd(a: u32, b: u32) -> u32 {
265    if b == 0 { a } else { gcd(b, a % b) }
266}
267
268/// Summary of a set of durations, in nanoseconds; zeros when empty.
269#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
270#[serde(deny_unknown_fields)]
271pub struct CuCostStats {
272    pub samples: u64,
273    pub min_ns: u64,
274    pub p50_ns: u64,
275    pub mean_ns: f64,
276    pub p95_ns: u64,
277    pub p99_ns: u64,
278    pub max_ns: u64,
279}
280
281impl CuCostStats {
282    /// Summarizes `samples`, which it sorts in place.
283    pub fn from_samples(samples: &mut [u64]) -> Self {
284        if samples.is_empty() {
285            return Self::default();
286        }
287        samples.sort_unstable();
288        let percentile = |q: f64| {
289            // Nearest-rank index; `core` has no `round`.
290            let index = ((samples.len() - 1) as f64 * q + 0.5) as usize;
291            samples[index]
292        };
293        Self {
294            samples: samples.len() as u64,
295            min_ns: samples[0],
296            p50_ns: percentile(0.50),
297            mean_ns: samples.iter().map(|&v| v as f64).sum::<f64>() / samples.len() as f64,
298            p95_ns: percentile(0.95),
299            p99_ns: percentile(0.99),
300            max_ns: samples[samples.len() - 1],
301        }
302    }
303}
304
305/// Measured latency of one chain.
306#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
307#[serde(deny_unknown_fields)]
308pub struct CuChainProfile {
309    pub deadline_ms: u32,
310    pub latency: CuCostStats,
311    /// Samples over the deadline.
312    pub misses: u64,
313}
314
315/// Delivered rate of one source against the contract's period.
316#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
317#[serde(deny_unknown_fields)]
318pub struct CuSourceProfile {
319    pub period_ms: u32,
320    pub fired: u64,
321    /// Intervals of the period over the source's firing span (at least the
322    /// run's window less one period).
323    pub expected: f64,
324    /// `(fired - 1) / expected`: the intervals between firings over the
325    /// intervals the period allows; 1 for a source firing on its period.
326    pub delivered_rate: f64,
327}
328
329/// A stable identity of one mission graph: its nodes, types and edges. A
330/// profile applies only to a config with the same signature.
331pub fn graph_signature(graph: &crate::config::CuGraph, mission: Option<&str>) -> String {
332    let mut parts = Vec::new();
333    parts.push(format!("mission={}", mission.unwrap_or("default")));
334    let mut nodes: Vec<_> = graph.get_all_nodes();
335    nodes.sort_by_key(|a| a.1.get_id());
336    for (_, node) in nodes {
337        let flavor = match node.get_flavor() {
338            Flavor::Task => "task",
339            Flavor::Bridge => "bridge",
340        };
341        parts.push(format!(
342            "node|{}|{}|{flavor}|kind={:?}|background={}|pool={}|anytime={:?}",
343            node.get_id(),
344            node.get_type(),
345            node.get_declared_task_kind(),
346            node.is_background(),
347            node.background_pool(),
348            node.anytime(),
349        ));
350        if let Some(resources) = node.get_resources() {
351            let mut resources: Vec<_> = resources.iter().collect();
352            resources.sort_by_key(|(binding, _)| *binding);
353            for (binding, resource) in resources {
354                parts.push(format!("resource|{}|{binding}|{resource}", node.get_id()));
355            }
356        }
357        if let Some(config) = node.get_instance_config() {
358            let mut values: Vec<_> = config.0.iter().collect();
359            values.sort_by_key(|(key, _)| *key);
360            for (key, value) in values {
361                parts.push(format!("config|{}|{key}|{value}", node.get_id()));
362            }
363        }
364    }
365    let endpoint = |node: &str, channel: Option<&str>| match channel {
366        Some(channel) => format!("{node}/{channel}"),
367        None => node.to_string(),
368    };
369    let mut edges: Vec<String> = graph
370        .edges()
371        .map(|cnx| {
372            format!(
373                "edge|{}|{}|{}",
374                endpoint(cnx.src.as_str(), cnx.src_channel.as_deref()),
375                endpoint(cnx.dst.as_str(), cnx.dst_channel.as_deref()),
376                cnx.msg
377            )
378        })
379        .collect();
380    edges.sort();
381    parts.extend(edges);
382    let joined = parts.join("\n");
383    let mut hash: u64 = 0xcbf29ce484222325;
384    for byte in joined.as_bytes() {
385        hash ^= u64::from(*byte);
386        hash = hash.wrapping_mul(0x100000001b3);
387    }
388    format!("fnv1a64:{hash:016x}")
389}
390
391impl CuProfile {
392    pub fn new(config_signature: String, mission: String) -> Self {
393        Self {
394            version: PROFILE_VERSION,
395            config_signature,
396            mission,
397            copperlists: 0,
398            window_ns: 0,
399            copperlist_span: CuCostStats::default(),
400            operations: BTreeMap::new(),
401            chains: BTreeMap::new(),
402            sources: BTreeMap::new(),
403        }
404    }
405
406    pub fn serialize_ron(&self) -> CuResult<String> {
407        ron::ser::to_string_pretty(self, ron::ser::PrettyConfig::default())
408            .map_err(|e| CuError::new_with_cause("Could not serialize profile", e))
409    }
410
411    pub fn deserialize_ron(text: &str) -> CuResult<Self> {
412        let profile: Self = ron::from_str(text)
413            .map_err(|e| CuError::new_with_cause("Could not parse profile", e))?;
414        if profile.version != PROFILE_VERSION {
415            return Err(CuError::from(format!(
416                "Unsupported profile version {}; expected {PROFILE_VERSION}",
417                profile.version
418            )));
419        }
420        Ok(profile)
421    }
422
423    #[cfg(feature = "std")]
424    pub fn read(path: &std::path::Path) -> CuResult<Self> {
425        let text = std::fs::read_to_string(path)
426            .map_err(|e| CuError::new_with_cause("Could not read profile", e))?;
427        Self::deserialize_ron(&text)
428    }
429
430    #[cfg(feature = "std")]
431    pub fn write(&self, path: &std::path::Path) -> CuResult<()> {
432        std::fs::write(path, self.serialize_ron()?)
433            .map_err(|e| CuError::new_with_cause("Could not write profile", e))
434    }
435}
436
437#[cfg(test)]
438mod tests {
439    use super::*;
440
441    fn config() -> CuConfig {
442        CuConfig::deserialize_ron(
443            r#"(
444            tasks: [(id: "src", type: "Source"), (id: "work", type: "Work"), (id: "sink", type: "Sink"), (id: "other", type: "Source")],
445            cnx: [(src: "src", dst: "work", msg: "u32"), (src: "work", dst: "sink", msg: "u32"), (src: "other", dst: "__nc__", msg: "u32")],
446        )"#,
447        )
448        .unwrap()
449    }
450
451    fn contract() -> CuContract {
452        CuContract::deserialize_ron(
453            r#"(
454            chains: [(id: "hot", source: "src", sink: "sink", deadline_ms: 10)],
455            sources: [(task: "src", period_ms: 20)],
456            cpus: [0, 1],
457            max_in_flight: 2,
458        )"#,
459        )
460        .unwrap()
461    }
462
463    #[test]
464    fn contract_parses_with_defaults_and_validates() {
465        let contract = contract();
466        assert_eq!(contract.headroom, 0.2);
467        contract.validate(&config(), None).unwrap();
468        assert_eq!(
469            CuContract::deserialize_ron(&contract.serialize_ron().unwrap()).unwrap(),
470            contract
471        );
472    }
473
474    #[test]
475    fn contract_rejects_unknown_tasks_unreachable_sinks_and_bad_values() {
476        let config = config();
477        let mutations: [fn(&mut CuContract); 9] = [
478            |c| c.chains[0].sink = "missing".into(),
479            |c| c.chains[0].sink = "other".into(),
480            |c| c.chains[0].deadline_ms = 0,
481            |c| c.chains.push(c.chains[0].clone()),
482            |c| c.sources[0].task = "missing".into(),
483            |c| c.sources.push(c.sources[0].clone()),
484            |c| c.cpus = vec![1, 1],
485            |c| c.max_in_flight = 0,
486            |c| c.headroom = 1.0,
487        ];
488        for (index, mutate) in mutations.iter().enumerate() {
489            let mut invalid = contract();
490            mutate(&mut invalid);
491            assert!(invalid.validate(&config, None).is_err(), "mutation {index}");
492        }
493    }
494
495    #[test]
496    fn cost_stats_summarize_and_profile_round_trips() {
497        let mut samples = [5u64, 1, 4, 2, 3];
498        let stats = CuCostStats::from_samples(&mut samples);
499        assert_eq!(
500            (stats.samples, stats.min_ns, stats.p50_ns, stats.max_ns),
501            (5, 1, 3, 5)
502        );
503        assert_eq!(stats.mean_ns, 3.0);
504        assert_eq!(CuCostStats::from_samples(&mut []), CuCostStats::default());
505        let mut profile = CuProfile::new("sig".into(), "default".into());
506        profile.operations.insert(
507            "mission:default|task:src|phase:whole".into(),
508            CuOperationProfile {
509                fired: stats,
510                skipped: CuCostStats::default(),
511                firing_rate_hz: 50.0,
512                firing: None,
513            },
514        );
515        let text = profile.serialize_ron().unwrap();
516        assert_eq!(CuProfile::deserialize_ron(&text).unwrap(), profile);
517        assert!(CuProfile::deserialize_ron(&text.replace("version: 1", "version: 9")).is_err());
518    }
519
520    #[test]
521    fn a_firing_pattern_is_detected_and_projected_on_another_modulus() {
522        // Fires in CopperLists 3, 6, 9, ... observed from id 2.
523        let fired: Vec<bool> = (2u64..40).map(|id| id.is_multiple_of(3)).collect();
524        let pattern = CuFiringPattern::detect(2, &fired, 64).unwrap();
525        assert_eq!((pattern.period, pattern.phases.clone()), (3, vec![0]));
526        assert_eq!(pattern.probability(3, 0), 1.0);
527        assert_eq!(pattern.probability(3, 1), 0.0);
528        // Against a window of 4, every offset fires in a third of its CopperLists.
529        assert!((pattern.probability(4, 1) - 1.0 / 3.0).abs() < 1e-9);
530        // Below two periods of evidence there is no pattern.
531        assert!(CuFiringPattern::detect(0, &[true, false, false, true, false], 64).is_none());
532        let every = CuFiringPattern::detect(0, &[true; 8], 64).unwrap();
533        assert_eq!((every.period, every.phases), (1, vec![0]));
534    }
535}
cu29_runtime/planner/pgs.rs

cu29_runtime/planner/
pgs.rs
