cu29_runtime/planner/
explicit_schedule.rs

1//! Portable exact process schedules and their build-time enforcement.
2
3use super::AssembledPlan;
4use super::CuMissionPlan;
5use super::CuPlanner;
6use super::LaneOccurrence;
7use super::LanePlan;
8use super::LaneWorker;
9use super::PlanningInput;
10use super::Serial;
11use super::StepOrder;
12use super::assemble_from_order;
13use super::assemble_runtime_plan_with_planner;
14use super::build_plan_graph;
15use super::mission_graphs;
16use super::schedule::PlanShape;
17use super::step_key;
18use crate::config::ComponentConfig;
19use crate::config::CuConfig;
20use crate::config::CuGraph;
21use crate::config::NodeId;
22use crate::config::PlannerConfig;
23use crate::config::PlannerKind;
24use crate::config::RuntimeConfig;
25use crate::config::Value;
26use crate::curuntime::CuExecutionUnit;
27use crate::curuntime::CuStepPhase;
28use alloc::collections::BTreeMap;
29use alloc::format;
30use alloc::string::String;
31use alloc::vec::Vec;
32use cu29_traits::CuError;
33use cu29_traits::CuResult;
34use serde::Deserialize;
35use serde::Serialize;
36
37/// An experimental portable process schedule for every mission in an app.
38///
39/// Copper validates a plan against the static task graph and generates typed
40/// calls and CopperList slots at compile time. The serialized shape is the
41/// schema: incompatible additions or removals are rejected by serde.
42#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
43#[serde(deny_unknown_fields)]
44pub struct CuPlan {
45    /// Resources (`bundle.resource`) that several components may use concurrently.
46    #[serde(default, skip_serializing_if = "Vec::is_empty")]
47    pub concurrent_resources: Vec<String>,
48    /// Schedule indexed by mission id (`default` without named missions).
49    pub missions: BTreeMap<String, CuMissionPlan>,
50}
51
52impl CuPlan {
53    /// Resolve and validate the scheduler selected by this configuration.
54    pub fn from_config(config: &CuConfig) -> CuResult<Self> {
55        super::resolve_schedule(config)
56    }
57
58    /// Export the canonical schedule for a repeating number of CopperLists.
59    pub fn from_config_cyclic(config: &CuConfig, copperlists_per_cycle: u32) -> CuResult<Self> {
60        Self::from_config_cyclic_with_resources(config, copperlists_per_cycle, &[])
61    }
62
63    pub(crate) fn from_config_cyclic_with_resources(
64        config: &CuConfig,
65        copperlists_per_cycle: u32,
66        concurrent_resources: &[String],
67    ) -> CuResult<Self> {
68        if copperlists_per_cycle == 0 {
69            return Err(CuError::from("copperlists_per_cycle must be positive"));
70        }
71        let mut missions = BTreeMap::new();
72        for (mission, graph) in mission_graphs(config) {
73            let assembled = assemble_runtime_plan_with_planner(config, graph, &Serial)?;
74            missions.insert(
75                mission.clone(),
76                PlanShape::new(&assembled, config, graph, &mission, concurrent_resources)?
77                    .cyclic_plan(copperlists_per_cycle)?,
78            );
79        }
80        Ok(Self {
81            concurrent_resources: concurrent_resources.to_vec(),
82            missions,
83        })
84    }
85
86    /// Validate inventory, placement, precedence, recurrence, and capacity.
87    pub fn validate(&self, config: &CuConfig) -> CuResult<()> {
88        self.validate_orders(config).map(|_| ())
89    }
90
91    fn validate_orders(&self, config: &CuConfig) -> CuResult<BTreeMap<String, Vec<usize>>> {
92        let missions = mission_graphs(config);
93        if !self.missions.keys().eq(missions.iter().map(|(id, _)| id)) {
94            return Err(CuError::from(
95                "Execution plan must contain exactly the configured missions",
96            ));
97        }
98        let mut orders = BTreeMap::new();
99        for (mission, graph) in missions {
100            let canonical = assemble_runtime_plan_with_planner(config, graph, &Serial)?;
101            let shape = PlanShape::new(
102                &canonical,
103                config,
104                graph,
105                &mission,
106                &self.concurrent_resources,
107            )?;
108            let order = self.missions[&mission]
109                .validate(config, &shape)
110                .map_err(|error| CuError::from(format!("Plan for mission '{mission}': {error}")))?;
111            orders.insert(mission, order);
112        }
113        Ok(orders)
114    }
115
116    /// Raise CopperList storage to the largest admission window in this plan.
117    pub fn provide_capacity(&self, config: &mut CuConfig) {
118        let needed = self
119            .missions
120            .values()
121            .map(|mission| mission.max_in_flight as usize)
122            .max()
123            .unwrap_or(0);
124        let current = config
125            .logging
126            .as_ref()
127            .and_then(|logging| logging.copperlist_count)
128            .unwrap_or(super::DEFAULT_COPPERLIST_COUNT);
129        if needed > current {
130            config
131                .logging
132                .get_or_insert_with(Default::default)
133                .copperlist_count = Some(needed);
134        }
135    }
136
137    /// Serialize this plan as human-editable RON.
138    pub fn serialize_ron(&self) -> CuResult<String> {
139        ron::ser::to_string_pretty(self, ron::ser::PrettyConfig::default())
140            .map_err(|error| CuError::new_with_cause("Could not serialize execution plan", error))
141    }
142
143    /// Parse a plan from RON.
144    pub fn deserialize_ron(text: &str) -> CuResult<Self> {
145        ron::from_str(text)
146            .map_err(|error| CuError::new_with_cause("Could not parse execution plan", error))
147    }
148
149    /// Read a standalone plan file in host tooling or a build script.
150    #[cfg(feature = "std")]
151    pub fn read(path: &std::path::Path) -> CuResult<Self> {
152        let text = std::fs::read_to_string(path)
153            .map_err(|error| CuError::new_with_cause("Could not read execution plan", error))?;
154        Self::deserialize_ron(&text)
155    }
156
157    /// Write a standalone plan file in host tooling or a build script.
158    #[cfg(feature = "std")]
159    pub fn write(&self, path: &std::path::Path) -> CuResult<()> {
160        std::fs::write(path, self.serialize_ron()?)
161            .map_err(|error| CuError::new_with_cause("Could not write execution plan", error))
162    }
163}
164
165/// Enforce an exact saved process schedule during runtime generation.
166#[derive(Clone, Debug)]
167pub struct ExplicitSchedule {
168    plan: CuPlan,
169}
170
171impl CuPlanner for ExplicitSchedule {
172    fn plan(&self, _input: &PlanningInput<'_>) -> CuResult<CuPlan> {
173        Ok(self.plan.clone())
174    }
175}
176
177impl ExplicitSchedule {
178    /// Take ownership of a saved plan.
179    pub fn new(plan: CuPlan) -> Self {
180        Self { plan }
181    }
182
183    /// Validate all missions and select this exact plan in `config`.
184    pub fn apply(&self, config: &mut CuConfig) -> CuResult<()> {
185        let mut prepared = config.clone();
186        self.plan.provide_capacity(&mut prepared);
187        self.plan.validate(&prepared)?;
188        let value = cu29_value::to_value(&self.plan)
189            .map_err(|error| CuError::new_with_cause("Could not encode execution plan", error))?;
190        let value = Value::deserialize(value)
191            .map_err(|error| CuError::new_with_cause("Could not embed execution plan", error))?;
192        let mut params = ComponentConfig::default();
193        params.set("plan", value);
194        prepared
195            .runtime
196            .get_or_insert_with(RuntimeConfig::default)
197            .planner = Some(PlannerConfig {
198            kind: PlannerKind::ExplicitSchedule,
199            config: Some(params),
200        });
201        *config = prepared;
202        Ok(())
203    }
204
205    pub(super) fn assemble(
206        &self,
207        config: &CuConfig,
208        graph: &CuGraph,
209        mission: &str,
210    ) -> CuResult<AssembledPlan> {
211        let orders = self.plan.validate_orders(config)?;
212        self.assemble_steps(config, graph, mission, &orders[mission])
213            .map_err(|error| {
214                CuError::from(format!(
215                    "ExplicitSchedule plan for mission '{mission}': {error}"
216                ))
217            })
218    }
219
220    fn assemble_steps(
221        &self,
222        config: &CuConfig,
223        graph: &CuGraph,
224        mission: &str,
225        order: &[usize],
226    ) -> CuResult<AssembledPlan> {
227        let plan = self
228            .plan
229            .missions
230            .get(mission)
231            .ok_or_else(|| CuError::from("Missing mission"))?;
232        let requested = if plan.is_serial() {
233            plan.serial_keys()?
234        } else {
235            plan.layout_keys(order)
236        };
237        let canonical = assemble_runtime_plan_with_planner(config, graph, &Serial)?;
238        let keys = execution_keys(&canonical, mission)?;
239        let by_key: BTreeMap<_, _> = keys.iter().enumerate().map(|(i, key)| (key, i)).collect();
240        let mut node_order = Vec::with_capacity(canonical.entities.len());
241        for key in &requested {
242            let index = *by_key
243                .get(key)
244                .ok_or_else(|| CuError::from(format!("Unknown process step '{key}'")))?;
245            let CuExecutionUnit::Step(step) = &canonical.execution.steps[index] else {
246                return Err(CuError::from("Nested loops cannot be materialized"));
247            };
248            if step.phase != CuStepPhase::AnytimeRefine {
249                node_order.push(step.node_id);
250            }
251        }
252        let mut assembled =
253            assemble_from_order(build_plan_graph(config, graph)?, StepOrder(node_order))?;
254        let remapped_keys = execution_keys(&assembled, mission)?;
255        let mut units: BTreeMap<_, _> = remapped_keys
256            .into_iter()
257            .zip(assembled.execution.steps)
258            .collect();
259        assembled.execution.steps = requested
260            .iter()
261            .map(|key| {
262                units.remove(key).ok_or_else(|| {
263                    CuError::from(format!("Could not materialize process step '{key}'"))
264                })
265            })
266            .collect::<CuResult<Vec<_>>>()?;
267        assembled.background = plan.background.clone();
268        if !plan.is_serial() {
269            let step_of: BTreeMap<_, _> = requested
270                .iter()
271                .enumerate()
272                .map(|(index, key)| (key.as_str(), index))
273                .collect();
274            assembled.lanes = Some(LanePlan {
275                copperlists_per_cycle: plan.copperlists_per_cycle,
276                max_in_flight: plan.max_in_flight,
277                occurrences: plan
278                    .steps
279                    .iter()
280                    .map(|step| LaneOccurrence {
281                        step: step_of[step.key.as_str()],
282                        copperlist: step.copperlist,
283                    })
284                    .collect(),
285                workers: plan
286                    .workers
287                    .iter()
288                    .map(|worker| LaneWorker {
289                        id: worker.id.clone(),
290                        placement: worker.placement.clone(),
291                        occurrences: worker.steps.iter().map(|&index| index as usize).collect(),
292                    })
293                    .collect(),
294                dispatcher: plan.dispatcher.clone(),
295                dependencies: plan.dependencies.clone(),
296            });
297        }
298        Ok(assembled)
299    }
300}
301
302pub(super) fn execution_keys(plan: &AssembledPlan, mission: &str) -> CuResult<Vec<String>> {
303    let mut refines: BTreeMap<NodeId, u32> = BTreeMap::new();
304    plan.execution
305        .steps
306        .iter()
307        .map(|unit| {
308            let CuExecutionUnit::Step(step) = unit else {
309                return Err(CuError::from(
310                    "Nested loops are not supported by portable plans",
311                ));
312            };
313            let ordinal = if step.phase == CuStepPhase::AnytimeRefine {
314                let next = refines.entry(step.node_id).or_default();
315                *next += 1;
316                Some(*next)
317            } else {
318                None
319            };
320            Ok(step_key(
321                mission,
322                &plan.entities[step.node_id as usize],
323                step.phase,
324                ordinal,
325            ))
326        })
327        .collect()
328}
cu29_runtime/planner/explicit_schedule.rs

cu29_runtime/planner/
explicit_schedule.rs
