1#[proc_macro_attribute]
9pub fn copper_runtime(args: TokenStream, input: TokenStream) -> TokenStream {
10 #[cfg(feature = "macro_debug")]
11 eprintln!("[entry]");
12 let mut application_struct = parse_macro_input!(input as ItemStruct);
13
14 let application_name = &application_struct.ident;
15 let builder_name = format_ident!("{}Builder", application_name);
16 let runtime_args = match CopperRuntimeArgs::parse_tokens(args.into()) {
17 Ok(runtime_args) => runtime_args,
18 Err(err) => return err.to_compile_error().into(),
19 };
20 let config_file = runtime_args.config_path.clone();
21 let sim_mode = runtime_args.sim_mode;
22 let ignore_resources = runtime_args.ignore_resources;
23
24 #[cfg(feature = "std")]
25 let std = true;
26
27 #[cfg(not(feature = "std"))]
28 let std = false;
29 let signal_handler = cfg!(feature = "signal-handler");
30 let parallel_rt_enabled = cfg!(feature = "parallel-rt");
31 let logstream_enabled = cfg!(feature = "logstream");
32 let clock_sync_enabled = cfg!(feature = "clock-sync");
33 let rt_guard = rtsan_guard_tokens();
34
35 if ignore_resources && !sim_mode {
36 return return_error(
37 "`ignore_resources` is only supported when `sim_mode` is enabled".to_string(),
38 );
39 }
40
41 let resolved_runtime_config = match resolve_runtime_config(&runtime_args) {
51 Ok(resolved_runtime_config) => resolved_runtime_config,
52 Err(e) => return return_error(e.to_string()),
53 };
54 let subsystem_code = resolved_runtime_config.subsystem_code;
55 let subsystem_id = resolved_runtime_config.subsystem_id.clone();
56 let config_features = resolved_runtime_config.active_features.clone();
57 let copper_config_content = resolved_runtime_config.bundled_local_config_content.clone();
58 let copper_config = resolved_runtime_config.local_config;
59 if copper_config.log_streaming.is_some() && !logstream_enabled {
60 return return_error(
61 "copperconfig.ron declares log_streaming but the cu29 'logstream' feature is disabled"
62 .to_string(),
63 );
64 }
65 if copper_config.log_streaming.is_some() && ignore_resources {
66 return return_error(
67 "log_streaming resource bindings cannot be used with ignore_resources".to_string(),
68 );
69 }
70 if copper_config.runtime.as_ref().and_then(|r| r.clock_sync_config()).is_some() && !clock_sync_enabled {
71 return return_error("runtime.clock requires the cu29 'clock-sync' feature".to_string());
72 }
73 let copperlist_count = copper_config
74 .logging
75 .as_ref()
76 .and_then(|logging| logging.copperlist_count)
77 .unwrap_or(DEFAULT_COPPERLIST_COUNT);
78 let pipeline_selected = copper_config.planner_kind() == PlannerKind::Pipeline;
79 if pipeline_selected && copper_config.runtime.as_ref().and_then(|r| r.clock_sync_config()).is_some() {
80 return return_error("Clock synchronization requires the Serial planner".to_string());
81 }
82 if pipeline_selected && !(std && parallel_rt_enabled) {
83 return return_error(
84 "runtime planner kind Pipeline requires the cu29 'parallel-rt' feature".to_string(),
85 );
86 }
87 let pipeline_max_in_flight = match copper_config.planner_config() {
88 Some(planner) => match planner.max_in_flight(copperlist_count) {
89 Ok(limit) => limit,
90 Err(error) => return return_error(error.to_string()),
91 },
92 None => 1,
93 };
94 let pipeline_max_in_flight_tokens =
95 proc_macro2::Literal::usize_unsuffixed(pipeline_max_in_flight);
96 let local_keyframe_logging_enabled = copper_config
97 .logging
98 .as_ref()
99 .is_none_or(|logging| logging.enable_keyframe_logging && logging.enable_task_logging);
100 let keyframe_logging_enabled = local_keyframe_logging_enabled || logstream_enabled;
104 let copperlist_count_tokens = proc_macro2::Literal::usize_unsuffixed(copperlist_count);
105 let caller_root = utils::caller_crate_root();
106 let (git_commit, git_dirty) = detect_git_info(&caller_root);
107 let git_commit_tokens = if let Some(commit) = git_commit {
108 quote! { Some(#commit.to_string()) }
109 } else {
110 quote! { None }
111 };
112 let git_dirty_tokens = if let Some(dirty) = git_dirty {
113 quote! { Some(#dirty) }
114 } else {
115 quote! { None }
116 };
117 let subsystem_code_literal = proc_macro2::Literal::u16_unsuffixed(subsystem_code);
118 let subsystem_id_tokens = if let Some(subsystem_id) = subsystem_id.as_deref() {
119 quote! { Some(#subsystem_id) }
120 } else {
121 quote! { None }
122 };
123
124 #[cfg(feature = "macro_debug")]
125 eprintln!("[build monitor type]");
126 let monitor_configs = copper_config.get_monitor_configs();
127 let (monitor_type, monitor_instanciator_body) = if monitor_configs.is_empty() {
128 (
129 quote! { NoMonitor },
130 quote! {
131 let monitor_metadata = metadata.with_subsystem_id(#subsystem_id_tokens);
132 let monitor = NoMonitor::new(monitor_metadata, runtime)
133 .expect("Failed to create NoMonitor.");
134 monitor
135 },
136 )
137 } else if monitor_configs.len() == 1 {
138 let only_monitor_type = parse_str::<Type>(monitor_configs[0].get_type())
139 .expect("Could not transform the monitor type name into a Rust type.");
140 (
141 quote! { #only_monitor_type },
142 quote! {
143 let monitor_metadata = metadata.with_monitor_config(
144 config
145 .get_monitor_configs()
146 .first()
147 .and_then(|entry| entry.get_config().cloned())
148 )
149 .with_subsystem_id(#subsystem_id_tokens);
150 let monitor = #only_monitor_type::new(monitor_metadata, runtime)
151 .expect("Failed to create the given monitor.");
152 monitor
153 },
154 )
155 } else {
156 let monitor_types: Vec<Type> = monitor_configs
157 .iter()
158 .map(|monitor_config| {
159 parse_str::<Type>(monitor_config.get_type())
160 .expect("Could not transform the monitor type name into a Rust type.")
161 })
162 .collect();
163 let monitor_bindings: Vec<Ident> = (0..monitor_types.len())
164 .map(|idx| format_ident!("__cu_monitor_{idx}"))
165 .collect();
166 let monitor_indices: Vec<syn::Index> =
167 (0..monitor_types.len()).map(syn::Index::from).collect();
168
169 let monitor_builders: Vec<proc_macro2::TokenStream> = monitor_types
170 .iter()
171 .zip(monitor_bindings.iter())
172 .zip(monitor_indices.iter())
173 .map(|((monitor_ty, monitor_binding), monitor_idx)| {
174 quote! {
175 let __cu_monitor_cfg_entry = config
176 .get_monitor_configs()
177 .get(#monitor_idx)
178 .and_then(|entry| entry.get_config().cloned());
179 let __cu_monitor_metadata = metadata
180 .clone()
181 .with_monitor_config(__cu_monitor_cfg_entry)
182 .with_subsystem_id(#subsystem_id_tokens);
183 let #monitor_binding = #monitor_ty::new(__cu_monitor_metadata, runtime.clone())
184 .expect("Failed to create one of the configured monitors.");
185 }
186 })
187 .collect();
188 let tuple_type: TypeTuple = parse_quote! { (#(#monitor_types),*,) };
189 (
190 quote! { #tuple_type },
191 quote! {
192 #(#monitor_builders)*
193 let monitor: #tuple_type = (#(#monitor_bindings),*,);
194 monitor
195 },
196 )
197 };
198
199 #[cfg(feature = "macro_debug")]
201 eprintln!("[build runtime field]");
202 let runtime_field: Field = if sim_mode {
204 parse_quote! {
205 copper_runtime: cu29::curuntime::CuRuntime<CuSimTasks, CuBridges, CuStampedDataSet, #monitor_type, #copperlist_count_tokens>
206 }
207 } else {
208 parse_quote! {
209 copper_runtime: cu29::curuntime::CuRuntime<CuTasks, CuBridges, CuStampedDataSet, #monitor_type, #copperlist_count_tokens>
210 }
211 };
212 let lifecycle_sink_field: Field = parse_quote! {
213 runtime_lifecycle_sink: Option<Box<::cu29::curuntime::RuntimeLifecycleSink>>
214 };
215 let logger_runtime_field: Field = parse_quote! {
216 logger_runtime: cu29::prelude::LoggerRuntime
217 };
218
219 #[cfg(feature = "macro_debug")]
220 eprintln!("[match struct anonymity]");
221 match &mut application_struct.fields {
222 Named(fields_named) => {
223 fields_named.named.push(runtime_field);
224 fields_named.named.push(lifecycle_sink_field);
225 fields_named.named.push(parse_quote! { lifecycle_running: bool });
226 fields_named.named.push(parse_quote! { shutdown_completed: bool });
227 fields_named.named.push(parse_quote! { stop_reason: ::cu29::curuntime::RuntimeStopReason });
228 fields_named.named.push(logger_runtime_field);
229 if clock_sync_enabled {
230 fields_named.named.push(parse_quote! { clock_maintenance: Option<::cu29::clock_sync::ClockMaintenance> });
231 if sim_mode { fields_named.named.push(parse_quote! { replay_clock_sync: Option<::cu29::clock::sync::ClockSync> }); }
232 }
233 }
234 Unnamed(fields_unnamed) => {
235 fields_unnamed.unnamed.push(runtime_field);
236 fields_unnamed.unnamed.push(lifecycle_sink_field);
237 fields_unnamed.unnamed.push(parse_quote! { lifecycle_running: bool });
238 fields_unnamed.unnamed.push(parse_quote! { shutdown_completed: bool });
239 fields_unnamed.unnamed.push(parse_quote! { stop_reason: ::cu29::curuntime::RuntimeStopReason });
240 fields_unnamed.unnamed.push(logger_runtime_field);
241 if clock_sync_enabled {
242 fields_unnamed.unnamed.push(parse_quote! { clock_maintenance: Option<::cu29::clock_sync::ClockMaintenance> });
243 if sim_mode { fields_unnamed.unnamed.push(parse_quote! { replay_clock_sync: Option<::cu29::clock::sync::ClockSync> }); }
244 }
245 }
246 Fields::Unit => {
247 panic!(
248 "This struct is a unit struct, it should have named or unnamed fields. use struct Something {{}} and not struct Something;"
249 )
250 }
251 };
252
253 let all_missions = sorted_mission_graphs(&copper_config);
254 let constant_modules = match build_constant_modules(&copper_config.constants) {
255 Ok(modules) => modules,
256 Err(error) => return return_error(error.to_string()),
257 };
258 let constant_fingerprints = match copper_config
259 .constants
260 .iter()
261 .map(|constant| {
262 constant.semantic_fingerprint().map(|fingerprint| {
263 (
264 constant.module_path().to_string(),
265 constant.id().to_string(),
266 fingerprint,
267 )
268 })
269 })
270 .collect::<Result<Vec<_>, _>>()
271 {
272 Ok(fingerprints) => fingerprints,
273 Err(error) => return return_error(error),
274 };
275 let task_input_layouts = match collect_task_input_layouts(&all_missions) {
276 Ok(layouts) => layouts,
277 Err(e) => return return_error(e.to_string()),
278 };
279 let mut all_missions_tokens = Vec::<proc_macro2::TokenStream>::new();
280 for (mission_index, (mission, graph)) in all_missions.iter().enumerate() {
281 let mission_index = mission_index as u32;
282 let mission_names = all_missions.iter().map(|(name, _)| name.as_str()).collect::<Vec<_>>();
283 let git_commit_tokens = git_commit_tokens.clone();
284 let git_dirty_tokens = git_dirty_tokens.clone();
285 let mission_mod = parse_str::<Ident>(mission.as_str())
286 .expect("Could not make an identifier of the mission name");
287 let mission_constant_contents = constant_modules.child_contents(&mission_mod);
288
289 #[cfg(feature = "macro_debug")]
290 eprintln!("[extract tasks ids & types]");
291 let task_specs = match CuTaskSpecSet::from_graph(graph) {
292 Ok(specs) => specs,
293 Err(e) => return return_error(e.to_string()),
294 };
295
296 let culist_channel_usage = collect_bridge_channel_usage(graph);
297 let mut culist_bridge_specs =
298 build_bridge_specs(&copper_config, graph, &culist_channel_usage);
299 let (culist_plan, culist_exec_entities, culist_plan_to_original, culist_schedule) =
300 match build_execution_plan_with_schedule(
301 &copper_config,
302 graph,
303 mission,
304 &mut culist_bridge_specs,
305 ) {
306 Ok(plan) => plan,
307 Err(e) => {
308 return return_error(format!(
309 "Could not compute copperlist plan for mission '{mission}': {e}"
310 ));
311 }
312 };
313 let lane_plan = culist_schedule.lanes.as_ref();
314 let distributed_selected = lane_plan.is_some() && !sim_mode;
315 if distributed_selected && !(std && parallel_rt_enabled) {
316 return return_error(format!(
317 "Mission '{mission}': this plan requires the cu29 'parallel-rt' feature"
318 ));
319 }
320 let mut task_specs = task_specs;
321 for entry in &culist_schedule.background {
322 let Some(index) = task_specs.ids.iter().position(|id| *id == entry.task) else {
323 return return_error(format!(
324 "Mission '{mission}': background plan entry names unknown task '{}'",
325 entry.task
326 ));
327 };
328 if let CuPlanBackgroundResult::Lag { lag } = entry.result {
329 if lag != 1 || entry.max_running != 1 {
330 return return_error(format!(
331 "Mission '{mission}': background task '{}' asks for result lag {lag} with max_running {}; this runtime implements lag 1 with max_running 1.",
332 entry.task, entry.max_running
333 ));
334 }
335 task_specs.background_result_lags[index] = Some(lag);
336 }
337 }
338 let task_specs = task_specs;
339 let schedule_max_in_flight = lane_plan
340 .map(|plan| plan.max_in_flight as usize)
341 .unwrap_or(1);
342 let schedule_max_in_flight_tokens =
343 proc_macro2::Literal::usize_unsuffixed(schedule_max_in_flight);
344
345 let (anytime_refine_ordinals, anytime_refine_totals): (
353 Vec<Option<u32>>,
354 HashMap<NodeId, u32>,
355 ) = {
356 let mut running: HashMap<NodeId, u32> = HashMap::new();
357 let ordinals = culist_plan
358 .steps
359 .iter()
360 .map(|unit| match unit {
361 CuExecutionUnit::Step(step) if step.phase == CuStepPhase::AnytimeRefine => {
362 let ordinal = running.entry(step.node_id).or_insert(0u32);
363 *ordinal += 1;
364 Some(*ordinal)
365 }
366 _ => None,
367 })
368 .collect();
369 (ordinals, running)
370 };
371
372 let task_names = collect_task_names(graph);
373 let (culist_call_order, node_output_positions) = collect_culist_metadata(
374 &culist_plan,
375 &culist_exec_entities,
376 &mut culist_bridge_specs,
377 &culist_plan_to_original,
378 );
379
380 #[cfg(feature = "macro_debug")]
381 {
382 eprintln!("[runtime plan for mission {mission}]");
383 eprintln!("{culist_plan:?}");
384 }
385
386 let culist_support: proc_macro2::TokenStream = gen_culist_support(
387 CuListSupportOptions {
388 cuconfig: &copper_config,
389 distributed: distributed_selected,
390 },
391 Some(mission.as_str()),
392 &culist_plan,
393 &culist_call_order,
394 &node_output_positions,
395 &task_names,
396 &culist_bridge_specs,
397 );
398
399 let (
400 resources_module,
401 resources_instanciator_fn,
402 task_resource_mappings,
403 bridge_resource_mappings,
404 logstream_resource_specs,
405 ) = if ignore_resources {
406 let bundle_specs: Vec<BundleSpec> = Vec::new();
407 let resource_specs: Vec<ResourceKeySpec> = Vec::new();
408 let (resources_module, resources_instanciator_fn) =
409 match build_resources_module(&bundle_specs) {
410 Ok(tokens) => tokens,
411 Err(e) => return return_error(e.to_string()),
412 };
413 let task_resource_mappings =
414 match build_task_resource_mappings(&resource_specs, &task_specs, sim_mode) {
415 Ok(tokens) => tokens,
416 Err(e) => return return_error(e.to_string()),
417 };
418 let bridge_resource_mappings =
419 build_bridge_resource_mappings(&resource_specs, &culist_bridge_specs, sim_mode);
420 (
421 resources_module,
422 resources_instanciator_fn,
423 task_resource_mappings,
424 bridge_resource_mappings,
425 Vec::new(),
426 )
427 } else {
428 let bundle_specs = match build_bundle_specs(&copper_config, mission.as_str()) {
429 Ok(specs) => specs,
430 Err(e) => return return_error(e.to_string()),
431 };
432
433 let resource_specs = match collect_resource_specs(
434 graph,
435 &task_specs,
436 &culist_bridge_specs,
437 &bundle_specs,
438 ) {
439 Ok(specs) => specs,
440 Err(e) => return return_error(e.to_string()),
441 };
442
443 let (resources_module, resources_instanciator_fn) =
444 match build_resources_module(&bundle_specs) {
445 Ok(tokens) => tokens,
446 Err(e) => return return_error(e.to_string()),
447 };
448 let task_resource_mappings =
449 match build_task_resource_mappings(&resource_specs, &task_specs, sim_mode) {
450 Ok(tokens) => tokens,
451 Err(e) => return return_error(e.to_string()),
452 };
453 let bridge_resource_mappings =
454 build_bridge_resource_mappings(&resource_specs, &culist_bridge_specs, sim_mode);
455 let logstream_resource_specs = if sim_mode {
456 Vec::new()
457 } else {
458 match build_logstream_resource_specs(
459 &copper_config,
460 mission.as_str(),
461 &resource_specs,
462 ) {
463 Ok(specs) => specs,
464 Err(e) => return return_error(e.to_string()),
465 }
466 };
467 (
468 resources_module,
469 resources_instanciator_fn,
470 task_resource_mappings,
471 bridge_resource_mappings,
472 logstream_resource_specs,
473 )
474 };
475
476 #[cfg(feature = "clock-sync")]
477 let clock_reference_factory = match clock_sync::factory(&copper_config, mission, sim_mode) {
478 Ok(factory) => Some(factory), Err(error) => return return_error(error.to_string()),
479 };
480 #[cfg(not(feature = "clock-sync"))]
481 let clock_reference_factory: Option<proc_macro2::TokenStream> = None;
482 let clock_maintenance_field = clock_sync_enabled.then(|| quote! { pub clock_maintenance: Option<::cu29::clock_sync::ClockMaintenance>, });
483 let clock_maintenance_none = clock_sync_enabled.then(|| quote! { clock_maintenance: None, });
484 let clock_maintenance_destructure = clock_sync_enabled.then(|| quote! { mut clock_maintenance, });
485 let clock_maintenance_init = clock_sync_enabled.then(|| quote! { clock_maintenance, });
486 let replay_clock_init = (clock_sync_enabled && sim_mode).then(|| quote! { replay_clock_sync: None, });
487 let clock_maintenance_compat = (clock_sync_enabled && !sim_mode).then(|| quote! {
488 if clock_maintenance.is_none() {
489 if let Some((reference, settings)) = #mission_mod::clock_reference_instanciator(&config, &mut resources)? {
490 clock_maintenance = Some(::cu29::clock_sync::ClockMaintenance::new(reference, clock.clone(), settings)?);
491 }
492 }
493 });
494 let clock_start = (clock_sync_enabled && !sim_mode).then(|| quote! {
495 if let Some(maintenance) = &mut self.clock_maintenance { maintenance.start()?; }
496 let next_id = self.copper_runtime.copperlists_manager.next_cl_id();
497 let record = self.clock_maintenance.as_mut().and_then(|maintenance| maintenance.take_record(next_id));
498 if let Some(record) = record { self.log_runtime_lifecycle_event(RuntimeLifecycleEvent::ClockSync(record))?; }
499 });
500 let clock_stop = (clock_sync_enabled && !sim_mode).then(|| quote! {
501 if let Some(maintenance) = &mut self.clock_maintenance { maintenance.stop()?; }
502 });
503 let clock_maintain = (clock_sync_enabled && !sim_mode).then(|| quote! {
504 let clock_result = self.clock_maintenance.as_mut().map(|maintenance| maintenance.maintain()).transpose();
505 let next_id = self.copper_runtime.copperlists_manager.next_cl_id();
506 let record = self.clock_maintenance.as_mut().and_then(|maintenance| maintenance.take_record(next_id));
507 if let Some(record) = record { self.log_runtime_lifecycle_event(RuntimeLifecycleEvent::ClockSync(record))?; }
508 clock_result?;
509 });
510
511 let task_ids = task_specs.ids.clone();
512 let autogenerated_output_warnings: Vec<proc_macro2::TokenStream> = task_specs
513 .ids
514 .iter()
515 .zip(task_specs.cutypes.iter())
516 .zip(task_specs.stateless_flags.iter())
517 .zip(task_specs.autogenerated_output_flags.iter())
518 .filter_map(|(((task_id, task_kind), stateless), autogenerated)| {
519 if !*autogenerated {
520 return None;
521 }
522 let warn_ident = format_ident!(
523 "__CU_AUTOGEN_FLOATING_OUTPUT_WARNING__{}",
524 config_id_to_enum(task_id)
525 );
526 let kind_str = match task_kind {
527 CuTaskType::Source => "source",
528 CuTaskType::Regular if *stateless => "stateless_task",
529 CuTaskType::Regular => "task",
530 CuTaskType::Sink => return None,
531 };
532 let note = format!(
533 "Task '{task_id}' is declared as kind '{kind_str}' but has no declared outputs. Copper synthesized a hidden floating output slot from the task trait. Add a real consumer or `dst: \"__nc__\"` if you want this to stay explicit."
534 );
535 Some(quote! {
536 #[allow(dead_code)]
537 #[deprecated(note = #note)]
538 const #warn_ident: () = ();
539 const _: () = {
540 let _ = #warn_ident;
541 };
542 })
543 })
544 .collect();
545 let ids = build_monitored_ids(&task_ids, &mut culist_bridge_specs);
546 let parallel_rt_stage_entries = if distributed_selected {
547 match build_parallel_rt_stage_entries(
548 &culist_plan,
549 &culist_exec_entities,
550 &task_specs,
551 &culist_bridge_specs,
552 ) {
553 Ok(entries) => entries,
554 Err(e) => return return_error(e.to_string()),
555 }
556 } else {
557 Vec::new()
558 };
559 let parallel_rt_metadata_defs = if std {
560 Some(quote! {
561 pub const PARALLEL_RT_STAGES: &'static [cu29::parallel_rt::ParallelRtStageMetadata] =
562 &[#( #parallel_rt_stage_entries ),*];
563 pub const PARALLEL_RT_METADATA: cu29::parallel_rt::ParallelRtMetadata =
564 cu29::parallel_rt::ParallelRtMetadata::new(
565 PARALLEL_RT_STAGES,
566 #schedule_max_in_flight_tokens,
567 );
568 })
569 } else {
570 None
571 };
572 let monitored_component_entries: Vec<proc_macro2::TokenStream> = ids
573 .iter()
574 .enumerate()
575 .map(|(idx, id)| {
576 let id_lit = LitStr::new(id, Span::call_site());
577 if idx < task_specs.task_types.len() {
578 let task_ty = &task_specs.task_types[idx];
579 let component_type = match task_specs.cutypes[idx] {
580 CuTaskType::Source => quote! { cu29::monitoring::ComponentType::Source },
581 CuTaskType::Regular => quote! { cu29::monitoring::ComponentType::Task },
582 CuTaskType::Sink => quote! { cu29::monitoring::ComponentType::Sink },
583 };
584 quote! {
585 cu29::monitoring::MonitorComponentMetadata::new(
586 #id_lit,
587 #component_type,
588 Some(stringify!(#task_ty)),
589 )
590 }
591 } else {
592 quote! {
593 cu29::monitoring::MonitorComponentMetadata::new(
594 #id_lit,
595 cu29::monitoring::ComponentType::Bridge,
596 None,
597 )
598 }
599 }
600 })
601 .collect();
602 let culist_component_mapping = match build_monitor_culist_component_mapping(
603 &culist_plan,
604 &culist_exec_entities,
605 &culist_bridge_specs,
606 ) {
607 Ok(mapping) => mapping,
608 Err(e) => return return_error(e),
609 };
610
611 let runtime_task_types: Vec<Type> = (0..task_specs.ids.len())
612 .map(|index| runtime_task_type_for_index(&task_specs, graph, index, sim_mode))
613 .collect();
614
615 let task_reflect_read_arms: Vec<proc_macro2::TokenStream> = task_specs
616 .ids
617 .iter()
618 .enumerate()
619 .map(|(index, task_id)| {
620 let task_index = syn::Index::from(index);
621 let task_id_lit = LitStr::new(task_id, Span::call_site());
622 let value = distributed_selected.then(|| quote!(.value));
623 quote! {
624 #task_id_lit => Some(&self.copper_runtime.tasks.#task_index #value as &dyn cu29::reflect::Reflect),
625 }
626 })
627 .collect();
628
629 let task_reflect_write_arms: Vec<proc_macro2::TokenStream> = task_specs
630 .ids
631 .iter()
632 .enumerate()
633 .map(|(index, task_id)| {
634 let task_index = syn::Index::from(index);
635 let task_id_lit = LitStr::new(task_id, Span::call_site());
636 let value = distributed_selected.then(|| quote!(.value));
637 quote! {
638 #task_id_lit => Some(&mut self.copper_runtime.tasks.#task_index #value as &mut dyn cu29::reflect::Reflect),
639 }
640 })
641 .collect();
642
643 let task_debug_state_type_path_arms: Vec<proc_macro2::TokenStream> = task_specs
644 .ids
645 .iter()
646 .zip(runtime_task_types.iter())
647 .enumerate()
648 .map(|(index, (task_id, task_type))| {
649 let task_id_lit = LitStr::new(task_id, Span::call_site());
650 let task_trait = task_trait_for_specs(&task_specs, index);
651 quote! {
652 #task_id_lit => Some(<#task_type as #task_trait>::debug_state_type_path()),
653 }
654 })
655 .collect();
656
657 let task_debug_state_read_arms: Vec<proc_macro2::TokenStream> = task_specs
658 .ids
659 .iter()
660 .zip(runtime_task_types.iter())
661 .enumerate()
662 .map(|(index, (task_id, task_type))| {
663 let task_index = syn::Index::from(index);
664 let task_id_lit = LitStr::new(task_id, Span::call_site());
665 let task_trait = task_trait_for_specs(&task_specs, index);
666 let value = distributed_selected.then(|| quote!(.value));
667 quote! {
668 #task_id_lit => Some(
669 <#task_type as #task_trait>::with_debug_state(
670 &self.copper_runtime.tasks.#task_index #value,
671 f,
672 )
673 ),
674 }
675 })
676 .collect();
677
678 let task_debug_state_registration_calls: Vec<proc_macro2::TokenStream> = task_specs
679 .ids
680 .iter()
681 .enumerate()
682 .map(|(index, _)| {
683 let task_type = &runtime_task_types[index];
684 let task_trait = task_trait_for_specs(&task_specs, index);
685 quote! {
686 <#task_type as #task_trait>::register_debug_state_types(registry);
687 }
688 })
689 .collect();
690
691 let mut reflect_registry_types: BTreeMap<String, Type> = BTreeMap::new();
692 let mut add_reflect_type = |ty: Type| {
693 let key = quote! { #ty }.to_string();
694 reflect_registry_types.entry(key).or_insert(ty);
695 };
696
697 let mut sim_bridge_channel_decls = Vec::<proc_macro2::TokenStream>::new();
698 let bridge_runtime_types: Vec<Type> = culist_bridge_specs
699 .iter()
700 .map(|spec| {
701 if sim_mode && !spec.run_in_sim {
702 let (tx_set_ident, tx_id_ident, rx_set_ident, rx_id_ident) =
703 sim_bridge_channel_set_idents(spec.tuple_index);
704
705 if !spec.tx_channels.is_empty() {
706 let tx_entries = spec.tx_channels.iter().map(|channel| {
707 let entry_ident = Ident::new(
708 &channel.const_ident.to_string().to_lowercase(),
709 Span::call_site(),
710 );
711 let msg_type = &channel.msg_type;
712 quote! { #entry_ident => #msg_type, }
713 });
714 sim_bridge_channel_decls.push(quote! {
715 cu29::tx_channels! {
716 pub struct #tx_set_ident : #tx_id_ident {
717 #(#tx_entries)*
718 }
719 }
720 });
721 }
722
723 if !spec.rx_channels.is_empty() {
724 let rx_entries = spec.rx_channels.iter().map(|channel| {
725 let entry_ident = Ident::new(
726 &channel.const_ident.to_string().to_lowercase(),
727 Span::call_site(),
728 );
729 let msg_type = &channel.msg_type;
730 quote! { #entry_ident => #msg_type, }
731 });
732 sim_bridge_channel_decls.push(quote! {
733 cu29::rx_channels! {
734 pub struct #rx_set_ident : #rx_id_ident {
735 #(#rx_entries)*
736 }
737 }
738 });
739 }
740 }
741 runtime_bridge_type_for_spec(spec, sim_mode)
742 })
743 .collect();
744 let sim_bridge_channel_defs = quote! { #(#sim_bridge_channel_decls)* };
745
746 for (bridge_index, bridge_spec) in culist_bridge_specs.iter().enumerate() {
747 add_reflect_type(bridge_runtime_types[bridge_index].clone());
748 for channel in bridge_spec
749 .rx_channels
750 .iter()
751 .chain(bridge_spec.tx_channels.iter())
752 {
753 add_reflect_type(channel.msg_type.clone());
754 }
755 }
756
757 for output_pack in extract_output_packs(&culist_plan) {
758 for msg_type in output_pack.msg_types {
759 add_reflect_type(msg_type);
760 }
761 }
762
763 let reflect_type_registration_calls: Vec<proc_macro2::TokenStream> = reflect_registry_types
764 .values()
765 .map(|ty| {
766 quote! {
767 registry.register::<#ty>();
768 }
769 })
770 .collect();
771
772 let bridge_component_types: Vec<Type> = if distributed_selected {
773 bridge_runtime_types
774 .iter()
775 .map(|ty| parse_quote! { cu29::copperlist::ComponentRegion<#ty> })
776 .collect()
777 } else {
778 bridge_runtime_types.clone()
779 };
780 let bridges_type_tokens: proc_macro2::TokenStream = if bridge_component_types.is_empty() {
781 quote! { () }
782 } else {
783 let tuple: TypeTuple = parse_quote! { (#(#bridge_component_types),*,) };
784 quote! { #tuple }
785 };
786
787 let bridge_binding_idents: Vec<Ident> = culist_bridge_specs
788 .iter()
789 .enumerate()
790 .map(|(idx, _)| format_ident!("bridge_{idx}"))
791 .collect();
792
793 let bridge_init_statements: Vec<proc_macro2::TokenStream> = culist_bridge_specs
794 .iter()
795 .enumerate()
796 .map(|(idx, spec)| {
797 let binding_ident = &bridge_binding_idents[idx];
798 let bridge_mapping_ref = bridge_resource_mappings.refs[idx].clone();
799 let bridge_type = &bridge_runtime_types[idx];
800 let bridge_name = spec.id.clone();
801 let config_index = syn::Index::from(spec.config_index);
802 let binding_error = LitStr::new(
803 &format!("Failed to bind resources for bridge '{}'", bridge_name),
804 Span::call_site(),
805 );
806 let tx_configs: Vec<proc_macro2::TokenStream> = spec
807 .tx_channels
808 .iter()
809 .map(|channel| {
810 let const_ident = &channel.const_ident;
811 let channel_name = channel.id.clone();
812 let channel_config_index = syn::Index::from(channel.config_index);
813 quote! {
814 {
815 let (channel_route, channel_config) = match &bridge_cfg.channels[#channel_config_index] {
816 cu29::config::BridgeChannelConfigRepresentation::Tx { route, config, .. } => {
817 (route.clone(), config.clone())
818 }
819 _ => panic!(
820 "Bridge '{}' channel '{}' expected to be Tx",
821 #bridge_name,
822 #channel_name
823 ),
824 };
825 cu29::cubridge::BridgeChannelConfig::from_static(
826 &<#bridge_type as cu29::cubridge::CuBridge>::Tx::#const_ident,
827 channel_route,
828 channel_config,
829 )
830 }
831 }
832 })
833 .collect();
834 let rx_configs: Vec<proc_macro2::TokenStream> = spec
835 .rx_channels
836 .iter()
837 .map(|channel| {
838 let const_ident = &channel.const_ident;
839 let channel_name = channel.id.clone();
840 let channel_config_index = syn::Index::from(channel.config_index);
841 quote! {
842 {
843 let (channel_route, channel_config) = match &bridge_cfg.channels[#channel_config_index] {
844 cu29::config::BridgeChannelConfigRepresentation::Rx { route, config, .. } => {
845 (route.clone(), config.clone())
846 }
847 _ => panic!(
848 "Bridge '{}' channel '{}' expected to be Rx",
849 #bridge_name,
850 #channel_name
851 ),
852 };
853 cu29::cubridge::BridgeChannelConfig::from_static(
854 &<#bridge_type as cu29::cubridge::CuBridge>::Rx::#const_ident,
855 channel_route,
856 channel_config,
857 )
858 }
859 }
860 })
861 .collect();
862 quote! {
863 let #binding_ident = {
864 let bridge_cfg = config
865 .bridges
866 .get(#config_index)
867 .unwrap_or_else(|| panic!("Bridge '{}' missing from configuration", #bridge_name));
868 let bridge_mapping = #bridge_mapping_ref;
869 let bridge_resources = <<#bridge_type as cu29::cubridge::CuBridge>::Resources<'_> as ResourceBindings>::from_bindings(
870 resources,
871 bridge_mapping,
872 )
873 .map_err(|e| cu29::CuError::new_with_cause(#binding_error, e))?;
874 let tx_channels: &[cu29::cubridge::BridgeChannelConfig<
875 <<#bridge_type as cu29::cubridge::CuBridge>::Tx as cu29::cubridge::BridgeChannelSet>::Id,
876 >] = &[#(#tx_configs),*];
877 let rx_channels: &[cu29::cubridge::BridgeChannelConfig<
878 <<#bridge_type as cu29::cubridge::CuBridge>::Rx as cu29::cubridge::BridgeChannelSet>::Id,
879 >] = &[#(#rx_configs),*];
880 <#bridge_type as cu29::cubridge::CuBridge>::new(
881 bridge_cfg.config.as_ref(),
882 tx_channels,
883 rx_channels,
884 bridge_resources,
885 )?
886 };
887 }
888 })
889 .collect();
890
891 let bridges_instanciator = if culist_bridge_specs.is_empty() {
892 quote! {
893 pub fn bridges_instanciator(_config: &CuConfig, resources: &mut ResourceManager) -> CuResult<CuBridges> {
894 let _ = resources;
895 Ok(())
896 }
897 }
898 } else {
899 let bridge_bindings = bridge_binding_idents.iter().map(|binding| {
900 if distributed_selected {
901 quote! { cu29::copperlist::ComponentRegion::new(#binding) }
902 } else {
903 quote! { #binding }
904 }
905 });
906 quote! {
907 pub fn bridges_instanciator(config: &CuConfig, resources: &mut ResourceManager) -> CuResult<CuBridges> {
908 #(#bridge_init_statements)*
909 Ok((#(#bridge_bindings),*,))
910 }
911 }
912 };
913
914 let all_sim_tasks_types = runtime_task_types.clone();
915
916 #[cfg(feature = "macro_debug")]
917 eprintln!("[build task tuples]");
918
919 let task_types: Vec<Type> = if distributed_selected {
920 task_specs
921 .task_types
922 .iter()
923 .map(|ty| parse_quote! { cu29::copperlist::ComponentRegion<#ty> })
924 .collect()
925 } else {
926 task_specs.task_types.clone()
927 };
928 let task_types_tuple: TypeTuple = if task_types.is_empty() {
931 parse_quote! { () }
932 } else {
933 parse_quote! { (#(#task_types),*,) }
934 };
935
936 let task_types_tuple_sim: TypeTuple = if all_sim_tasks_types.is_empty() {
937 parse_quote! { () }
938 } else {
939 parse_quote! { (#(#all_sim_tasks_types),*,) }
940 };
941
942 #[cfg(feature = "macro_debug")]
943 eprintln!("[gen instances]");
944
945 let thread_pool_indices: HashMap<&str, usize> = copper_config
950 .runtime
951 .as_ref()
952 .map(|runtime| {
953 runtime
954 .thread_pools
955 .iter()
956 .enumerate()
957 .map(|(index, pool)| (pool.id.as_str(), index))
958 .collect()
959 })
960 .unwrap_or_default();
961 for (task_index, pool_name) in task_specs.background_pools.iter().enumerate() {
962 if !task_specs.background_flags[task_index] {
963 continue;
964 }
965 if pool_name == RT_POOL {
971 return return_error(format!(
972 "Background task '{}' may not use the reserved '{RT_POOL}' thread pool; it is dedicated to the parallel-rt execution engine.",
973 task_specs.ids[task_index]
974 ));
975 }
976 if !thread_pool_indices.contains_key(pool_name.as_str()) {
977 return return_error(format!(
978 "Background task '{}' references undefined thread pool '{}'. Define it under runtime.thread_pools.",
979 task_specs.ids[task_index], pool_name
980 ));
981 }
982 }
983 let task_pool_indices: Vec<usize> = task_specs
984 .background_pools
985 .iter()
986 .map(|pool_name| {
987 thread_pool_indices
988 .get(pool_name.as_str())
989 .copied()
990 .unwrap_or(0)
991 })
992 .collect();
993
994 let task_sim_instances_init_code = all_sim_tasks_types
995 .iter()
996 .enumerate()
997 .map(|(index, ty)| {
998 let additional_error_info = format!(
999 "Failed to get create instance for {}, instance index {}.",
1000 task_specs.type_names[index], index
1001 );
1002 let mapping_ref = task_resource_mappings.refs[index].clone();
1003 let background = task_specs.background_flags[index]
1004 && !(sim_mode
1005 && task_specs.cutypes[index] == CuTaskType::Source
1006 && !task_specs.run_in_sim_flags[index]);
1007 let inner_task_type = &task_specs.async_inner_task_types[index];
1008 match task_specs.cutypes[index] {
1009 CuTaskType::Source => {
1010 if background {
1011 let pool_index = task_pool_indices[index];
1012 let pool_name = task_specs.background_pools[index].clone();
1013 quote! {
1014 {
1015 let inner_resources = <<#inner_task_type as CuSrcTask>::Resources<'_> as ResourceBindings>::from_bindings(
1016 resources,
1017 #mapping_ref,
1018 ).map_err(|e| e.add_cause(#additional_error_info))?;
1019 let threadpool = thread_pools
1020 .get(#pool_index)
1021 .and_then(|slot| slot.clone())
1022 .ok_or_else(|| CuError::from(format!(
1023 "Background task at index {} requested thread pool '{}' but it was not provided",
1024 #index, #pool_name,
1025 )))?;
1026 let resources = cu29::cuasynctask::CuAsyncSrcTaskResources {
1027 inner: inner_resources,
1028 threadpool,
1029 };
1030 <#ty as CuSrcTask>::new(all_instances_configs[#index], resources)
1031 .map_err(|e| e.add_cause(#additional_error_info))?
1032 }
1033 }
1034 } else {
1035 quote! {
1036 {
1037 let resources = <<#ty as CuSrcTask>::Resources<'_> as ResourceBindings>::from_bindings(
1038 resources,
1039 #mapping_ref,
1040 ).map_err(|e| e.add_cause(#additional_error_info))?;
1041 <#ty as CuSrcTask>::new(all_instances_configs[#index], resources)
1042 .map_err(|e| e.add_cause(#additional_error_info))?
1043 }
1044 }
1045 }
1046 }
1047 CuTaskType::Regular => {
1048 if background {
1049 let pool_index = task_pool_indices[index];
1050 let pool_name = task_specs.background_pools[index].clone();
1051 quote! {
1052 {
1053 let inner_resources = <<#inner_task_type as CuTask>::Resources<'_> as ResourceBindings>::from_bindings(
1054 resources,
1055 #mapping_ref,
1056 ).map_err(|e| e.add_cause(#additional_error_info))?;
1057 let threadpool = thread_pools
1058 .get(#pool_index)
1059 .and_then(|slot| slot.clone())
1060 .ok_or_else(|| CuError::from(format!(
1061 "Background task at index {} requested thread pool '{}' but it was not provided",
1062 #index, #pool_name,
1063 )))?;
1064 let resources = cu29::cuasynctask::CuAsyncTaskResources {
1065 inner: inner_resources,
1066 threadpool,
1067 };
1068 <#ty as CuTask>::new(all_instances_configs[#index], resources)
1069 .map_err(|e| e.add_cause(#additional_error_info))?
1070 }
1071 }
1072 } else {
1073 let regular_trait = task_trait_for_specs(&task_specs, index);
1074 quote! {
1075 {
1076 let resources = <<#ty as #regular_trait>::Resources<'_> as ResourceBindings>::from_bindings(
1077 resources,
1078 #mapping_ref,
1079 ).map_err(|e| e.add_cause(#additional_error_info))?;
1080 <#ty as #regular_trait>::new(all_instances_configs[#index], resources)
1081 .map_err(|e| e.add_cause(#additional_error_info))?
1082 }
1083 }
1084 }
1085 }
1086 CuTaskType::Sink => quote! {
1087 {
1088 let resources = <<#ty as CuSinkTask>::Resources<'_> as ResourceBindings>::from_bindings(
1089 resources,
1090 #mapping_ref,
1091 ).map_err(|e| e.add_cause(#additional_error_info))?;
1092 <#ty as CuSinkTask>::new(all_instances_configs[#index], resources)
1093 .map_err(|e| e.add_cause(#additional_error_info))?
1094 }
1095 },
1096 }
1097 })
1098 .collect::<Vec<_>>();
1099
1100 let task_instances_init_code = task_specs
1101 .instantiation_types
1102 .iter()
1103 .zip(&task_specs.background_flags)
1104 .enumerate()
1105 .map(|(index, (task_type, background))| {
1106 let additional_error_info = format!(
1107 "Failed to get create instance for {}, instance index {}.",
1108 task_specs.type_names[index], index
1109 );
1110 let mapping_ref = task_resource_mappings.refs[index].clone();
1111 let inner_task_type = &task_specs.async_inner_task_types[index];
1112 match task_specs.cutypes[index] {
1113 CuTaskType::Source => {
1114 if *background {
1115 let pool_index = task_pool_indices[index];
1116 let pool_name = task_specs.background_pools[index].clone();
1117 quote! {
1118 {
1119 let inner_resources = <<#inner_task_type as CuSrcTask>::Resources<'_> as ResourceBindings>::from_bindings(
1120 resources,
1121 #mapping_ref,
1122 ).map_err(|e| e.add_cause(#additional_error_info))?;
1123 let threadpool = thread_pools
1124 .get(#pool_index)
1125 .and_then(|slot| slot.clone())
1126 .ok_or_else(|| CuError::from(format!(
1127 "Background task at index {} requested thread pool '{}' but it was not provided",
1128 #index, #pool_name,
1129 )))?;
1130 let resources = cu29::cuasynctask::CuAsyncSrcTaskResources {
1131 inner: inner_resources,
1132 threadpool,
1133 };
1134 <#task_type as CuSrcTask>::new(all_instances_configs[#index], resources)
1135 .map_err(|e| e.add_cause(#additional_error_info))?
1136 }
1137 }
1138 } else {
1139 quote! {
1140 {
1141 let resources = <<#task_type as CuSrcTask>::Resources<'_> as ResourceBindings>::from_bindings(
1142 resources,
1143 #mapping_ref,
1144 ).map_err(|e| e.add_cause(#additional_error_info))?;
1145 <#task_type as CuSrcTask>::new(all_instances_configs[#index], resources)
1146 .map_err(|e| e.add_cause(#additional_error_info))?
1147 }
1148 }
1149 }
1150 }
1151 CuTaskType::Regular => {
1152 if *background {
1153 let pool_index = task_pool_indices[index];
1154 let pool_name = task_specs.background_pools[index].clone();
1155 quote! {
1156 {
1157 let inner_resources = <<#inner_task_type as CuTask>::Resources<'_> as ResourceBindings>::from_bindings(
1158 resources,
1159 #mapping_ref,
1160 ).map_err(|e| e.add_cause(#additional_error_info))?;
1161 let threadpool = thread_pools
1162 .get(#pool_index)
1163 .and_then(|slot| slot.clone())
1164 .ok_or_else(|| CuError::from(format!(
1165 "Background task at index {} requested thread pool '{}' but it was not provided",
1166 #index, #pool_name,
1167 )))?;
1168 let resources = cu29::cuasynctask::CuAsyncTaskResources {
1169 inner: inner_resources,
1170 threadpool,
1171 };
1172 <#task_type as CuTask>::new(all_instances_configs[#index], resources)
1173 .map_err(|e| e.add_cause(#additional_error_info))?
1174 }
1175 }
1176 } else {
1177 let regular_trait = task_trait_for_specs(&task_specs, index);
1178 quote! {
1179 {
1180 let resources = <<#task_type as #regular_trait>::Resources<'_> as ResourceBindings>::from_bindings(
1181 resources,
1182 #mapping_ref,
1183 ).map_err(|e| e.add_cause(#additional_error_info))?;
1184 <#task_type as #regular_trait>::new(all_instances_configs[#index], resources)
1185 .map_err(|e| e.add_cause(#additional_error_info))?
1186 }
1187 }
1188 }
1189 }
1190 CuTaskType::Sink => quote! {
1191 {
1192 let resources = <<#task_type as CuSinkTask>::Resources<'_> as ResourceBindings>::from_bindings(
1193 resources,
1194 #mapping_ref,
1195 ).map_err(|e| e.add_cause(#additional_error_info))?;
1196 <#task_type as CuSinkTask>::new(all_instances_configs[#index], resources)
1197 .map_err(|e| e.add_cause(#additional_error_info))?
1198 }
1199 },
1200 }
1201 })
1202 .collect::<Vec<_>>();
1203 let task_instances_init_code: Vec<_> = if distributed_selected {
1204 task_instances_init_code
1205 .into_iter()
1206 .map(|instance| quote! { cu29::copperlist::ComponentRegion::new(#instance) })
1207 .collect()
1208 } else {
1209 task_instances_init_code
1210 };
1211
1212 let mut keyframe_task_restore_order = Vec::new();
1213 for unit in &culist_plan.steps {
1214 let CuExecutionUnit::Step(step) = unit else {
1215 panic!("Execution loops are not supported in runtime generation");
1216 };
1217 let ExecutionEntityKind::Task { task_index } =
1218 &culist_exec_entities[step.node_id as usize].kind
1219 else {
1220 continue;
1221 };
1222 if !keyframe_task_restore_order.contains(task_index) {
1223 keyframe_task_restore_order.push(*task_index);
1224 }
1225 }
1226 if keyframe_task_restore_order.len() != task_specs.task_types.len() {
1227 return return_error(format!(
1228 "Keyframe restore order covers {} task steps but mission declares {} tasks",
1229 keyframe_task_restore_order.len(),
1230 task_specs.task_types.len()
1231 ));
1232 }
1233 let (
1236 task_start_calls,
1237 task_stop_calls,
1238 task_preprocess_calls,
1239 task_postprocess_calls,
1240 ): (Vec<_>, Vec<_>, Vec<_>, Vec<_>) = itertools::multiunzip(
1241 (0..task_specs.task_types.len())
1242 .map(|index| {
1243 let task_index = int2sliceindex(index as u32);
1244 let task_enum_name = config_id_to_enum(&task_specs.ids[index]);
1245 let enum_name = Ident::new(&task_enum_name, Span::call_site());
1246 (
1247 { let monitoring_action = quote! {
1249 let decision = self.copper_runtime.monitor.process_error(cu29::monitoring::ComponentId::new(#index), CuComponentState::Start, &error);
1250 match decision {
1251 Decision::Abort => {
1252 debug!(ctx, "Start: ABORT decision from monitoring. Component '{}' errored out \
1253 during start. Aborting all the other starts.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#index)));
1254 return Ok(());
1255
1256 }
1257 Decision::Ignore => {
1258 debug!(ctx, "Start: IGNORE decision from monitoring. Component '{}' errored out \
1259 during start. The runtime will continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#index)));
1260 }
1261 Decision::Shutdown => {
1262 debug!(ctx, "Start: SHUTDOWN decision from monitoring. Component '{}' errored out \
1263 during start. The runtime cannot continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#index)));
1264 return Err(CuError::new_with_cause("Component errored out during start.", error));
1265 }
1266 }
1267 };
1268
1269 let call_sim_callback = if sim_mode {
1270 quote! {
1271 let ovr = sim_callback(SimStep::#enum_name(CuTaskCallbackState::Start));
1273
1274 let doit = if let SimOverride::Errored(reason) = ovr {
1275 let error: CuError = reason.into();
1276 #monitoring_action
1277 false
1278 }
1279 else {
1280 ovr == SimOverride::ExecuteByRuntime
1281 };
1282 }
1283 } else {
1284 quote! {
1285 let doit = true; }
1287 };
1288
1289
1290 let alloc_open = alloc_scope_open_tokens();
1291 let alloc_close = alloc_scope_close_tokens(
1292 quote! { self.copper_runtime.monitor },
1293 quote! { #index },
1294 quote! { CuComponentState::Start },
1295 );
1296 quote! {
1297 #call_sim_callback
1298 if doit {
1299 self.copper_runtime.record_execution_marker(
1300 cu29::monitoring::ExecutionMarker {
1301 component_id: cu29::monitoring::ComponentId::new(#index),
1302 step: CuComponentState::Start,
1303 culistid: None,
1304 }
1305 );
1306 let task = &mut self.copper_runtime.tasks.#task_index;
1307 ctx.set_current_task(#index);
1308 #alloc_open
1309 let __cu_step_result = task.start(&ctx);
1310 #alloc_close
1311 if let Err(error) = __cu_step_result {
1312 #monitoring_action
1313 }
1314 }
1315 }
1316 },
1317 { let monitoring_action = quote! {
1319 let decision = self.copper_runtime.monitor.process_error(cu29::monitoring::ComponentId::new(#index), CuComponentState::Stop, &error);
1320 match decision {
1321 Decision::Abort => {
1322 debug!(ctx, "Stop: ABORT decision from monitoring. Component '{}' errored out \
1323 during stop. Aborting all the other starts.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#index)));
1324 return Ok(());
1325
1326 }
1327 Decision::Ignore => {
1328 debug!(ctx, "Stop: IGNORE decision from monitoring. Component '{}' errored out \
1329 during stop. The runtime will continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#index)));
1330 }
1331 Decision::Shutdown => {
1332 debug!(ctx, "Stop: SHUTDOWN decision from monitoring. Component '{}' errored out \
1333 during stop. The runtime cannot continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#index)));
1334 return Err(CuError::new_with_cause("Component errored out during stop.", error));
1335 }
1336 }
1337 };
1338 let call_sim_callback = if sim_mode {
1339 quote! {
1340 let ovr = sim_callback(SimStep::#enum_name(CuTaskCallbackState::Stop));
1342
1343 let doit = if let SimOverride::Errored(reason) = ovr {
1344 let error: CuError = reason.into();
1345 #monitoring_action
1346 false
1347 }
1348 else {
1349 ovr == SimOverride::ExecuteByRuntime
1350 };
1351 }
1352 } else {
1353 quote! {
1354 let doit = true; }
1356 };
1357 let alloc_open = alloc_scope_open_tokens();
1358 let alloc_close = alloc_scope_close_tokens(
1359 quote! { self.copper_runtime.monitor },
1360 quote! { #index },
1361 quote! { CuComponentState::Stop },
1362 );
1363 quote! {
1364 #call_sim_callback
1365 if doit {
1366 self.copper_runtime.record_execution_marker(
1367 cu29::monitoring::ExecutionMarker {
1368 component_id: cu29::monitoring::ComponentId::new(#index),
1369 step: CuComponentState::Stop,
1370 culistid: None,
1371 }
1372 );
1373 let task = &mut self.copper_runtime.tasks.#task_index;
1374 ctx.set_current_task(#index);
1375 #alloc_open
1376 let __cu_step_result = task.stop(&ctx);
1377 #alloc_close
1378 if let Err(error) = __cu_step_result {
1379 #monitoring_action
1380 }
1381 }
1382 }
1383 },
1384 { let monitoring_action = quote! {
1386 let decision = monitor.process_error(cu29::monitoring::ComponentId::new(#index), CuComponentState::Preprocess, &error);
1387 match decision {
1388 Decision::Abort => {
1389 debug!(ctx, "Preprocess: ABORT decision from monitoring. Component '{}' errored out \
1390 during preprocess. Aborting all the other starts.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#index)));
1391 return Ok(());
1392
1393 }
1394 Decision::Ignore => {
1395 debug!(ctx, "Preprocess: IGNORE decision from monitoring. Component '{}' errored out \
1396 during preprocess. The runtime will continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#index)));
1397 }
1398 Decision::Shutdown => {
1399 debug!(ctx, "Preprocess: SHUTDOWN decision from monitoring. Component '{}' errored out \
1400 during preprocess. The runtime cannot continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#index)));
1401 return Err(CuError::new_with_cause("Component errored out during preprocess.", error));
1402 }
1403 }
1404 };
1405 let call_sim_callback = if sim_mode {
1406 quote! {
1407 let ovr = sim_callback(SimStep::#enum_name(CuTaskCallbackState::Preprocess));
1409
1410 let doit = if let SimOverride::Errored(reason) = ovr {
1411 let error: CuError = reason.into();
1412 #monitoring_action
1413 false
1414 } else {
1415 ovr == SimOverride::ExecuteByRuntime
1416 };
1417 }
1418 } else {
1419 quote! {
1420 let doit = true; }
1422 };
1423 let alloc_open = alloc_scope_open_tokens();
1424 let alloc_close = alloc_scope_close_tokens(
1425 quote! { monitor },
1426 quote! { #index },
1427 quote! { CuComponentState::Preprocess },
1428 );
1429 quote! {
1430 #call_sim_callback
1431 if doit {
1432 execution_probe.record(cu29::monitoring::ExecutionMarker {
1433 component_id: cu29::monitoring::ComponentId::new(#index),
1434 step: CuComponentState::Preprocess,
1435 culistid: None,
1436 });
1437 ctx.set_current_task(#index);
1438 #alloc_open
1439 let maybe_error = {
1440 #rt_guard
1441 tasks.#task_index.preprocess(&ctx)
1442 };
1443 #alloc_close
1444 if let Err(error) = maybe_error {
1445 #monitoring_action
1446 }
1447 }
1448 }
1449 },
1450 { let monitoring_action = quote! {
1452 let decision = monitor.process_error(cu29::monitoring::ComponentId::new(#index), CuComponentState::Postprocess, &error);
1453 match decision {
1454 Decision::Abort => {
1455 debug!(ctx, "Postprocess: ABORT decision from monitoring. Component '{}' errored out \
1456 during postprocess. Aborting all the other starts.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#index)));
1457 return Ok(());
1458
1459 }
1460 Decision::Ignore => {
1461 debug!(ctx, "Postprocess: IGNORE decision from monitoring. Component '{}' errored out \
1462 during postprocess. The runtime will continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#index)));
1463 }
1464 Decision::Shutdown => {
1465 debug!(ctx, "Postprocess: SHUTDOWN decision from monitoring. Component '{}' errored out \
1466 during postprocess. The runtime cannot continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#index)));
1467 return Err(CuError::new_with_cause("Component errored out during postprocess.", error));
1468 }
1469 }
1470 };
1471 let call_sim_callback = if sim_mode {
1472 quote! {
1473 let ovr = sim_callback(SimStep::#enum_name(CuTaskCallbackState::Postprocess));
1475
1476 let doit = if let SimOverride::Errored(reason) = ovr {
1477 let error: CuError = reason.into();
1478 #monitoring_action
1479 false
1480 } else {
1481 ovr == SimOverride::ExecuteByRuntime
1482 };
1483 }
1484 } else {
1485 quote! {
1486 let doit = true; }
1488 };
1489 let alloc_open = alloc_scope_open_tokens();
1490 let alloc_close = alloc_scope_close_tokens(
1491 quote! { monitor },
1492 quote! { #index },
1493 quote! { CuComponentState::Postprocess },
1494 );
1495 quote! {
1496 #call_sim_callback
1497 if doit {
1498 execution_probe.record(cu29::monitoring::ExecutionMarker {
1499 component_id: cu29::monitoring::ComponentId::new(#index),
1500 step: CuComponentState::Postprocess,
1501 culistid: None,
1502 });
1503 ctx.set_current_task(#index);
1504 #alloc_open
1505 let maybe_error = {
1506 #rt_guard
1507 tasks.#task_index.postprocess(&ctx)
1508 };
1509 #alloc_close
1510 if let Err(error) = maybe_error {
1511 #monitoring_action
1512 }
1513 }
1514 }
1515 }
1516 )
1517 })
1518 );
1519
1520 let bridge_start_calls: Vec<proc_macro2::TokenStream> = culist_bridge_specs
1521 .iter()
1522 .map(|spec| {
1523 let bridge_index = int2sliceindex(spec.tuple_index as u32);
1524 let monitor_index = syn::Index::from(
1525 spec.monitor_index
1526 .expect("Bridge missing monitor index for start"),
1527 );
1528 let enum_ident = Ident::new(
1529 &config_id_to_enum(&format!("{}_bridge", spec.id)),
1530 Span::call_site(),
1531 );
1532 let call_sim = if sim_mode {
1533 quote! {
1534 let doit = {
1535 let state = SimStep::#enum_ident(cu29::simulation::CuBridgeLifecycleState::Start);
1536 let ovr = sim_callback(state);
1537 if let SimOverride::Errored(reason) = ovr {
1538 let error: CuError = reason.into();
1539 let decision = self.copper_runtime.monitor.process_error(cu29::monitoring::ComponentId::new(#monitor_index), CuComponentState::Start, &error);
1540 match decision {
1541 Decision::Abort => { debug!(ctx, "Start: ABORT decision from monitoring. Component '{}' errored out during start. Aborting all the other starts.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index))); return Ok(()); }
1542 Decision::Ignore => { debug!(ctx, "Start: IGNORE decision from monitoring. Component '{}' errored out during start. The runtime will continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index))); false }
1543 Decision::Shutdown => { debug!(ctx, "Start: SHUTDOWN decision from monitoring. Component '{}' errored out during start. The runtime cannot continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index))); return Err(CuError::new_with_cause("Component errored out during start.", error)); }
1544 }
1545 } else {
1546 ovr == SimOverride::ExecuteByRuntime
1547 }
1548 };
1549 }
1550 } else {
1551 quote! { let doit = true; }
1552 };
1553 let alloc_open = alloc_scope_open_tokens();
1554 let alloc_close = alloc_scope_close_tokens(
1555 quote! { self.copper_runtime.monitor },
1556 quote! { #monitor_index },
1557 quote! { CuComponentState::Start },
1558 );
1559 quote! {
1560 {
1561 #call_sim
1562 if !doit { return Ok(()); }
1563 self.copper_runtime.record_execution_marker(
1564 cu29::monitoring::ExecutionMarker {
1565 component_id: cu29::monitoring::ComponentId::new(#monitor_index),
1566 step: CuComponentState::Start,
1567 culistid: None,
1568 }
1569 );
1570 ctx.set_current_component(#monitor_index);
1571 ctx.clear_current_task();
1572 let bridge = &mut self.copper_runtime.bridges.#bridge_index;
1573 #alloc_open
1574 let __cu_step_result = bridge.start(&ctx);
1575 #alloc_close
1576 if let Err(error) = __cu_step_result {
1577 let decision = self.copper_runtime.monitor.process_error(cu29::monitoring::ComponentId::new(#monitor_index), CuComponentState::Start, &error);
1578 match decision {
1579 Decision::Abort => {
1580 debug!(ctx, "Start: ABORT decision from monitoring. Component '{}' errored out during start. Aborting all the other starts.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index)));
1581 return Ok(());
1582 }
1583 Decision::Ignore => {
1584 debug!(ctx, "Start: IGNORE decision from monitoring. Component '{}' errored out during start. The runtime will continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index)));
1585 }
1586 Decision::Shutdown => {
1587 debug!(ctx, "Start: SHUTDOWN decision from monitoring. Component '{}' errored out during start. The runtime cannot continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index)));
1588 return Err(CuError::new_with_cause("Component errored out during start.", error));
1589 }
1590 }
1591 }
1592 }
1593 }
1594 })
1595 .collect();
1596
1597 let bridge_stop_calls: Vec<proc_macro2::TokenStream> = culist_bridge_specs
1598 .iter()
1599 .map(|spec| {
1600 let bridge_index = int2sliceindex(spec.tuple_index as u32);
1601 let monitor_index = syn::Index::from(
1602 spec.monitor_index
1603 .expect("Bridge missing monitor index for stop"),
1604 );
1605 let enum_ident = Ident::new(
1606 &config_id_to_enum(&format!("{}_bridge", spec.id)),
1607 Span::call_site(),
1608 );
1609 let call_sim = if sim_mode {
1610 quote! {
1611 let doit = {
1612 let state = SimStep::#enum_ident(cu29::simulation::CuBridgeLifecycleState::Stop);
1613 let ovr = sim_callback(state);
1614 if let SimOverride::Errored(reason) = ovr {
1615 let error: CuError = reason.into();
1616 let decision = self.copper_runtime.monitor.process_error(cu29::monitoring::ComponentId::new(#monitor_index), CuComponentState::Stop, &error);
1617 match decision {
1618 Decision::Abort => { debug!(ctx, "Stop: ABORT decision from monitoring. Component '{}' errored out during stop. Aborting all the other stops.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index))); return Ok(()); }
1619 Decision::Ignore => { debug!(ctx, "Stop: IGNORE decision from monitoring. Component '{}' errored out during stop. The runtime will continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index))); false }
1620 Decision::Shutdown => { debug!(ctx, "Stop: SHUTDOWN decision from monitoring. Component '{}' errored out during stop. The runtime cannot continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index))); return Err(CuError::new_with_cause("Component errored out during stop.", error)); }
1621 }
1622 } else {
1623 ovr == SimOverride::ExecuteByRuntime
1624 }
1625 };
1626 }
1627 } else {
1628 quote! { let doit = true; }
1629 };
1630 let alloc_open = alloc_scope_open_tokens();
1631 let alloc_close = alloc_scope_close_tokens(
1632 quote! { self.copper_runtime.monitor },
1633 quote! { #monitor_index },
1634 quote! { CuComponentState::Stop },
1635 );
1636 quote! {
1637 {
1638 #call_sim
1639 if !doit { return Ok(()); }
1640 self.copper_runtime.record_execution_marker(
1641 cu29::monitoring::ExecutionMarker {
1642 component_id: cu29::monitoring::ComponentId::new(#monitor_index),
1643 step: CuComponentState::Stop,
1644 culistid: None,
1645 }
1646 );
1647 ctx.set_current_component(#monitor_index);
1648 ctx.clear_current_task();
1649 let bridge = &mut self.copper_runtime.bridges.#bridge_index;
1650 #alloc_open
1651 let __cu_step_result = bridge.stop(&ctx);
1652 #alloc_close
1653 if let Err(error) = __cu_step_result {
1654 let decision = self.copper_runtime.monitor.process_error(cu29::monitoring::ComponentId::new(#monitor_index), CuComponentState::Stop, &error);
1655 match decision {
1656 Decision::Abort => {
1657 debug!(ctx, "Stop: ABORT decision from monitoring. Component '{}' errored out during stop. Aborting all the other stops.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index)));
1658 return Ok(());
1659 }
1660 Decision::Ignore => {
1661 debug!(ctx, "Stop: IGNORE decision from monitoring. Component '{}' errored out during stop. The runtime will continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index)));
1662 }
1663 Decision::Shutdown => {
1664 debug!(ctx, "Stop: SHUTDOWN decision from monitoring. Component '{}' errored out during stop. The runtime cannot continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index)));
1665 return Err(CuError::new_with_cause("Component errored out during stop.", error));
1666 }
1667 }
1668 }
1669 }
1670 }
1671 })
1672 .collect();
1673
1674 let bridge_preprocess_calls: Vec<proc_macro2::TokenStream> = culist_bridge_specs
1675 .iter()
1676 .map(|spec| {
1677 let bridge_index = int2sliceindex(spec.tuple_index as u32);
1678 let monitor_index = syn::Index::from(
1679 spec.monitor_index
1680 .expect("Bridge missing monitor index for preprocess"),
1681 );
1682 let enum_ident = Ident::new(
1683 &config_id_to_enum(&format!("{}_bridge", spec.id)),
1684 Span::call_site(),
1685 );
1686 let call_sim = if sim_mode {
1687 quote! {
1688 let doit = {
1689 let state = SimStep::#enum_ident(cu29::simulation::CuBridgeLifecycleState::Preprocess);
1690 let ovr = sim_callback(state);
1691 if let SimOverride::Errored(reason) = ovr {
1692 let error: CuError = reason.into();
1693 let decision = monitor.process_error(cu29::monitoring::ComponentId::new(#monitor_index), CuComponentState::Preprocess, &error);
1694 match decision {
1695 Decision::Abort => { debug!(ctx, "Preprocess: ABORT decision from monitoring. Component '{}' errored out during preprocess. Aborting all the other starts.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index))); return Ok(()); }
1696 Decision::Ignore => { debug!(ctx, "Preprocess: IGNORE decision from monitoring. Component '{}' errored out during preprocess. The runtime will continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index))); false }
1697 Decision::Shutdown => { debug!(ctx, "Preprocess: SHUTDOWN decision from monitoring. Component '{}' errored out during preprocess. The runtime cannot continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index))); return Err(CuError::new_with_cause("Component errored out during preprocess.", error)); }
1698 }
1699 } else {
1700 ovr == SimOverride::ExecuteByRuntime
1701 }
1702 };
1703 }
1704 } else {
1705 quote! { let doit = true; }
1706 };
1707 let alloc_open = alloc_scope_open_tokens();
1708 let alloc_close = alloc_scope_close_tokens(
1709 quote! { monitor },
1710 quote! { #monitor_index },
1711 quote! { CuComponentState::Preprocess },
1712 );
1713 quote! {
1714 {
1715 #call_sim
1716 if doit {
1717 ctx.set_current_component(#monitor_index);
1718 ctx.clear_current_task();
1719 let bridge = &mut __cu_bridges.#bridge_index;
1720 execution_probe.record(cu29::monitoring::ExecutionMarker {
1721 component_id: cu29::monitoring::ComponentId::new(#monitor_index),
1722 step: CuComponentState::Preprocess,
1723 culistid: None,
1724 });
1725 #alloc_open
1726 let maybe_error = {
1727 #rt_guard
1728 bridge.preprocess(&ctx)
1729 };
1730 #alloc_close
1731 if let Err(error) = maybe_error {
1732 let decision = monitor.process_error(cu29::monitoring::ComponentId::new(#monitor_index), CuComponentState::Preprocess, &error);
1733 match decision {
1734 Decision::Abort => {
1735 debug!(ctx, "Preprocess: ABORT decision from monitoring. Component '{}' errored out during preprocess. Aborting all the other starts.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index)));
1736 return Ok(());
1737 }
1738 Decision::Ignore => {
1739 debug!(ctx, "Preprocess: IGNORE decision from monitoring. Component '{}' errored out during preprocess. The runtime will continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index)));
1740 }
1741 Decision::Shutdown => {
1742 debug!(ctx, "Preprocess: SHUTDOWN decision from monitoring. Component '{}' errored out during preprocess. The runtime cannot continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index)));
1743 return Err(CuError::new_with_cause("Component errored out during preprocess.", error));
1744 }
1745 }
1746 }
1747 }
1748 }
1749 }
1750 })
1751 .collect();
1752
1753 let bridge_postprocess_calls: Vec<proc_macro2::TokenStream> = culist_bridge_specs
1754 .iter()
1755 .map(|spec| {
1756 let freeze_bridge = keyframe_freeze_bridge_tokens(keyframe_logging_enabled);
1757 let bridge_index = int2sliceindex(spec.tuple_index as u32);
1758 let monitor_index = syn::Index::from(
1759 spec.monitor_index
1760 .expect("Bridge missing monitor index for postprocess"),
1761 );
1762 let enum_ident = Ident::new(
1763 &config_id_to_enum(&format!("{}_bridge", spec.id)),
1764 Span::call_site(),
1765 );
1766 let call_sim = if sim_mode {
1767 quote! {
1768 let doit = {
1769 let state = SimStep::#enum_ident(cu29::simulation::CuBridgeLifecycleState::Postprocess);
1770 let ovr = sim_callback(state);
1771 if let SimOverride::Errored(reason) = ovr {
1772 let error: CuError = reason.into();
1773 let decision = monitor.process_error(cu29::monitoring::ComponentId::new(#monitor_index), CuComponentState::Postprocess, &error);
1774 match decision {
1775 Decision::Abort => { debug!(ctx, "Postprocess: ABORT decision from monitoring. Component '{}' errored out during postprocess. Aborting all the other starts.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index))); return Ok(()); }
1776 Decision::Ignore => { debug!(ctx, "Postprocess: IGNORE decision from monitoring. Component '{}' errored out during postprocess. The runtime will continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index))); false }
1777 Decision::Shutdown => { debug!(ctx, "Postprocess: SHUTDOWN decision from monitoring. Component '{}' errored out during postprocess. The runtime cannot continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index))); return Err(CuError::new_with_cause("Component errored out during postprocess.", error)); }
1778 }
1779 } else {
1780 ovr == SimOverride::ExecuteByRuntime
1781 }
1782 };
1783 }
1784 } else {
1785 quote! { let doit = true; }
1786 };
1787 let alloc_open = alloc_scope_open_tokens();
1788 let alloc_close = alloc_scope_close_tokens(
1789 quote! { monitor },
1790 quote! { #monitor_index },
1791 quote! { CuComponentState::Postprocess },
1792 );
1793 quote! {
1794 {
1795 #call_sim
1796 if doit {
1797 ctx.set_current_component(#monitor_index);
1798 ctx.clear_current_task();
1799 let bridge = &mut __cu_bridges.#bridge_index;
1800 #freeze_bridge
1801 execution_probe.record(cu29::monitoring::ExecutionMarker {
1802 component_id: cu29::monitoring::ComponentId::new(#monitor_index),
1803 step: CuComponentState::Postprocess,
1804 culistid: Some(clid),
1805 });
1806 #alloc_open
1807 let maybe_error = {
1808 #rt_guard
1809 bridge.postprocess(&ctx)
1810 };
1811 #alloc_close
1812 if let Err(error) = maybe_error {
1813 let decision = monitor.process_error(cu29::monitoring::ComponentId::new(#monitor_index), CuComponentState::Postprocess, &error);
1814 match decision {
1815 Decision::Abort => {
1816 debug!(ctx, "Postprocess: ABORT decision from monitoring. Component '{}' errored out during postprocess. Aborting all the other starts.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index)));
1817 return Ok(());
1818 }
1819 Decision::Ignore => {
1820 debug!(ctx, "Postprocess: IGNORE decision from monitoring. Component '{}' errored out during postprocess. The runtime will continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index)));
1821 }
1822 Decision::Shutdown => {
1823 debug!(ctx, "Postprocess: SHUTDOWN decision from monitoring. Component '{}' errored out during postprocess. The runtime cannot continue.", #mission_mod::monitor_component_label(cu29::monitoring::ComponentId::new(#monitor_index)));
1824 return Err(CuError::new_with_cause("Component errored out during postprocess.", error));
1825 }
1826 }
1827 }
1828 }
1829 }
1830 }
1831 })
1832 .collect();
1833
1834 let mut start_calls = bridge_start_calls;
1835 start_calls.extend(task_start_calls);
1836 let mut stop_calls = task_stop_calls;
1837 stop_calls.extend(bridge_stop_calls);
1838 let mut preprocess_calls = bridge_preprocess_calls;
1839 preprocess_calls.extend(task_preprocess_calls);
1840 let mut postprocess_calls = task_postprocess_calls;
1841 postprocess_calls.extend(bridge_postprocess_calls);
1842 let parallel_rt_run_supported = std && parallel_rt_enabled && distributed_selected;
1843
1844 let output_pack_sizes = collect_output_pack_sizes(&culist_plan);
1845 let runtime_plan_code_and_logging: Vec<(
1846 proc_macro2::TokenStream,
1847 proc_macro2::TokenStream,
1848 )> = culist_plan
1849 .steps
1850 .iter()
1851 .enumerate()
1852 .map(|(step_index, unit)| match unit {
1853 CuExecutionUnit::Step(step) => {
1854 #[cfg(feature = "macro_debug")]
1855 eprintln!(
1856 "{} -> {} as {:?}/{:?}. task_id: {} Input={:?}, Output={:?}",
1857 step.node.get_id(),
1858 step.node.get_type(),
1859 step.task_type,
1860 step.phase,
1861 step.node_id,
1862 step.input_msg_indices_types,
1863 step.output_msg_pack
1864 );
1865
1866 match &culist_exec_entities[step.node_id as usize].kind {
1867 ExecutionEntityKind::Task { task_index } => {
1868 let step_ctx = StepGenerationContext::new(
1869 &output_pack_sizes,
1870 &task_input_layouts,
1871 mission.as_str(),
1872 sim_mode,
1873 keyframe_logging_enabled,
1874 &mission_mod,
1875 ParallelLifecyclePlacement::default(),
1876 false,
1877 );
1878 let task_instance = {
1879 let node_index = int2sliceindex(*task_index as u32);
1880 quote! { tasks.#node_index }
1881 };
1882 match step.phase {
1883 CuStepPhase::Whole => generate_task_execution_tokens(
1884 step,
1885 *task_index,
1886 &task_specs,
1887 &runtime_task_types[*task_index],
1888 step_ctx,
1889 TaskExecutionTokens::new(quote! {}, task_instance),
1890 ),
1891 CuStepPhase::AnytimeBase => generate_anytime_base_block(
1892 step,
1893 *task_index,
1894 &task_specs,
1895 &step_ctx,
1896 &task_instance,
1897 ),
1898 CuStepPhase::AnytimeRefine => {
1899 let ordinal = anytime_refine_ordinals[step_index]
1900 .expect("refine step without a precomputed ordinal");
1901 let total = anytime_refine_totals[&step.node_id];
1902 (
1903 generate_anytime_refine_block(
1904 step,
1905 *task_index,
1906 &task_specs,
1907 &step_ctx,
1908 &task_instance,
1909 ordinal,
1910 total,
1911 ),
1912 quote! {},
1913 )
1914 }
1915 }
1916 }
1917 ExecutionEntityKind::BridgeRx {
1918 bridge_index,
1919 channel_index,
1920 } => {
1921 let spec = &culist_bridge_specs[*bridge_index];
1922 generate_bridge_rx_execution_tokens(
1923 step,
1924 spec,
1925 *channel_index,
1926 StepGenerationContext::new(
1927 &output_pack_sizes,
1928 &task_input_layouts,
1929 mission.as_str(),
1930 sim_mode,
1931 keyframe_logging_enabled,
1932 &mission_mod,
1933 ParallelLifecyclePlacement::default(),
1934 false,
1935 ),
1936 {
1937 let bridge_tuple_index =
1938 int2sliceindex(spec.tuple_index as u32);
1939 quote! { let bridge = &mut __cu_bridges.#bridge_tuple_index; }
1940 },
1941 )
1942 }
1943 ExecutionEntityKind::BridgeTx {
1944 bridge_index,
1945 channel_index,
1946 } => {
1947 let spec = &culist_bridge_specs[*bridge_index];
1948 generate_bridge_tx_execution_tokens(
1949 step,
1950 spec,
1951 *channel_index,
1952 StepGenerationContext::new(
1953 &output_pack_sizes,
1954 &task_input_layouts,
1955 mission.as_str(),
1956 sim_mode,
1957 keyframe_logging_enabled,
1958 &mission_mod,
1959 ParallelLifecyclePlacement::default(),
1960 false,
1961 ),
1962 {
1963 let bridge_tuple_index =
1964 int2sliceindex(spec.tuple_index as u32);
1965 quote! { let bridge = &mut __cu_bridges.#bridge_tuple_index; }
1966 },
1967 )
1968 }
1969 }
1970 }
1971 CuExecutionUnit::Loop(_) => {
1972 panic!("Execution loops are not supported in runtime generation");
1973 }
1974 })
1975 .collect();
1976 let parallel_lifecycle_placements = if parallel_rt_run_supported {
1977 Some(build_parallel_lifecycle_placements(
1978 &culist_plan,
1979 &culist_exec_entities,
1980 ))
1981 } else {
1982 None
1983 };
1984 let restore_parallel_placements = (std && parallel_rt_enabled && distributed_selected)
1987 .then(|| build_parallel_lifecycle_placements(&culist_plan, &culist_exec_entities));
1988 let mut keyframe_restore_order = Vec::<ParallelLifecycleKey>::new();
1989 if let Some(placements) = restore_parallel_placements.as_ref() {
1990 for (step_index, unit) in culist_plan.steps.iter().enumerate() {
1991 let CuExecutionUnit::Step(step) = unit else {
1992 panic!("Execution loops are not supported in runtime generation");
1993 };
1994 match &culist_exec_entities[step.node_id as usize].kind {
1995 ExecutionEntityKind::Task { task_index } => {
1996 if step.phase != CuStepPhase::AnytimeRefine
1997 && !keyframe_restore_order
1998 .contains(&ParallelLifecycleKey::Task(*task_index))
1999 {
2000 keyframe_restore_order.push(ParallelLifecycleKey::Task(*task_index));
2001 }
2002 }
2003 ExecutionEntityKind::BridgeRx { bridge_index, .. }
2004 | ExecutionEntityKind::BridgeTx { bridge_index, .. } => {
2005 if placements[step_index].postprocess {
2006 keyframe_restore_order
2007 .push(ParallelLifecycleKey::Bridge(*bridge_index));
2008 }
2009 }
2010 }
2011 }
2012 } else {
2013 keyframe_restore_order.extend(
2014 keyframe_task_restore_order
2015 .iter()
2016 .copied()
2017 .map(ParallelLifecycleKey::Task),
2018 );
2019 keyframe_restore_order
2020 .extend((0..culist_bridge_specs.len()).map(ParallelLifecycleKey::Bridge));
2021 }
2022 let keyframe_restore_code: Vec<proc_macro2::TokenStream> = keyframe_restore_order
2023 .iter()
2024 .map(|component| match component {
2025 ParallelLifecycleKey::Task(index) => {
2026 let task_tuple_index = syn::Index::from(*index);
2027 let skip_substituted = sim_mode
2028 && task_specs.cutypes[*index] != CuTaskType::Regular
2029 && !task_specs.run_in_sim_flags[*index];
2030 if skip_substituted || task_specs.stateless_flags[*index] {
2031 quote! {
2032 let _ = frames.next_frame()?;
2033 }
2034 } else {
2035 quote! {
2036 let frame = frames.next_frame()?;
2037 cu29::curuntime::thaw_keyframe_component(
2038 &mut tasks.#task_tuple_index,
2039 frame,
2040 )?;
2041 }
2042 }
2043 }
2044 ParallelLifecycleKey::Bridge(index) => {
2045 let bridge_tuple_index = syn::Index::from(*index);
2046 if sim_mode && !culist_bridge_specs[*index].run_in_sim {
2047 quote! {
2048 let _ = frames.next_frame()?;
2049 }
2050 } else {
2051 quote! {
2052 let frame = frames.next_frame()?;
2053 cu29::curuntime::thaw_keyframe_component(
2054 &mut __cu_bridges.#bridge_tuple_index,
2055 frame,
2056 )?;
2057 }
2058 }
2059 }
2060 })
2061 .collect();
2062 let keyframe_preallocation_code: Vec<proc_macro2::TokenStream> = keyframe_restore_order
2063 .iter()
2064 .map(|component| match component {
2065 ParallelLifecycleKey::Task(index) => {
2066 let task_tuple_index = syn::Index::from(*index);
2067 let keyframe_state = if task_specs.stateless_flags[*index] {
2068 quote! { cu29::cutask::CuStatelessTaskState }
2069 } else {
2070 quote! { tasks.#task_tuple_index }
2071 };
2072 quote! {
2073 kf_manager.include_capture_capacity(&#keyframe_state)?;
2074 }
2075 }
2076 ParallelLifecycleKey::Bridge(index) => {
2077 let bridge_tuple_index = syn::Index::from(*index);
2078 quote! {
2079 kf_manager.include_capture_capacity(&__cu_bridges.#bridge_tuple_index)?;
2080 }
2081 }
2082 })
2083 .collect();
2084 let runtime_plan_parallel_code_and_logging: Option<
2085 Vec<(proc_macro2::TokenStream, proc_macro2::TokenStream)>,
2086 > = if parallel_rt_run_supported {
2087 Some(
2088 culist_plan
2089 .steps
2090 .iter()
2091 .enumerate()
2092 .filter_map(|(step_index, unit)| match unit {
2093 CuExecutionUnit::Step(step) => match &culist_exec_entities
2094 [step.node_id as usize]
2095 .kind
2096 {
2097 ExecutionEntityKind::Task { task_index } => {
2098 let task_index_ts = int2sliceindex(*task_index as u32);
2099 if step.phase == CuStepPhase::AnytimeRefine {
2104 return None;
2105 }
2106 if step.phase == CuStepPhase::AnytimeBase {
2107 let step_ctx = StepGenerationContext::new(
2108 &output_pack_sizes,
2109 &task_input_layouts,
2110 mission.as_str(),
2111 false,
2112 keyframe_logging_enabled,
2113 &mission_mod,
2114 ParallelLifecyclePlacement::default(),
2115 true,
2116 );
2117 let task_instance = quote! { (*task) };
2118 let (base_block, logging) = generate_anytime_base_block(
2119 step,
2120 *task_index,
2121 &task_specs,
2122 &step_ctx,
2123 &task_instance,
2124 );
2125 let total = anytime_refine_totals[&step.node_id];
2126 let refine_blocks: Vec<proc_macro2::TokenStream> = (1..=total)
2127 .map(|ordinal| {
2128 generate_anytime_refine_block(
2129 step,
2130 *task_index,
2131 &task_specs,
2132 &step_ctx,
2133 &task_instance,
2134 ordinal,
2135 total,
2136 )
2137 })
2138 .collect();
2139 let (parallel_pre, parallel_post) = parallel_task_lifecycle_tokens(
2140 task_trait_for_specs(&task_specs, *task_index),
2141 &task_specs.task_types[*task_index],
2142 task_specs.stateless_flags[*task_index],
2143 *task_index,
2144 &mission_mod,
2145 &task_instance,
2146 parallel_lifecycle_placements
2147 .as_ref()
2148 .expect("parallel lifecycle placements missing")[step_index],
2149 );
2150 let job_local = anytime_job_local_tokens(&task_specs, *task_index);
2151 let body = quote! {
2152 let task = unsafe { step_rt.task_ptrs.#task_index_ts.as_mut() };
2153 #parallel_pre
2154 #job_local
2155 #base_block
2156 #(#refine_blocks)*
2157 #parallel_post
2158 };
2159 return Some((wrap_process_step_tokens(true, body), logging));
2160 }
2161 let task_access = if task_specs.stateless_flags[*task_index] {
2162 quote! { let task = unsafe { step_rt.task_ptrs.#task_index_ts.as_ref() }; }
2163 } else {
2164 quote! { let task = unsafe { step_rt.task_ptrs.#task_index_ts.as_mut() }; }
2165 };
2166 Some(generate_task_execution_tokens(
2167 step,
2168 *task_index,
2169 &task_specs,
2170 &task_specs.task_types[*task_index],
2171 StepGenerationContext::new(
2172 &output_pack_sizes,
2173 &task_input_layouts,
2174 mission.as_str(),
2175 false,
2176 keyframe_logging_enabled,
2177 &mission_mod,
2178 parallel_lifecycle_placements
2179 .as_ref()
2180 .expect("parallel lifecycle placements missing")[step_index],
2181 true,
2182 ),
2183 TaskExecutionTokens::new(quote! {
2184 #task_access
2185 }, quote! { (*task) }),
2186 ))
2187 }
2188 ExecutionEntityKind::BridgeRx {
2189 bridge_index,
2190 channel_index,
2191 } => {
2192 let spec = &culist_bridge_specs[*bridge_index];
2193 let bridge_index_ts = int2sliceindex(spec.tuple_index as u32);
2194 Some(generate_bridge_rx_execution_tokens(
2195 step,
2196 spec,
2197 *channel_index,
2198 StepGenerationContext::new(
2199 &output_pack_sizes,
2200 &task_input_layouts,
2201 mission.as_str(),
2202 false,
2203 keyframe_logging_enabled,
2204 &mission_mod,
2205 parallel_lifecycle_placements
2206 .as_ref()
2207 .expect("parallel lifecycle placements missing")
2208 [step_index],
2209 true,
2210 ),
2211 quote! {
2212 let bridge = unsafe { step_rt.bridge_ptrs.#bridge_index_ts.as_mut() };
2213 },
2214 ))
2215 }
2216 ExecutionEntityKind::BridgeTx {
2217 bridge_index,
2218 channel_index,
2219 } => {
2220 let spec = &culist_bridge_specs[*bridge_index];
2221 let bridge_index_ts = int2sliceindex(spec.tuple_index as u32);
2222 Some(generate_bridge_tx_execution_tokens(
2223 step,
2224 spec,
2225 *channel_index,
2226 StepGenerationContext::new(
2227 &output_pack_sizes,
2228 &task_input_layouts,
2229 mission.as_str(),
2230 false,
2231 keyframe_logging_enabled,
2232 &mission_mod,
2233 parallel_lifecycle_placements
2234 .as_ref()
2235 .expect("parallel lifecycle placements missing")[step_index],
2236 true,
2237 ),
2238 quote! {
2239 let bridge = unsafe { step_rt.bridge_ptrs.#bridge_index_ts.as_mut() };
2240 },
2241 ))
2242 }
2243 },
2244 CuExecutionUnit::Loop(_) => {
2245 panic!("Execution loops are not supported in runtime generation");
2246 }
2247 })
2248 .collect(),
2249 )
2250 } else {
2251 None
2252 };
2253
2254 let sim_support = if sim_mode {
2255 Some(gen_sim_support(
2256 &culist_plan,
2257 &culist_exec_entities,
2258 &culist_bridge_specs,
2259 ))
2260 } else {
2261 None
2262 };
2263
2264 let recorded_replay_support = if sim_mode {
2265 Some(gen_recorded_replay_support(
2266 &culist_plan,
2267 &culist_exec_entities,
2268 &culist_bridge_specs,
2269 ))
2270 } else {
2271 None
2272 };
2273
2274 let (run_one_iteration, start_all_tasks, stop_all_tasks, run) = if sim_mode {
2275 (
2276 quote! {
2277 #[allow(deprecated)] fn run_one_iteration(&mut self, sim_callback: &mut impl FnMut(SimStep) -> SimOverride) -> CuResult<()>
2279 },
2280 quote! {
2281 #[allow(deprecated)] fn start_all_tasks(&mut self, sim_callback: &mut impl FnMut(SimStep) -> SimOverride) -> CuResult<()>
2283 },
2284 quote! {
2285 #[allow(deprecated)] fn stop_all_tasks(&mut self, sim_callback: &mut impl FnMut(SimStep) -> SimOverride) -> CuResult<()>
2287 },
2288 quote! {
2289 #[allow(deprecated)] fn run(&mut self, sim_callback: &mut impl FnMut(SimStep) -> SimOverride) -> CuResult<()>
2291 },
2292 )
2293 } else {
2294 (
2295 quote! {
2296 #[allow(deprecated)] fn run_one_iteration(&mut self) -> CuResult<()>
2298 },
2299 quote! {
2300 #[allow(deprecated)] fn start_all_tasks(&mut self) -> CuResult<()>
2302 },
2303 quote! {
2304 #[allow(deprecated)] fn stop_all_tasks(&mut self) -> CuResult<()>
2306 },
2307 quote! {
2308 #[allow(deprecated)] fn run(&mut self) -> CuResult<()>
2310 },
2311 )
2312 };
2313
2314 let sim_callback_arg = if sim_mode {
2315 Some(quote!(sim_callback))
2316 } else {
2317 None
2318 };
2319
2320 let app_trait = if sim_mode {
2321 quote!(CuSimApplication)
2322 } else {
2323 quote!(CuApplication)
2324 };
2325
2326 let sim_callback_on_new_calls = task_specs.ids.iter().enumerate().map(|(i, id)| {
2327 let enum_name = config_id_to_enum(id);
2328 let enum_ident = Ident::new(&enum_name, Span::call_site());
2329 quote! {
2330 sim_callback(SimStep::#enum_ident(CuTaskCallbackState::New(all_instances_configs[#i].cloned())));
2332 }
2333 });
2334
2335 let sim_callback_on_new_bridges = culist_bridge_specs.iter().map(|spec| {
2336 let enum_ident = Ident::new(
2337 &config_id_to_enum(&format!("{}_bridge", spec.id)),
2338 Span::call_site(),
2339 );
2340 let cfg_index = syn::Index::from(spec.config_index);
2341 quote! {
2342 sim_callback(SimStep::#enum_ident(
2343 cu29::simulation::CuBridgeLifecycleState::New(config.bridges[#cfg_index].config.clone())
2344 ));
2345 }
2346 });
2347
2348 let sim_callback_on_new = if sim_mode {
2349 Some(quote! {
2350 let graph = config.get_graph(Some(#mission)).expect("Could not find the mission #mission");
2351 let all_instances_configs: Vec<Option<&ComponentConfig>> = graph
2352 .get_all_nodes()
2353 .iter()
2354 .map(|(_, node)| node.get_instance_config())
2355 .collect();
2356 #(#sim_callback_on_new_calls)*
2357 #(#sim_callback_on_new_bridges)*
2358 })
2359 } else {
2360 None
2361 };
2362
2363 let anytime_job_locals = build_anytime_job_locals(&task_specs);
2364 let anytime_policy_defs = build_anytime_policy_defs(&task_specs);
2365 let (runtime_plan_code, preprocess_logging_calls): (Vec<_>, Vec<_>) =
2366 itertools::multiunzip(runtime_plan_code_and_logging);
2367 let process_step_tasks_type = if sim_mode {
2368 quote!(CuSimTasks)
2369 } else {
2370 quote!(CuTasks)
2371 };
2372 let parallel_keyframe_component_indices: Vec<usize> = culist_plan
2373 .steps
2374 .iter()
2375 .filter_map(|unit| {
2376 let CuExecutionUnit::Step(step) = unit else {
2377 panic!("Execution loops are not supported in runtime generation");
2378 };
2379 if step.phase == CuStepPhase::AnytimeRefine {
2380 return None;
2381 }
2382 let component = match &culist_exec_entities[step.node_id as usize].kind {
2383 ExecutionEntityKind::Task { task_index } => {
2384 ParallelLifecycleKey::Task(*task_index)
2385 }
2386 ExecutionEntityKind::BridgeRx { bridge_index, .. }
2387 | ExecutionEntityKind::BridgeTx { bridge_index, .. } => {
2388 ParallelLifecycleKey::Bridge(*bridge_index)
2389 }
2390 };
2391 Some(
2392 keyframe_restore_order
2393 .iter()
2394 .position(|candidate| *candidate == component)
2395 .expect("parallel keyframe component missing from restore order"),
2396 )
2397 })
2398 .collect();
2399 let (parallel_process_step_idents, parallel_process_step_fn_defs): (
2400 Vec<Ident>,
2401 Vec<proc_macro2::TokenStream>,
2402 ) = if let Some(runtime_plan_parallel_code_and_logging) =
2403 &runtime_plan_parallel_code_and_logging
2404 {
2405 let (runtime_plan_parallel_step_code, _): (Vec<_>, Vec<_>) =
2406 itertools::multiunzip(runtime_plan_parallel_code_and_logging.clone());
2407 let parallel_process_step_idents: Vec<Ident> = (0..runtime_plan_parallel_step_code
2408 .len())
2409 .map(|index| format_ident!("__cu_parallel_process_step_{index}"))
2410 .collect();
2411 let parallel_process_step_fn_defs: Vec<proc_macro2::TokenStream> =
2412 parallel_process_step_idents
2413 .iter()
2414 .zip(runtime_plan_parallel_step_code.iter())
2415 .zip(parallel_keyframe_component_indices.iter())
2416 .map(|((step_ident, step_code), component)| {
2417 let keyframe_local = keyframe_logging_enabled.then(|| {
2418 quote! {
2419 let kf_manager = ParallelKeyFrameAccessor::new(
2420 step_rt.keyframe_ptr,
2421 step_rt.keyframe_len,
2422 #component,
2423 );
2424 }
2425 });
2426 quote! {
2427 #[inline(always)]
2428 fn #step_ident(
2429 step_rt: &mut ParallelProcessStepRuntime<'_>,
2430 ) -> cu29::curuntime::ProcessStepResult {
2431 let clock = step_rt.clock;
2432 let execution_probe = step_rt.execution_probe;
2433 let monitor = step_rt.monitor;
2434 #keyframe_local
2435 let clid = step_rt.clid;
2436 let ctx = &mut step_rt.ctx;
2437 let msgs = unsafe {
2441 core::ptr::addr_of_mut!((*step_rt.culist).msgs.0)
2442 };
2443 #[allow(unused_macros)]
2444 macro_rules! __cu_input {
2445 ($field:tt) => { unsafe { &(*msgs).$field.value } };
2446 ($field:tt, $port:tt) => {
2447 unsafe { &(*msgs).$field.value.$port }
2448 };
2449 }
2450 #[allow(unused_macros)]
2451 macro_rules! __cu_output {
2452 ($field:tt) => { unsafe { &mut (*msgs).$field.value } };
2453 ($field:tt, $port:tt) => {
2454 unsafe { &mut (*msgs).$field.value.$port }
2455 };
2456 }
2457 #step_code
2458 }
2459 }
2460 })
2461 .collect();
2462 (parallel_process_step_idents, parallel_process_step_fn_defs)
2463 } else {
2464 (Vec::new(), Vec::new())
2465 };
2466 let parallel_process_stage_count_tokens =
2467 proc_macro2::Literal::usize_unsuffixed(parallel_process_step_idents.len());
2468 let parallel_task_ptrs_type = if runtime_task_types.is_empty() {
2469 quote! { () }
2470 } else {
2471 let elems = runtime_task_types
2472 .iter()
2473 .map(|ty| quote! { ParallelSharedPtr<#ty> });
2474 quote! { (#(#elems),*,) }
2475 };
2476 let parallel_task_ptr_values = if runtime_task_types.is_empty() {
2477 quote! { () }
2478 } else {
2479 let elems = (0..runtime_task_types.len()).map(|index| {
2480 let index = syn::Index::from(index);
2481 quote! { ParallelSharedPtr::new(&mut runtime.tasks.#index.value as *mut _) }
2482 });
2483 quote! { (#(#elems),*,) }
2484 };
2485 let parallel_bridge_ptrs_type = if bridge_runtime_types.is_empty() {
2486 quote! { () }
2487 } else {
2488 let elems = bridge_runtime_types
2489 .iter()
2490 .map(|ty| quote! { ParallelSharedPtr<#ty> });
2491 quote! { (#(#elems),*,) }
2492 };
2493 let parallel_bridge_ptr_values = if bridge_runtime_types.is_empty() {
2494 quote! { () }
2495 } else {
2496 let elems = (0..bridge_runtime_types.len()).map(|index| {
2497 let index = syn::Index::from(index);
2498 quote! { ParallelSharedPtr::new(&mut runtime.bridges.#index.value as *mut _) }
2499 });
2500 quote! { (#(#elems),*,) }
2501 };
2502 let parallel_keyframe_accessor = keyframe_logging_enabled.then(|| {
2503 quote! {
2504 struct ParallelKeyFrameAccessor {
2505 ptr: *mut u8,
2506 len: usize,
2507 component: usize,
2508 }
2509
2510 impl ParallelKeyFrameAccessor {
2511 #[inline(always)]
2512 fn new(
2513 ptr: *mut u8,
2514 len: usize,
2515 component: usize,
2516 ) -> Self {
2517 Self { ptr, len, component }
2518 }
2519
2520 #[inline(always)]
2521 fn freeze_task(
2522 &self,
2523 _culistid: u64,
2524 task: &impl cu29::cutask::Freezable,
2525 ) -> CuResult<usize> {
2526 unsafe {
2529 cu29::curuntime::KeyFrame::freeze_region(
2530 self.ptr,
2531 self.len,
2532 self.component,
2533 task,
2534 )
2535 }
2536 }
2537
2538 #[inline(always)]
2539 fn freeze_any(
2540 &self,
2541 culistid: u64,
2542 item: &impl cu29::cutask::Freezable,
2543 ) -> CuResult<usize> {
2544 self.freeze_task(culistid, item)
2545 }
2546 }
2547 }
2548 });
2549 let parallel_process_step_keyframe_fields = keyframe_logging_enabled.then(|| {
2550 quote! {
2551 keyframe_ptr: *mut u8,
2552 keyframe_len: usize,
2553 }
2554 });
2555 let parallel_rt_support_tokens = if parallel_rt_run_supported {
2556 quote! {
2557 type ParallelTaskPtrs = #parallel_task_ptrs_type;
2558 type ParallelBridgePtrs = #parallel_bridge_ptrs_type;
2559
2560 struct ParallelSharedPtr<T>(*mut T);
2561
2562 impl<T> Clone for ParallelSharedPtr<T> {
2563 #[inline(always)]
2564 fn clone(&self) -> Self {
2565 *self
2566 }
2567 }
2568
2569 impl<T> Copy for ParallelSharedPtr<T> {}
2570
2571 impl<T> ParallelSharedPtr<T> {
2572 #[inline(always)]
2573 const fn new(ptr: *mut T) -> Self {
2574 Self(ptr)
2575 }
2576
2577 #[inline(always)]
2578 const fn from_ref(ptr: *const T) -> Self {
2579 Self(ptr as *mut T)
2580 }
2581
2582 #[inline(always)]
2583 unsafe fn as_mut<'a>(self) -> &'a mut T {
2584 unsafe { &mut *self.0 }
2585 }
2586
2587 #[inline(always)]
2588 unsafe fn as_ref<'a>(self) -> &'a T {
2589 unsafe { &*self.0 }
2590 }
2591 }
2592
2593 unsafe impl<T: Send> Send for ParallelSharedPtr<T> {}
2594 unsafe impl<T: Send> Sync for ParallelSharedPtr<T> {}
2595
2596 #parallel_keyframe_accessor
2597
2598 struct ParallelProcessStepRuntime<'a> {
2599 clock: &'a RobotClock,
2600 execution_probe: &'a cu29::monitoring::RuntimeExecutionProbe,
2601 monitor: &'a #monitor_type,
2602 task_ptrs: &'a ParallelTaskPtrs,
2603 bridge_ptrs: &'a ParallelBridgePtrs,
2604 #parallel_process_step_keyframe_fields
2605 culist: *mut CuList,
2606 clid: u64,
2607 ctx: cu29::context::CuContext,
2608 }
2609
2610 #[inline(always)]
2611 fn assert_parallel_rt_send_bounds()
2612 where
2613 CuList: Send,
2614 #process_step_tasks_type: Send,
2615 CuBridges: Send,
2616 #monitor_type: Sync,
2617 {
2618 }
2619
2620 #(#parallel_process_step_fn_defs)*
2621 }
2622 } else {
2623 quote! {}
2624 };
2625
2626 let config_load_stmt = build_config_load_stmt(
2627 std,
2628 application_name,
2629 subsystem_id.as_deref(),
2630 &config_features,
2631 );
2632 let constant_override_warning = if std {
2633 let comparisons = constant_fingerprints
2634 .iter()
2635 .map(|(module, id, fingerprint)| {
2636 quote! { (#module, #id) => fingerprint != #fingerprint, }
2637 });
2638 Some(quote! {
2639 for constant in &config.constants {
2640 let changed = constant
2641 .semantic_fingerprint()
2642 .map_or(true, |fingerprint| match (constant.module_path(), constant.id()) {
2643 #(#comparisons)*
2644 _ => true,
2645 });
2646 if changed {
2647 ::cu29::prelude::warning!(
2648 "Runtime configuration tried to override compile-time constant '{}'; the value baked into this binary will be used.",
2649 constant.qualified_id()
2650 );
2651 }
2652 }
2653 })
2654 } else {
2655 None
2656 };
2657
2658 let copperlist_count_check = quote! {
2659 let configured_copperlist_count = config
2660 .logging
2661 .as_ref()
2662 .and_then(|logging| logging.copperlist_count)
2663 .unwrap_or(#copperlist_count_tokens);
2664 if configured_copperlist_count != #copperlist_count_tokens {
2665 return Err(CuError::from(format!(
2666 "Configured logging.copperlist_count ({configured_copperlist_count}) does not match the runtime compiled into this binary ({})",
2667 #copperlist_count_tokens
2668 )));
2669 }
2670 };
2671 let planner_kind_tokens = match copper_config.planner_kind() {
2672 PlannerKind::Serial => quote! { cu29::config::PlannerKind::Serial },
2673 PlannerKind::TaskOrder => quote! { cu29::config::PlannerKind::TaskOrder },
2674 PlannerKind::ExplicitSchedule => {
2675 quote! { cu29::config::PlannerKind::ExplicitSchedule }
2676 }
2677 PlannerKind::Pipeline => quote! { cu29::config::PlannerKind::Pipeline },
2678 };
2679 let planner_config_check = quote! {
2680 let configured_planner_kind = config.planner_kind();
2681 if configured_planner_kind != #planner_kind_tokens {
2682 return Err(CuError::from(format!(
2683 "Configured planner kind ({configured_planner_kind:?}) does not match the runtime compiled into this binary ({:?})",
2684 #planner_kind_tokens,
2685 )));
2686 }
2687 if configured_planner_kind == cu29::config::PlannerKind::Pipeline {
2688 let configured_max_in_flight = config
2689 .planner_config()
2690 .expect("Pipeline planner configuration disappeared")
2691 .max_in_flight(configured_copperlist_count)?;
2692 if configured_max_in_flight != #pipeline_max_in_flight_tokens {
2693 return Err(CuError::from(format!(
2694 "Configured Pipeline max_in_flight ({configured_max_in_flight}) does not match the runtime compiled into this binary ({})",
2695 #pipeline_max_in_flight_tokens,
2696 )));
2697 }
2698 }
2699 };
2700 let keyframe_logging_check = quote! {
2701 let configured_keyframe_logging = config
2702 .logging
2703 .as_ref()
2704 .is_none_or(|logging| logging.enable_keyframe_logging && logging.enable_task_logging);
2705 if configured_keyframe_logging != #local_keyframe_logging_enabled
2708 && !(#sim_mode && #logstream_enabled && !configured_keyframe_logging) {
2709 return Err(CuError::from(format!(
2710 "Configured keyframe logging ({configured_keyframe_logging}) does not match the runtime compiled into this binary ({})",
2711 #local_keyframe_logging_enabled
2712 )));
2713 }
2714 };
2715 let logstream_topology_check = if logstream_enabled {
2716 let expected_count = logstream_resource_specs.len();
2717 let destination_checks = logstream_resource_specs.iter().map(|spec| {
2718 let index = syn::Index::from(spec.destination_index);
2719 let destination = &copper_config
2720 .log_streaming
2721 .as_ref()
2722 .expect("logstream specs require config")
2723 .destinations[spec.destination_index];
2724 let expected_id = LitStr::new(&destination.id, Span::call_site());
2725 let expected_type = LitStr::new(&destination.transport.type_, Span::call_site());
2726 let expected_resource =
2727 LitStr::new(&destination.transport.resource, Span::call_site());
2728 let expected_feedback = match &destination.feedback {
2729 Some(feedback) => {
2730 let ty = LitStr::new(&feedback.transport.type_, Span::call_site());
2731 let resource = LitStr::new(&feedback.transport.resource, Span::call_site());
2732 quote! { Some((#ty, #resource)) }
2733 }
2734 None => quote! { None },
2735 };
2736 quote! {
2737 let destination = &configured_logstream_destinations[#index];
2738 if destination.id != #expected_id
2739 || destination.transport.type_ != #expected_type
2740 || destination.transport.resource != #expected_resource
2741 || destination.feedback.as_ref().map(|f| (f.transport.type_.as_str(), f.transport.resource.as_str())) != #expected_feedback
2742 {
2743 return Err(CuError::from(format!(
2744 "Configured log_streaming destination topology at index {} does not match the runtime compiled into this binary",
2745 #index,
2746 )));
2747 }
2748 }
2749 });
2750 quote! {
2751 let configured_logstream_destinations = config
2752 .log_streaming
2753 .as_ref()
2754 .map_or(&[][..], |streaming| streaming.destinations.as_slice());
2755 if configured_logstream_destinations.len() != #expected_count {
2756 return Err(CuError::from(format!(
2757 "Configured log_streaming destination count ({}) does not match the runtime compiled into this binary ({})",
2758 configured_logstream_destinations.len(),
2759 #expected_count,
2760 )));
2761 }
2762 #(#destination_checks)*
2763 }
2764 } else {
2765 quote! {}
2766 };
2767
2768 let prepare_config_sig = if std {
2769 quote! {
2770 fn prepare_config(
2771 instance_id: u32,
2772 config_override: Option<CuConfig>,
2773 ) -> CuResult<(CuConfig, RuntimeLifecycleConfigSource)>
2774 }
2775 } else {
2776 quote! {
2777 fn prepare_config() -> CuResult<(CuConfig, RuntimeLifecycleConfigSource)>
2778 }
2779 };
2780
2781 let prepare_config_call = if std {
2782 quote! { Self::prepare_config(instance_id, config_override)? }
2783 } else {
2784 quote! { Self::prepare_config()? }
2785 };
2786
2787 let prepare_resources_sig = if std {
2788 quote! {
2789 pub fn prepare_resources_for_instance(
2790 instance_id: u32,
2791 config_override: Option<CuConfig>,
2792 ) -> CuResult<AppResources>
2793 }
2794 } else {
2795 quote! {
2796 pub fn prepare_resources() -> CuResult<AppResources>
2797 }
2798 };
2799
2800 let prepare_resources_compat_fn = if std {
2801 Some(quote! {
2802 pub fn prepare_resources(
2803 config_override: Option<CuConfig>,
2804 ) -> CuResult<AppResources> {
2805 Self::prepare_resources_for_instance(0, config_override)
2806 }
2807 })
2808 } else {
2809 None
2810 };
2811
2812 let init_resources_compat_fn = if std {
2813 Some(quote! {
2814 pub fn init_resources_for_instance(
2815 instance_id: u32,
2816 config_override: Option<CuConfig>,
2817 ) -> CuResult<AppResources> {
2818 Self::prepare_resources_for_instance(instance_id, config_override)
2819 }
2820
2821 pub fn init_resources(
2822 config_override: Option<CuConfig>,
2823 ) -> CuResult<AppResources> {
2824 Self::prepare_resources(config_override)
2825 }
2826 })
2827 } else {
2828 Some(quote! {
2829 pub fn init_resources() -> CuResult<AppResources> {
2830 Self::prepare_resources()
2831 }
2832 })
2833 };
2834
2835 let build_with_resources_logstream_param = if logstream_enabled {
2836 quote! {
2837 logstream: Option<(
2838 Box<dyn ::cu29::logstream::CuStreamTx>,
2839 ::cu29::logstream::LogStreamSenderConfig,
2840 )>,
2841 }
2842 } else {
2843 quote! {}
2844 };
2845 let build_with_resources_sig = if sim_mode {
2846 quote! {
2847 fn build_with_resources<S: SectionStorage + 'static, L: UnifiedLogWrite<S> + 'static>(
2848 clock: RobotClock,
2849 unified_logger: Arc<Mutex<L>>,
2850 app_resources: AppResources,
2851 instance_id: u32,
2852 #build_with_resources_logstream_param
2853 sim_callback: &mut impl FnMut(SimStep) -> SimOverride,
2854 ) -> CuResult<Self>
2855 }
2856 } else {
2857 quote! {
2858 fn build_with_resources<S: SectionStorage + 'static, L: UnifiedLogWrite<S> + 'static>(
2859 clock: RobotClock,
2860 unified_logger: Arc<Mutex<L>>,
2861 app_resources: AppResources,
2862 instance_id: u32,
2863 #build_with_resources_logstream_param
2864 ) -> CuResult<Self>
2865 }
2866 };
2867 let parallel_rt_metadata_arg = if std {
2868 Some(quote! {
2869 &#mission_mod::PARALLEL_RT_METADATA,
2870 })
2871 } else {
2872 None
2873 };
2874
2875 let kill_handler = if std && signal_handler {
2876 Some(quote! {
2877 ctrlc::set_handler(move || {
2878 STOP_FLAG.store(true, Ordering::SeqCst);
2879 }).expect("Error setting Ctrl-C handler");
2880 })
2881 } else {
2882 None
2883 };
2884
2885 let run_loop = if std {
2886 quote! {{
2887 let mut rate_limiter = self
2888 .copper_runtime
2889 .runtime_config
2890 .rate_target_hz
2891 .map(|rate| cu29::curuntime::LoopRateLimiter::from_rate_target_hz(
2892 rate,
2893 self.copper_runtime.clock_ref(),
2894 ))
2895 .transpose()?;
2896 loop {
2897 let result = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(
2898 || <Self as #app_trait<S, L>>::run_one_iteration(self, #sim_callback_arg)
2899 )) {
2900 Ok(result) => result,
2901 Err(payload) => {
2902 self.stop_reason = cu29::curuntime::RuntimeStopReason::Panic;
2903 let panic_message = cu29::monitoring::panic_payload_to_string(payload.as_ref());
2904 self.copper_runtime.monitor.process_panic(&panic_message);
2905 let _ = self.log_runtime_lifecycle_event(RuntimeLifecycleEvent::Panic {
2906 message: panic_message.clone(),
2907 file: None,
2908 line: None,
2909 column: None,
2910 });
2911 Err(CuError::from(format!(
2912 "Panic while running one iteration: {}",
2913 panic_message
2914 )))
2915 }
2916 };
2917
2918 if let Some(rate_limiter) = rate_limiter.as_mut() {
2919 rate_limiter.limit(self.copper_runtime.clock_ref());
2920 }
2921
2922 if STOP_FLAG.load(Ordering::SeqCst) || result.is_err() {
2923 if result.is_err() && self.stop_reason != cu29::curuntime::RuntimeStopReason::Panic { self.stop_reason = cu29::curuntime::RuntimeStopReason::Error; }
2924 break result;
2925 }
2926 }
2927 }}
2928 } else {
2929 quote! {{
2930 let mut rate_limiter = self
2931 .copper_runtime
2932 .runtime_config
2933 .rate_target_hz
2934 .map(|rate| cu29::curuntime::LoopRateLimiter::from_rate_target_hz(
2935 rate,
2936 self.copper_runtime.clock_ref(),
2937 ))
2938 .transpose()?;
2939 loop {
2940 let result = <Self as #app_trait<S, L>>::run_one_iteration(self, #sim_callback_arg);
2941 if let Some(rate_limiter) = rate_limiter.as_mut() {
2942 rate_limiter.limit(self.copper_runtime.clock_ref());
2943 }
2944
2945 if STOP_FLAG.load(Ordering::SeqCst) || result.is_err() {
2946 if result.is_err() && self.stop_reason != cu29::curuntime::RuntimeStopReason::Panic { self.stop_reason = cu29::curuntime::RuntimeStopReason::Error; }
2947 break result;
2948 }
2949 }
2950 }}
2951 };
2952
2953 let parallel_keyframe_runtime = keyframe_logging_enabled.then(|| {
2954 quote! {
2955 let keyframe_interval = runtime.keyframes_manager.capture_interval();
2956 }
2957 });
2958 let parallel_keyframe_reset = keyframe_logging_enabled.then(|| {
2959 quote! {
2960 if let Some(keyframe) = culist.keyframe_mut() {
2961 keyframe.reset_distributed(clid, clock.now(), keyframe_interval);
2962 }
2963 }
2964 });
2965 let parallel_keyframe_capture = if keyframe_logging_enabled {
2966 quote! { let (keyframe_ptr, keyframe_len) = culist.keyframe_capture(); }
2967 } else {
2968 quote! {
2969 let keyframe_ptr = core::ptr::null_mut();
2970 let keyframe_len = 0usize;
2971 }
2972 };
2973 let parallel_keyframe_finish = if keyframe_logging_enabled {
2974 quote! { (cl_manager.last_keyframe_bytes, 0u64) }
2975 } else {
2976 quote! { (0u64, 0u64) }
2977 };
2978
2979 #[cfg(feature = "macro_debug")]
2980 eprintln!("[build the run methods]");
2981 let lane_executor_tokens = if parallel_rt_run_supported {
2982 match build_lane_executor_tokens(
2983 lane_plan.expect("distributed plan"),
2984 &culist_plan,
2985 ¶llel_process_step_idents,
2986 &mission_mod,
2987 keyframe_logging_enabled,
2988 ) {
2989 Ok(tokens) => Some(tokens),
2990 Err(error) => {
2991 return return_error(format!("Mission '{mission}': {error}"));
2992 }
2993 }
2994 } else {
2995 None
2996 };
2997 let run_body: proc_macro2::TokenStream = if let Some(lane) = lane_executor_tokens {
2998 let LaneExecutorTokens {
2999 worker_spawns: lane_worker_spawns,
3000 dispatcher_scheduling: lane_dispatcher_scheduling,
3001 completion_ready: lane_completion_ready,
3002 copperlists_per_cycle: lane_copperlists_per_cycle,
3003 max_in_flight: lane_max_in_flight,
3004 workers: lane_workers,
3005 } = lane;
3006 quote! {
3007 static STOP_FLAG: AtomicBool = AtomicBool::new(false);
3008
3009 #kill_handler
3010
3011 let mut tasks_started = false;
3012 let result = std::thread::scope(|scope| -> CuResult<()> {
3013 #mission_mod::assert_parallel_rt_send_bounds();
3014 #lane_dispatcher_scheduling
3015
3016 let runtime = &mut self.copper_runtime;
3017 let clock_handle = runtime.clock();
3018 let clock = &clock_handle;
3019 let instance_id = runtime.instance_id();
3020 let subsystem_code = runtime.subsystem_code();
3021 let execution_probe_ptr =
3022 #mission_mod::ParallelSharedPtr::from_ref(
3023 runtime.execution_probe.as_ref() as *const _,
3024 );
3025 let monitor_ptr =
3026 #mission_mod::ParallelSharedPtr::from_ref(
3027 &runtime.monitor as *const _,
3028 );
3029 let task_ptrs: #mission_mod::ParallelTaskPtrs = #parallel_task_ptr_values;
3030 let bridge_ptrs: #mission_mod::ParallelBridgePtrs = #parallel_bridge_ptr_values;
3031 let mut free_copperlists: Vec<Option<cu29::arena::CuSlotLease<CuStampedDataSet>>> =
3032 (0..#copperlist_count_tokens).map(|_| None).collect();
3033 for token in runtime.copperlists_manager.take_execution_slots() {
3034 let index = token.index();
3035 free_copperlists[index] = Some(token);
3036 }
3037 #parallel_keyframe_runtime
3038 let start_clid = runtime.copperlists_manager.next_cl_id();
3039 runtime.parallel_rt.reset_cursors(start_clid);
3040 debug_assert_eq!(
3041 runtime.parallel_rt.metadata().process_stage_count(),
3042 #parallel_process_stage_count_tokens,
3043 );
3044 let max_in_flight = #lane_max_in_flight;
3045 debug_assert_eq!(runtime.parallel_rt.in_flight_limit(), max_in_flight);
3046 let copperlists_per_cycle = #lane_copperlists_per_cycle;
3047 let lanes = std::sync::Arc::new(cu29::parallel_rt::LaneExecutor::new(
3048 start_clid,
3049 max_in_flight,
3050 #lane_copperlists_per_cycle as u32,
3051 #lane_workers,
3052 ));
3053 let mut in_flight_slots: Vec<Option<cu29::arena::CuSlotLease<CuStampedDataSet>>> =
3054 (0..max_in_flight).map(|_| None).collect();
3055 let mut returned_copperlists = Vec::with_capacity(#copperlist_count_tokens);
3056 let mut lane_handles = Vec::with_capacity(#lane_workers);
3057 #(#lane_worker_spawns)*
3058
3059 let execution_result = (|| -> CuResult<()> {
3060 <Self as #app_trait<S, L>>::start_all_tasks(self)?;
3061 tasks_started = true;
3062
3063 let runtime = &mut self.copper_runtime;
3064 let monitor = &runtime.monitor;
3065 let cl_manager = &mut runtime.copperlists_manager;
3066 let parallel_rt = &runtime.parallel_rt;
3067
3068 let mut dispatch_limiter = runtime
3069 .runtime_config
3070 .rate_target_hz
3071 .map(|rate| cu29::curuntime::LoopRateLimiter::from_rate_target_hz(rate, clock))
3072 .transpose()?;
3073 let mut in_flight = 0usize;
3074 let mut stop_launching = false;
3075 let mut next_launch_clid = start_clid;
3076 let mut next_commit_clid = start_clid;
3077 let mut fatal_error: Option<CuError> = None;
3078 let mut failed_clid: Option<u64> = None;
3079
3080 loop {
3081 let observed_progress = lanes.progress_generation();
3082 while let Some(recycled) = cl_manager.try_reclaim_slot()? {
3083 let index = recycled.index();
3084 free_copperlists[index] = Some(recycled);
3085 }
3086 if let Some((clid, error)) = lanes.take_failure() {
3087 fatal_error = Some(error);
3088 failed_clid = Some(clid);
3089 stop_launching = true;
3090 }
3091
3092 while in_flight > 0
3093 && failed_clid.is_none_or(|clid| next_commit_clid < clid)
3094 && #lane_completion_ready
3095 {
3096 if parallel_rt.current_commit_clid() != next_commit_clid {
3097 fatal_error = Some(CuError::from(
3098 "Parallel commit checkpoint out of sync",
3099 ));
3100 stop_launching = true;
3101 break;
3102 }
3103 let slot = (next_commit_clid % max_in_flight as u64) as usize;
3104 let mut culist = in_flight_slots[slot]
3105 .take()
3106 .expect("completed CopperList slot");
3107 let clid = next_commit_clid;
3108 debug_assert_eq!(culist.id, clid);
3109 let outcome = lanes.completion_outcome(clid)?;
3110 let mut commit_ctx = cu29::context::CuContext::from_runtime_metadata(
3111 clock.clone(), clid, instance_id, subsystem_code,
3112 #mission_mod::TASK_IDS,
3113 );
3114 commit_ctx.clear_current_component();
3115 commit_ctx.clear_current_task();
3116 let monitor_result = monitor.process_copperlist(
3117 &commit_ctx,
3118 #mission_mod::MONITOR_LAYOUT.view(
3119 &#mission_mod::collect_metadata(&culist),
3120 ),
3121 );
3122 if outcome == cu29::curuntime::ProcessStepOutcome::AbortCopperList {
3123 if let Some(keyframe) = culist.keyframe_mut() {
3124 keyframe.discard_distributed();
3125 }
3126 } else {
3127 #(#preprocess_logging_calls)*
3128 }
3129 if let cu29::curuntime::OwnedCopperListSubmission::Recycled(culist) =
3130 cl_manager.submit_slot(culist)?
3131 {
3132 let index = culist.index();
3133 free_copperlists[index] = Some(culist);
3134 }
3135 monitor_result?;
3136 if outcome == cu29::curuntime::ProcessStepOutcome::Continue {
3137 let (keyframe_bytes, dropped_keyframes_total) = #parallel_keyframe_finish;
3138 monitor.observe_copperlist_io(cu29::monitoring::CopperListIoStats {
3139 raw_culist_bytes: core::mem::size_of::<CuList>() as u64
3140 + cl_manager.last_handle_bytes,
3141 handle_bytes: cl_manager.last_handle_bytes,
3142 encoded_culist_bytes: cl_manager.last_encoded_bytes,
3143 keyframe_bytes,
3144 structured_log_bytes_total: ::cu29::prelude::structured_log_bytes_total(),
3145 culistid: clid,
3146 dropped_copperlists_total: cl_manager.dropped_copperlists_total(),
3147 dropped_keyframes_total,
3148 });
3149 }
3150 parallel_rt.release_commit(clid + 1);
3151 next_commit_clid += 1;
3152 in_flight -= 1;
3153 }
3154
3155 if fatal_error.is_some()
3156 && failed_clid.is_none_or(|clid| next_commit_clid >= clid)
3157 {
3158 break;
3159 }
3160
3161 if STOP_FLAG.load(Ordering::SeqCst)
3162 && !stop_launching
3163 && (next_launch_clid - start_clid).is_multiple_of(copperlists_per_cycle)
3164 {
3165 stop_launching = true;
3166 lanes.stop_admitting();
3167 }
3168 if !stop_launching && fatal_error.is_none() {
3169 let next_clid = next_launch_clid;
3170 let rate_ready = dispatch_limiter
3171 .as_ref()
3172 .map(|limiter| limiter.is_ready(clock))
3173 .unwrap_or(true);
3174 let arena_index =
3175 (next_clid % #copperlist_count_tokens as u64) as usize;
3176 let execution_slot =
3177 (next_clid % max_in_flight as u64) as usize;
3178
3179 if in_flight < max_in_flight
3180 && rate_ready
3181 && free_copperlists[arena_index].is_some()
3182 && in_flight_slots[execution_slot].is_none()
3183 {
3184 let clid = next_clid;
3185 let mut culist = free_copperlists[arena_index]
3186 .take()
3187 .expect("free CopperList slot disappeared");
3188 culist.reset(clid);
3189 #parallel_keyframe_reset
3190 culist.change_state(cu29::copperlist::CopperListState::Processing);
3191 let culist_ptr = culist.as_ptr().cast::<u8>();
3192 #parallel_keyframe_capture
3193 in_flight_slots[execution_slot] = Some(culist);
3194 lanes.admit_with_keyframe(
3195 clid,
3196 culist_ptr,
3197 keyframe_ptr,
3198 keyframe_len,
3199 );
3200 cl_manager.note_admission(clid + 1);
3201 next_launch_clid += 1;
3202 in_flight += 1;
3203 if let Some(limiter) = dispatch_limiter.as_mut() {
3204 limiter.mark_tick(clock);
3205 }
3206 continue;
3207 }
3208 }
3209
3210 if in_flight == 0 {
3211 if stop_launching || fatal_error.is_some() {
3212 break;
3213 }
3214 }
3215 let arena_index =
3216 (next_launch_clid % #copperlist_count_tokens as u64) as usize;
3217 if !stop_launching
3218 && free_copperlists[arena_index].is_none()
3219 && cl_manager.has_pending_slots()
3220 {
3221 let token = cl_manager.wait_reclaim_slot()?;
3222 let index = token.index();
3223 free_copperlists[index] = Some(token);
3224 continue;
3225 }
3226 let timeout = if stop_launching {
3227 None
3228 } else {
3229 dispatch_limiter
3230 .as_ref()
3231 .and_then(|limiter| limiter.remaining(clock))
3232 .map(std::time::Duration::from)
3233 };
3234 lanes.wait_for_progress(observed_progress, timeout);
3235 }
3236
3237 fatal_error.map_or(Ok(()), Err)
3238 })();
3239
3240 lanes.request_shutdown();
3241 for handle in lane_handles.drain(..) {
3242 let _ = handle.join();
3243 }
3244 for slot in in_flight_slots.into_iter().flatten() {
3245 let index = slot.index();
3246 free_copperlists[index] = Some(slot);
3247 }
3248 let cleanup_result = (|| -> CuResult<()> {
3249 let cl_manager = &mut self.copper_runtime.copperlists_manager;
3250 cl_manager.finish_pending_with(|slot| {
3251 let index = slot.index();
3252 free_copperlists[index] = Some(slot);
3253 })?;
3254 returned_copperlists.extend(free_copperlists.into_iter().flatten());
3255 debug_assert_eq!(returned_copperlists.len(), #copperlist_count_tokens);
3256 cl_manager.return_execution_slots(returned_copperlists);
3257 Ok(())
3258 })();
3259 execution_result.and(cleanup_result)
3260 });
3261
3262 if result.is_err() {
3263 error!("A task errored out: {}", &result);
3264 }
3265 if tasks_started {
3266 <Self as #app_trait<S, L>>::stop_all_tasks(self, #sim_callback_arg)?;
3267 }
3268 let _ = self.log_shutdown_completed();
3269 result
3270 }
3271 } else {
3272 quote! {
3273 static STOP_FLAG: AtomicBool = AtomicBool::new(false);
3274
3275 #kill_handler
3276
3277 <Self as #app_trait<S, L>>::start_all_tasks(self, #sim_callback_arg)?;
3278 let result = #run_loop;
3279
3280 if result.is_err() {
3281 error!("A task errored out: {}", &result);
3282 }
3283 <Self as #app_trait<S, L>>::stop_all_tasks(self, #sim_callback_arg)?;
3284 let _ = self.log_shutdown_completed();
3285 result
3286 }
3287 };
3288 let keyframe_manager_binding = keyframe_logging_enabled
3289 .then(|| quote! { let kf_manager = &mut runtime.keyframes_manager; });
3290 let keyframe_reset =
3291 keyframe_logging_enabled.then(|| quote! { kf_manager.try_reset(clid, clock)?; });
3292 let keyframe_finish =
3293 keyframe_logging_enabled.then(|| quote! { kf_manager.end_of_processing(clid)?; });
3294 let keyframe_bytes = if keyframe_logging_enabled {
3295 quote! { kf_manager.last_encoded_bytes }
3296 } else {
3297 quote! { 0 }
3298 };
3299 let dropped_keyframes_total = if keyframe_logging_enabled {
3300 quote! { kf_manager.dropped_keyframes_total() }
3301 } else {
3302 quote! { 0 }
3303 };
3304 let prepare_arena_keyframes = parallel_rt_run_supported.then(|| {
3305 quote! {
3306 runtime
3307 .copperlists_manager
3308 .prepare_keyframes(&kf_manager.capture_capacities);
3309 }
3310 });
3311 let keyframe_preallocation = keyframe_logging_enabled.then(|| {
3312 quote! {
3313 {
3314 let runtime = &mut self.copper_runtime;
3315 let tasks = &runtime.tasks;
3316 let __cu_bridges = &runtime.bridges;
3317 let kf_manager = &mut runtime.keyframes_manager;
3318 kf_manager.begin_capture_preallocation();
3319 #(#keyframe_preallocation_code)*
3320 kf_manager.finish_capture_preallocation()?;
3321 #prepare_arena_keyframes
3322 }
3323 }
3324 });
3325 let keyframe_finish_pending = keyframe_logging_enabled.then(|| {
3326 quote! { self.copper_runtime.keyframes_manager.finish_pending()?; }
3327 });
3328
3329 let live_completed = (sim_mode && logstream_enabled).then(|| {
3330 quote! {
3331 let _ = sim_callback(SimStep::CopperListCompleted(culist));
3332 }
3333 });
3334 let live_replay_impl = if sim_mode && logstream_enabled {
3335 live_twin::runtime_support(
3336 application_name,
3337 &mission_mod,
3338 mission_index,
3339 &culist_plan,
3340 &culist_exec_entities,
3341 )
3342 } else {
3343 quote! {}
3344 };
3345 let pipeline_iteration_keyframe = keyframe_logging_enabled.then(|| {
3346 quote! {
3347 if let Some(keyframe) = culist.keyframe_mut() {
3348 keyframe.reset_distributed(
3349 clid,
3350 clock.now(),
3351 runtime.keyframes_manager.capture_interval(),
3352 );
3353 }
3354 let (keyframe_ptr, keyframe_len) = culist.keyframe_capture();
3355 }
3356 });
3357 let pipeline_iteration_keyframe_field =
3358 keyframe_logging_enabled.then(|| quote! { keyframe_ptr, keyframe_len, });
3359 let iteration_body = if parallel_rt_run_supported {
3360 quote! {
3361 let runtime = &mut self.copper_runtime;
3364 let clock = runtime.clock();
3365 let instance_id = runtime.instance_id();
3366 let subsystem_code = runtime.subsystem_code();
3367 let task_ptrs: #mission_mod::ParallelTaskPtrs = #parallel_task_ptr_values;
3368 let bridge_ptrs: #mission_mod::ParallelBridgePtrs = #parallel_bridge_ptr_values;
3369 let monitor = &runtime.monitor;
3370 let cl_manager = &mut runtime.copperlists_manager;
3371 let clid = cl_manager.next_cl_id();
3372 let mut culist = cl_manager.acquire_iteration_slot()?;
3373 cl_manager.note_admission(clid + 1);
3374 culist.reset(clid);
3375 culist.change_state(cu29::copperlist::CopperListState::Processing);
3376 #pipeline_iteration_keyframe
3377 let mut step_rt = #mission_mod::ParallelProcessStepRuntime {
3378 clock: &clock,
3379 execution_probe: runtime.execution_probe.as_ref(),
3380 monitor,
3381 task_ptrs: &task_ptrs,
3382 bridge_ptrs: &bridge_ptrs,
3383 #pipeline_iteration_keyframe_field
3384 culist: culist.as_ptr(),
3385 clid,
3386 ctx: cu29::context::CuContext::from_runtime_metadata(
3387 clock.clone(), clid, instance_id, subsystem_code,
3388 #mission_mod::TASK_IDS,
3389 ),
3390 };
3391 #(
3392 match #parallel_process_step_idents(&mut step_rt) {
3393 Ok(cu29::curuntime::ProcessStepOutcome::Continue) => {}
3394 Ok(cu29::curuntime::ProcessStepOutcome::AbortCopperList) => {
3395 if let Some(keyframe) = culist.keyframe_mut() {
3396 keyframe.discard_distributed();
3397 }
3398 step_rt.ctx.clear_current_component();
3399 step_rt.ctx.clear_current_task();
3400 let monitor_result = monitor.process_copperlist(
3401 &step_rt.ctx,
3402 #mission_mod::MONITOR_LAYOUT.view(&#mission_mod::collect_metadata(&culist)),
3403 );
3404 drop(step_rt);
3405 if let cu29::curuntime::OwnedCopperListSubmission::Recycled(slot) =
3406 cl_manager.submit_slot(culist)?
3407 {
3408 cl_manager.recycle_slot(slot);
3409 }
3410 monitor_result?;
3411 return Ok(());
3412 }
3413 Err(error) => {
3414 drop(step_rt);
3415 cl_manager.recycle_slot(culist);
3416 return Err(error);
3417 }
3418 }
3419 )*
3420 step_rt.ctx.clear_current_component();
3421 step_rt.ctx.clear_current_task();
3422 let monitor_result = monitor.process_copperlist(
3423 &step_rt.ctx,
3424 #mission_mod::MONITOR_LAYOUT.view(&#mission_mod::collect_metadata(&culist)),
3425 );
3426 drop(step_rt);
3427 #(#preprocess_logging_calls)*
3428 if let cu29::curuntime::OwnedCopperListSubmission::Recycled(slot) =
3429 cl_manager.submit_slot(culist)?
3430 {
3431 cl_manager.recycle_slot(slot);
3432 }
3433 monitor_result?;
3434 monitor.observe_copperlist_io(cu29::monitoring::CopperListIoStats {
3435 raw_culist_bytes: core::mem::size_of::<CuList>() as u64
3436 + cl_manager.last_handle_bytes,
3437 handle_bytes: cl_manager.last_handle_bytes,
3438 encoded_culist_bytes: cl_manager.last_encoded_bytes,
3439 keyframe_bytes: cl_manager.last_keyframe_bytes,
3440 structured_log_bytes_total: ::cu29::prelude::structured_log_bytes_total(),
3441 culistid: clid,
3442 dropped_copperlists_total: cl_manager.dropped_copperlists_total(),
3443 dropped_keyframes_total: 0,
3444 });
3445 Ok(())
3446 }
3447 } else {
3448 quote! {
3449 let runtime = &mut self.copper_runtime;
3451 let clock_handle = runtime.clock();
3452 let clock = &clock_handle;
3453 let instance_id = runtime.instance_id();
3454 let subsystem_code = runtime.subsystem_code();
3455 let execution_probe = &runtime.execution_probe;
3456 let monitor = &mut runtime.monitor;
3457 let tasks = &mut runtime.tasks;
3458 let __cu_bridges = &mut runtime.bridges;
3459 let cl_manager = &mut runtime.copperlists_manager;
3460 #keyframe_manager_binding
3461 let iteration_clid = cl_manager.next_cl_id();
3462 let mut ctx = cu29::context::CuContext::from_runtime_metadata(
3463 clock.clone(), iteration_clid, instance_id, subsystem_code,
3464 #mission_mod::TASK_IDS,
3465 );
3466 let mut __cu_abort_copperlist = false;
3467 #(#preprocess_calls)*
3468 let culist = cl_manager.create()?;
3469 let clid = culist.id;
3470 debug_assert_eq!(clid, iteration_clid);
3471 #keyframe_reset
3472 culist.change_state(cu29::copperlist::CopperListState::Processing);
3473 let mut ctx = cu29::context::CuContext::from_runtime_metadata(
3474 clock.clone(), iteration_clid, instance_id, subsystem_code,
3475 #mission_mod::TASK_IDS,
3476 );
3477 {
3478 let msgs = &mut culist.msgs.0;
3479 #[allow(unused_macros)]
3480 macro_rules! __cu_input {
3481 ($field:tt) => { &msgs.$field };
3482 ($field:tt, $port:tt) => { &msgs.$field.$port };
3483 }
3484 #[allow(unused_macros)]
3485 macro_rules! __cu_output {
3486 ($field:tt) => { &mut msgs.$field };
3487 ($field:tt, $port:tt) => { &mut msgs.$field.$port };
3488 }
3489 #(#anytime_job_locals)*
3490 '__cu_process_steps: {
3491 #(#runtime_plan_code)*
3492 }
3493 }
3494 if __cu_abort_copperlist {
3495 ctx.clear_current_component();
3496 ctx.clear_current_task();
3497 let monitor_result = monitor.process_copperlist(
3498 &ctx,
3499 #mission_mod::MONITOR_LAYOUT.view(&#mission_mod::collect_metadata(&culist)),
3500 );
3501 cl_manager.end_of_processing(clid)?;
3502 monitor_result?;
3503 return Ok(());
3504 }
3505 ctx.clear_current_component();
3506 ctx.clear_current_task();
3507 let monitor_result = monitor.process_copperlist(
3508 &ctx,
3509 #mission_mod::MONITOR_LAYOUT.view(&#mission_mod::collect_metadata(&culist)),
3510 );
3511 #(#preprocess_logging_calls)*
3512 #live_completed
3513 cl_manager.end_of_processing(clid)?;
3514 monitor_result?;
3515 #(#postprocess_calls)*
3516 #keyframe_finish
3517 monitor.observe_copperlist_io(cu29::monitoring::CopperListIoStats {
3518 raw_culist_bytes: core::mem::size_of::<CuList>() as u64
3519 + cl_manager.last_handle_bytes,
3520 handle_bytes: cl_manager.last_handle_bytes,
3521 encoded_culist_bytes: cl_manager.last_encoded_bytes,
3522 keyframe_bytes: #keyframe_bytes,
3523 structured_log_bytes_total: ::cu29::prelude::structured_log_bytes_total(),
3524 culistid: clid,
3525 dropped_copperlists_total: cl_manager.dropped_copperlists_total(),
3526 dropped_keyframes_total: #dropped_keyframes_total,
3527 });
3528 Ok(())
3529 }
3530 };
3531 let stop_components_sig = if sim_mode {
3532 quote! { fn stop_components(&mut self, sim_callback: &mut impl FnMut(SimStep) -> SimOverride) -> CuResult<()> }
3533 } else { quote! { fn stop_components(&mut self) -> CuResult<()> } };
3534 let stop_components = quote! {
3535 #stop_components_sig {
3536 let result: CuResult<()> = (|| {
3537 let lifecycle_clid = self.copper_runtime.copperlists_manager.last_cl_id();
3538 let mut ctx = cu29::context::CuContext::from_runtime_metadata(
3539 self.copper_runtime.clock(),
3540 lifecycle_clid,
3541 self.copper_runtime.instance_id(),
3542 self.copper_runtime.subsystem_code(),
3543 #mission_mod::TASK_IDS,
3544 );
3545 #(#stop_calls)*
3546 ctx.clear_current_component();
3547 ctx.clear_current_task();
3548 self.copper_runtime.monitor.stop(&ctx)?;
3549 self.copper_runtime.copperlists_manager.finish_pending()?;
3550 #keyframe_finish_pending
3551 Ok(())
3552 })();
3553 let clock_result: CuResult<()> = (|| { #clock_stop Ok(()) })();
3554 let result = result.and(clock_result);
3555 match result {
3556 Ok(()) => {
3557 self.log_runtime_lifecycle_event(RuntimeLifecycleEvent::MissionStopped { reason: self.stop_reason })?;
3558 self.lifecycle_running = false;
3559 self.stop_reason = cu29::curuntime::RuntimeStopReason::Requested;
3560 Ok(())
3561 }
3562 Err(error) => {
3563 let _ = self.log_runtime_lifecycle_event(RuntimeLifecycleEvent::LifecycleFailed {
3564 operation: cu29::curuntime::RuntimeLifecycleOperation::Stop, error: error.to_string(),
3565 });
3566 Err(error)
3567 }
3568 }
3569 }
3570 };
3571 let drop_stop = if sim_mode {
3572 quote! { self.stop_components(&mut |_| SimOverride::ExecuteByRuntime) }
3573 } else { quote! { self.stop_components() } };
3574 let drop_panic = std.then(|| quote! {
3575 if std::thread::panicking() {
3576 self.stop_reason = cu29::curuntime::RuntimeStopReason::Panic;
3577 let _ = self.log_runtime_lifecycle_event(RuntimeLifecycleEvent::Panic {
3578 message: "Application is unwinding after a panic".to_string(), file: None, line: None, column: None,
3579 });
3580 }
3581 });
3582 let app_drop_impl = quote! {
3583 impl Drop for #application_name {
3584 fn drop(&mut self) {
3585 if self.shutdown_completed { return; }
3586 #drop_panic
3587 if self.lifecycle_running && #drop_stop.is_err() { return; }
3588 if let Err(error) = self.log_shutdown_completed() {
3589 let _ = self.log_runtime_lifecycle_event(RuntimeLifecycleEvent::LifecycleFailed {
3590 operation: cu29::curuntime::RuntimeLifecycleOperation::Shutdown, error: error.to_string(),
3591 });
3592 }
3593 }
3594 }
3595 };
3596 let run_methods: proc_macro2::TokenStream = quote! {
3597
3598 #run_one_iteration {
3599 let result: CuResult<()> = (|| { #clock_maintain #iteration_body })();
3600 if let Err(error) = &result {
3601 let _ = self.log_runtime_lifecycle_event(RuntimeLifecycleEvent::LifecycleFailed {
3602 operation: cu29::curuntime::RuntimeLifecycleOperation::Iteration, error: error.to_string(),
3603 });
3604 }
3605 result
3606 }
3607
3608 fn restore_keyframe(&mut self, keyframe: &KeyFrame) -> CuResult<()> {
3609 let runtime = &mut self.copper_runtime;
3610 let clock_handle = runtime.clock();
3611 let clock = &clock_handle;
3612 let tasks = &mut runtime.tasks;
3613 let __cu_bridges = &mut runtime.bridges;
3614 let mut frames = cu29::curuntime::KeyFramePayloadReader::new(keyframe)?;
3615 #(#keyframe_restore_code)*
3616 frames.finish()?;
3617 Ok(())
3618 }
3619
3620 #start_all_tasks {
3621 self.lifecycle_running = true;
3622 let result: CuResult<()> = (|| {
3623 #clock_start
3624 let lifecycle_clid = self.copper_runtime.copperlists_manager.last_cl_id();
3625 let mut ctx = cu29::context::CuContext::from_runtime_metadata(
3626 self.copper_runtime.clock(),
3627 lifecycle_clid,
3628 self.copper_runtime.instance_id(),
3629 self.copper_runtime.subsystem_code(),
3630 #mission_mod::TASK_IDS,
3631 );
3632 #(#start_calls)*
3633 #keyframe_preallocation
3634 ctx.clear_current_component();
3635 ctx.clear_current_task();
3636 self.copper_runtime.monitor.start(&ctx)?;
3637 Ok(())
3638 })();
3639 match result {
3640 Ok(()) => {
3641 self.log_runtime_lifecycle_event(RuntimeLifecycleEvent::MissionStarted)?;
3642 self.lifecycle_running = true;
3643 Ok(())
3644 }
3645 Err(error) => {
3646 let _ = self.log_runtime_lifecycle_event(RuntimeLifecycleEvent::LifecycleFailed {
3647 operation: cu29::curuntime::RuntimeLifecycleOperation::Start, error: error.to_string(),
3648 });
3649 Err(error)
3650 }
3651 }
3652 }
3653
3654 #stop_all_tasks { self.stop_components(#sim_callback_arg) }
3655
3656 #run {
3657 #run_body
3658 }
3659 };
3660
3661 let tasks_type = if sim_mode {
3662 quote!(CuSimTasks)
3663 } else {
3664 quote!(CuTasks)
3665 };
3666
3667 let tasks_instanciator_fn = if sim_mode {
3668 quote!(tasks_instanciator_sim)
3669 } else {
3670 quote!(tasks_instanciator)
3671 };
3672
3673 let app_impl_decl = if sim_mode {
3674 quote!(impl<S: SectionStorage + 'static, L: UnifiedLogWrite<S> + 'static> CuSimApplication<S, L> for #application_name)
3675 } else {
3676 quote!(impl<S: SectionStorage + 'static, L: UnifiedLogWrite<S> + 'static> CuApplication<S, L> for #application_name)
3677 };
3678
3679 let simstep_type_decl = if sim_mode {
3680 quote!(
3681 type Step<'z> = SimStep<'z>;
3682 )
3683 } else {
3684 quote!()
3685 };
3686
3687 let mission_id_method = if sim_mode {
3688 quote! {
3689 fn mission_id() -> Option<&'static str> {
3690 Some(#mission)
3691 }
3692 }
3693 } else {
3694 quote!()
3695 };
3696
3697 let app_resources_thread_pools_field = if std {
3698 quote! { pub thread_pools: Vec<Option<Arc<ThreadPool>>>, }
3699 } else {
3700 quote!()
3701 };
3702
3703 let app_resources_struct = quote! {
3704 pub struct AppResources {
3705 pub config: CuConfig,
3706 pub config_source: RuntimeLifecycleConfigSource,
3707 pub log_context: Option<cu29::prelude::SectionContext>,
3708 pub resources: ResourceManager,
3709 #app_resources_thread_pools_field
3710 #clock_maintenance_field
3711 }
3712 };
3713
3714 let prepare_config_fn = quote! {
3715 #prepare_config_sig {
3716 let config_filename = #config_file;
3717
3718 #[cfg(target_os = "none")]
3719 ::cu29::prelude::info!("CuApp init: config file {}", config_filename);
3720 #[cfg(target_os = "none")]
3721 ::cu29::prelude::info!("CuApp init: loading config");
3722 #config_load_stmt
3723 let mut config = config;
3724 if #sim_mode { config.log_streaming = None; }
3725 #constant_override_warning
3726 #copperlist_count_check
3727 #planner_config_check
3728 #keyframe_logging_check
3729 #logstream_topology_check
3730 #[cfg(target_os = "none")]
3731 ::cu29::prelude::info!("CuApp init: config loaded");
3732 if let Some(runtime) = &config.runtime {
3733 #[cfg(target_os = "none")]
3734 ::cu29::prelude::info!(
3735 "CuApp init: rate_target_hz={}",
3736 runtime.rate_target_hz.unwrap_or(0)
3737 );
3738 } else {
3739 #[cfg(target_os = "none")]
3740 ::cu29::prelude::info!("CuApp init: rate_target_hz=none");
3741 }
3742
3743 Ok((config, config_source))
3744 }
3745 };
3746
3747 let prepare_resources_thread_pools_stmt = if std {
3748 quote! {
3749 let thread_pools = #mission_mod::thread_pools_instanciator(&config)?;
3750 }
3751 } else {
3752 quote!()
3753 };
3754 let prepare_resources_thread_pools_init = if std {
3755 quote! { thread_pools, }
3756 } else {
3757 quote!()
3758 };
3759
3760 let prepare_resources_fn = quote! {
3761 #prepare_resources_sig {
3762 let (config, config_source) = #prepare_config_call;
3763
3764 #[cfg(target_os = "none")]
3765 ::cu29::prelude::info!("CuApp init: building resources");
3766 let resources = #mission_mod::resources_instanciator(&config)?;
3767 #prepare_resources_thread_pools_stmt
3768 #[cfg(target_os = "none")]
3769 ::cu29::prelude::info!("CuApp init: resources ready");
3770
3771 Ok(AppResources {
3772 config,
3773 config_source,
3774 log_context: None,
3775 resources,
3776 #prepare_resources_thread_pools_init
3777 #clock_maintenance_none
3778 })
3779 }
3780 };
3781
3782 let no_logstream_arg = logstream_enabled.then(|| quote! { None, });
3783 let builder_logstream_arg = logstream_enabled.then(|| quote! { self.logstream, });
3784 let new_with_resources_compat_fn = if sim_mode {
3785 quote! {
3786 pub fn new_with_resources<S: SectionStorage + 'static, L: UnifiedLogWrite<S> + 'static>(
3787 clock: RobotClock,
3788 unified_logger: Arc<Mutex<L>>,
3789 app_resources: AppResources,
3790 instance_id: u32,
3791 sim_callback: &mut impl FnMut(SimStep) -> SimOverride,
3792 ) -> CuResult<Self> {
3793 Self::build_with_resources(
3794 clock,
3795 unified_logger,
3796 app_resources,
3797 instance_id,
3798 #no_logstream_arg
3799 sim_callback,
3800 )
3801 }
3802 }
3803 } else {
3804 quote! {
3805 pub fn new_with_resources<S: SectionStorage + 'static, L: UnifiedLogWrite<S> + 'static>(
3806 clock: RobotClock,
3807 unified_logger: Arc<Mutex<L>>,
3808 app_resources: AppResources,
3809 instance_id: u32,
3810 ) -> CuResult<Self> {
3811 Self::build_with_resources(
3812 clock,
3813 unified_logger,
3814 app_resources,
3815 instance_id,
3816 #no_logstream_arg
3817 )
3818 }
3819 }
3820 };
3821
3822 let build_with_resources_thread_pools_destructure = if std {
3823 quote! { thread_pools, }
3824 } else {
3825 quote!()
3826 };
3827 let build_with_resources_thread_pools_call = if std {
3828 quote! { .with_thread_pools(thread_pools) }
3829 } else {
3830 quote!()
3831 };
3832 let local_keyframe_sink_init = if local_keyframe_logging_enabled {
3833 quote! {
3834 #[cfg(target_os = "none")]
3835 ::cu29::prelude::info!("CuApp new: creating keyframes stream");
3836 let local_keyframe_sink = cu29::prelude::stream_write_context::<KeyFrame, S>(
3837 unified_logger.clone(),
3838 UnifiedLogType::FrozenTasks,
3839 1024 * 1024 * 10, log_context, )?;
3841 #[cfg(target_os = "none")]
3842 ::cu29::prelude::info!("CuApp new: keyframes stream ready");
3843 }
3844 } else {
3845 quote! {
3846 let local_keyframe_sink = cu29::curuntime::NullWriteStream;
3847 }
3848 };
3849 let configured_logstream_session = (!logstream_resource_specs.is_empty()).then(|| {
3850 quote! {
3851 let logstream_session_id = ::cu29::logstream::new_session_id();
3852 let logstream_schema = <#mission_mod::CuStampedDataSet as ::cu29::logstream::capture::CaptureDataSet>::stream_schema();
3853 }
3854 });
3855 let configured_logstream_initializers = logstream_resource_specs.iter().map(|spec| {
3856 let index = syn::Index::from(spec.destination_index);
3857 let bundle_index = spec.stream.bundle_index;
3858 let provider_path = &spec.stream.provider_path;
3859 let resource_name = LitStr::new(&spec.stream.resource_name, Span::call_site());
3860 let transport_type = &spec.stream.transport_type;
3861 let transport_ident = format_ident!("__cu_logstream_transport_{}", spec.destination_index);
3862 let continuous_ident = format_ident!("__cu_logstream_continuous_{}", spec.destination_index);
3863 let recovery_ident = format_ident!("__cu_logstream_recovery_{}", spec.destination_index);
3864 let scheduled = if let Some(feedback) = &spec.feedback {
3865 let bundle = feedback.bundle_index;
3866 let provider = &feedback.provider_path;
3867 let name = LitStr::new(&feedback.resource_name, Span::call_site());
3868 let ty = &feedback.transport_type;
3869 quote! {
3870 let feedback_rx: #ty = resources.take(
3871 cu29::resource::ResourceKey::<()>::new(cu29::resource::BundleIndex::new(#bundle),
3872 cu29::resource::resource_index_by_name::<#provider>(#name)).typed::<#ty>()
3873 )?.0;
3874 let (mut #continuous_ident, #recovery_ident, sender_monitor) = ::cu29::logstream::scheduled_feedback_sinks::<#mission_mod::CuStampedDataSet, _>(
3875 ::cu29::logstream::SeparateFeedback { tx: #transport_ident, feedback_rx }, sender_config,
3876 if clock.is_mock() { RobotClock::new() } else { clock.clone() },
3877 )?;
3878 }
3879 } else {
3880 quote! {
3881 let (mut #continuous_ident, #recovery_ident, sender_monitor) = ::cu29::logstream::scheduled_sinks::<#mission_mod::CuStampedDataSet, _>(
3882 #transport_ident, sender_config,
3883 if clock.is_mock() { RobotClock::new() } else { clock.clone() },
3884 )?;
3885 }
3886 };
3887 quote! {
3888 let destination = &config
3889 .log_streaming
3890 .as_ref()
3891 .expect("compiled logstream configuration is missing")
3892 .destinations[#index];
3893 let plan = ::cu29::logstream::LogStreamPlan::resolve(destination)
3894 .map_err(|error| CuError::from(error.to_string()))?;
3895 let sender_config = plan
3896 .sender_config(
3897 ::cu29::logstream::StreamIdentity {
3898 session_id: logstream_session_id,
3899 sender_id: instance_id,
3900 },
3901 logstream_schema.clone(),
3902 log_context,
3903 )
3904 .map_err(|error| CuError::from(error.to_string()))?;
3905 let #transport_ident: #transport_type = resources
3906 .take(
3907 cu29::resource::ResourceKey::<()>::new(
3908 cu29::resource::BundleIndex::new(#bundle_index),
3909 cu29::resource::resource_index_by_name::<#provider_path>(#resource_name),
3910 )
3911 .typed::<#transport_type>(),
3912 )?
3913 .0;
3914 #scheduled
3915 let structured_outputs = (#continuous_ident.take_structured_log_sink(), structured_outputs);
3916 stream_monitors.push(sender_monitor.into_runtime_monitor(&destination.id,
3917 destination.link.bitrate_bps, usize::from(destination.fec.continuous.repair_every_source_symbols)));
3918 let #continuous_ident = if logstream_schema.reconstruction.is_empty() {
3919 #continuous_ident
3920 } else {
3921 #continuous_ident.with_encoder(::cu29::logstream::capture::encode_capture_record_into)
3922 };
3923 }
3924 });
3925 let configured_logstream_fanouts = logstream_resource_specs.iter().map(|spec| {
3926 let continuous_ident =
3927 format_ident!("__cu_logstream_continuous_{}", spec.destination_index);
3928 let recovery_ident =
3929 format_ident!("__cu_logstream_recovery_{}", spec.destination_index);
3930 quote! {
3931 let copperlist_sink = ::cu29::fanout::StaticFanoutSink::new(
3932 copperlist_sink,
3933 #continuous_ident,
3934 );
3935 let keyframe_sink = ::cu29::fanout::StaticFanoutSink::new(
3936 keyframe_sink,
3937 #recovery_ident,
3938 );
3939 }
3940 });
3941 let configured_logstream_count = logstream_resource_specs.len();
3942 let copperlist_output_graph = if logstream_enabled {
3943 quote! {
3944 let local_copperlist_output_required = config
3945 .logging
3946 .as_ref()
3947 .is_none_or(|logging| logging.enable_task_logging);
3948 let local_keyframe_output_required = config
3949 .logging
3950 .as_ref()
3951 .is_none_or(|logging| {
3952 logging.enable_task_logging && logging.enable_keyframe_logging
3953 });
3954 let logstream_output_required = #configured_logstream_count != 0 || logstream.is_some();
3955 #[allow(unused_mut)]
3956 let mut resources = resources;
3957 let mut stream_monitors = Vec::with_capacity(#configured_logstream_count + usize::from(logstream.is_some()));
3958 let structured_outputs = ();
3959 #configured_logstream_session
3960 #(#configured_logstream_initializers)*
3961 let injected_logstream_sinks = logstream
3962 .map(|(transport, mut sender_config)| {
3963 sender_config.continuous.identity.sender_id = instance_id;
3964 sender_config.recovery.finite.identity.sender_id = instance_id;
3965 let schema = <#mission_mod::CuStampedDataSet as ::cu29::logstream::capture::CaptureDataSet>::stream_schema();
3966 let mut manifest = ::cu29::logstream::SessionManifest::decode_record(&sender_config.recovery.manifest_record)
3967 .map_err(|e| CuError::from(e.to_string()))?;
3968 if manifest.application_schema != schema {
3969 return Err(CuError::from("injected sender must use the generated application schema"));
3970 }
3971 manifest.identity = sender_config.continuous.identity;
3972 manifest.context = log_context;
3973 sender_config.recovery.manifest_record = manifest.encode_record()
3974 .map_err(|e| CuError::from(e.to_string()))?;
3975 let bitrate = sender_config.pacing.bitrate_bps;
3976 let baseline = sender_config.continuous.repair_every_source_symbols;
3977 let (mut copperlist, keyframe, sender_monitor) = ::cu29::logstream::scheduled_sinks::<
3978 #mission_mod::CuStampedDataSet, _
3979 >(transport, sender_config,
3980 if clock.is_mock() { RobotClock::new() } else { clock.clone() })?;
3981 stream_monitors.push(sender_monitor.into_runtime_monitor("injected", bitrate, baseline));
3982 let structured = copperlist.take_structured_log_sink();
3983 let copperlist = if schema.reconstruction.is_empty() { copperlist }
3984 else { copperlist.with_encoder(::cu29::logstream::capture::encode_capture_record_into) };
3985 Ok::<_, CuError>(((copperlist, keyframe), structured))
3986 })
3987 .transpose()?;
3988 let stream_monitors: Option<::std::sync::Arc<[cu29::monitoring::LogStreamMonitor]>> =
3989 (!stream_monitors.is_empty()).then(|| stream_monitors.into());
3990 let (injected_logstream_sinks, injected_structured) = injected_logstream_sinks.unzip();
3991 let structured_outputs = (injected_structured.flatten(), structured_outputs);
3992 let (injected_logstream_sink, injected_logstream_keyframe_sink) = injected_logstream_sinks.unzip();
3993 let copperlist_sink = ::cu29::fanout::OptionalWriteStream::new(
3994 local_copperlist_output_required.then_some(local_copperlist_sink),
3995 );
3996 let keyframe_sink = ::cu29::fanout::OptionalWriteStream::new(
3997 local_keyframe_output_required.then_some(local_keyframe_sink),
3998 );
3999 #(#configured_logstream_fanouts)*
4000 let copperlist_sink = ::cu29::fanout::StaticFanoutSink::new(
4001 copperlist_sink,
4002 ::cu29::fanout::OptionalWriteStream::new(injected_logstream_sink),
4003 );
4004 let keyframe_sink = ::cu29::fanout::StaticFanoutSink::new(
4005 keyframe_sink,
4006 ::cu29::fanout::OptionalWriteStream::new(injected_logstream_keyframe_sink),
4007 );
4008 let output_requirements = cu29::curuntime::OutputRequirements::new(
4009 local_copperlist_output_required || logstream_output_required,
4010 local_keyframe_output_required || logstream_output_required,
4011 );
4012 }
4013 } else {
4014 quote! {
4015 let local_copperlist_output_required = config
4017 .logging
4018 .as_ref()
4019 .is_none_or(|logging| logging.enable_task_logging);
4020 let local_keyframe_output_required = config
4021 .logging
4022 .as_ref()
4023 .is_none_or(|logging| {
4024 logging.enable_task_logging && logging.enable_keyframe_logging
4025 });
4026 let output_requirements = cu29::curuntime::OutputRequirements::new(
4027 local_copperlist_output_required,
4028 local_keyframe_output_required,
4029 );
4030 let copperlist_sink = local_copperlist_sink;
4031 let keyframe_sink = local_keyframe_sink;
4032 }
4033 };
4034
4035 let monitor_instanciator = if logstream_enabled {
4036 quote! { |config: &CuConfig, metadata: cu29::monitoring::CuMonitoringMetadata, runtime: cu29::monitoring::CuMonitoringRuntime| {
4037 let runtime = match &stream_monitors {
4038 Some(streams) => runtime.with_log_streams(streams.clone()),
4039 None => runtime,
4040 };
4041 #mission_mod::monitor_instanciator(config, metadata, runtime)
4042 } }
4043 } else {
4044 quote! { #mission_mod::monitor_instanciator }
4045 };
4046
4047 let local_structured_logger_init = quote! {
4048 let local_structured_log_sink = ::cu29::prelude::LogStream::with_context(
4049 ::cu29::prelude::UnifiedLogType::StructuredLogLine,
4050 unified_logger.clone(), 4096 * 10, log_context,
4051 )?;
4052 };
4053 let (early_structured_logger_init, late_structured_logger_init) = if logstream_enabled {
4054 (
4055 quote! {},
4056 quote! {
4057 #local_structured_logger_init
4058 let logger_runtime = if logstream_output_required {
4059 ::cu29::prelude::LoggerRuntime::init(
4060 clock.clone(),
4061 ::cu29::logstream::StructuredLogStream::new(local_structured_log_sink, structured_outputs),
4062 None::<::cu29::prelude::NullLog>,
4063 )
4064 } else {
4065 ::cu29::prelude::LoggerRuntime::init(clock.clone(), local_structured_log_sink, None::<::cu29::prelude::NullLog>)
4066 };
4067 },
4068 )
4069 } else {
4070 (
4071 quote! {
4072 #local_structured_logger_init
4073 let logger_runtime = ::cu29::prelude::LoggerRuntime::init(
4074 clock.clone(), local_structured_log_sink, None::<::cu29::prelude::NullLog>,
4075 );
4076 },
4077 quote! {},
4078 )
4079 };
4080
4081 let build_with_resources_fn = quote! {
4082 #build_with_resources_sig {
4083 let AppResources {
4084 config,
4085 config_source,
4086 log_context,
4087 resources,
4088 #clock_maintenance_destructure
4089 #build_with_resources_thread_pools_destructure
4090 } = app_resources;
4091
4092 #[allow(unused_mut)]
4093 let mut resources = resources;
4094 #clock_maintenance_compat
4095 let log_context = match log_context {
4096 Some(context) => context,
4097 None => Self::prepare_log_metadata::<S, L>(&unified_logger, &config, instance_id)?,
4098 };
4099 #early_structured_logger_init
4100
4101 let mut default_section_size = size_of::<super::#mission_mod::CuList>() * 64;
4104 if let Some(section_size_mib) = config.logging.as_ref().and_then(|l| l.section_size_mib) {
4106 default_section_size = section_size_mib as usize * 1024usize * 1024usize;
4108 }
4109 #[cfg(target_os = "none")]
4110 ::cu29::prelude::info!(
4111 "CuApp new: copperlist section size={}",
4112 default_section_size
4113 );
4114 #[cfg(target_os = "none")]
4115 ::cu29::prelude::info!("CuApp new: creating copperlist stream");
4116 let local_copperlist_sink = cu29::prelude::stream_write_context::<#mission_mod::CuList, S>(
4117 unified_logger.clone(),
4118 UnifiedLogType::CopperList,
4119 default_section_size, log_context,
4120 )?;
4124 #[cfg(target_os = "none")]
4125 ::cu29::prelude::info!("CuApp new: copperlist stream ready");
4126
4127 #local_keyframe_sink_init
4128
4129 #copperlist_output_graph
4130 #late_structured_logger_init
4131
4132 #[cfg(target_os = "none")]
4133 ::cu29::prelude::info!("CuApp new: creating runtime lifecycle stream");
4134 let effective_config_ron = config.serialize_ron()?;
4135 ::cu29::logcodec::set_effective_config_ron::<super::#mission_mod::CuStampedDataSet>(&effective_config_ron);
4136 let mut local_lifecycle_sink = cu29::prelude::stream_write_context::<RuntimeLifecycleRecord, S>(
4137 unified_logger.clone(), UnifiedLogType::RuntimeLifecycle, 4096, log_context,
4138 )?;
4139
4140 #[cfg(target_os = "none")]
4141 ::cu29::prelude::info!("CuApp new: building runtime");
4142 let copper_runtime = CuRuntimeBuilder::<#mission_mod::#tasks_type, #mission_mod::CuBridges, #mission_mod::CuStampedDataSet, #monitor_type, #copperlist_count_tokens, _, _, _, _, _>::new(
4143 clock,
4144 &config,
4145 #mission,
4146 CuRuntimeParts::new(
4147 #mission_mod::#tasks_instanciator_fn,
4148 #mission_mod::MONITORED_COMPONENTS,
4149 #mission_mod::CULIST_COMPONENT_MAPPING,
4150 #parallel_rt_metadata_arg
4151 #monitor_instanciator,
4152 #mission_mod::bridges_instanciator,
4153 ),
4154 copperlist_sink,
4155 keyframe_sink,
4156 output_requirements,
4157 )
4158 .with_subsystem(#application_name::subsystem())
4159 .with_instance_id(instance_id)
4160 .with_resources(resources)
4161 #build_with_resources_thread_pools_call
4162 .build()?;
4163 #[cfg(target_os = "none")]
4164 ::cu29::prelude::info!("CuApp new: runtime built");
4165
4166 local_lifecycle_sink.log(&RuntimeLifecycleRecord {
4167 timestamp: copper_runtime.clock_ref().now(),
4168 event: RuntimeLifecycleEvent::Instantiated { config_source },
4169 })?;
4170 let application = Ok(#application_name {
4171 copper_runtime,
4172 #clock_maintenance_init
4173 #replay_clock_init
4174 runtime_lifecycle_sink: Some(Box::new(local_lifecycle_sink)),
4175 lifecycle_running: false,
4176 shutdown_completed: false,
4177 stop_reason: cu29::curuntime::RuntimeStopReason::Requested,
4178 logger_runtime,
4179 });
4180
4181 #sim_callback_on_new
4182
4183 application
4184 }
4185 };
4186
4187 let metadata_log_lock = if std {
4188 quote! { let mut logger = logger.lock().map_err(|_| CuError::from("Logger mutex poisoned"))?; }
4189 } else { quote! { let mut logger = logger.lock(); } };
4190 let seal_metadata = if cfg!(feature = "self-describing-logs") {
4191 let modules = mission_names.iter().map(|name| parse_str::<Ident>(name).unwrap()).collect::<Vec<_>>();
4192 let layout = if cfg!(feature = "flat-copperlist-encoding") { quote! { Flat } } else { quote! { Compact } };
4193 quote! {
4194 if let Some(error) = [#(super::#modules::VALUE_DECODE_CATALOG_ERROR),*].into_iter().flatten().next() {
4195 return Err(CuError::from(error));
4196 }
4197 let catalog = cu29::catalog_stream::CatalogDescription {
4198 layout: cu29::prelude::ValueDecodeCatalogLayout::#layout,
4199 missions: &[#(super::#modules::VALUE_DECODE_CATALOG_MISSION),*],
4200 };
4201 logger.seal_metadata(metadata, Some(&cu29::prelude::CompressedCatalog(&catalog)))?;
4202 }
4203 } else { quote! { logger.seal_metadata::<()>(metadata, None)?; } };
4204 let app_inherent_impl = quote! {
4205 impl #application_name {
4206 #stop_components
4207 const SUBSYSTEM: cu29::prelude::app::Subsystem =
4208 cu29::prelude::app::Subsystem::new(#subsystem_id_tokens, #subsystem_code_literal);
4209
4210 #[inline]
4211 pub fn subsystem() -> cu29::prelude::app::Subsystem {
4212 Self::SUBSYSTEM
4213 }
4214
4215 #[doc(hidden)]
4216 fn application_log_metadata(effective_config_ron: String) -> cu29::prelude::ApplicationMetadata {
4217 cu29::prelude::ApplicationMetadata {
4218 app_type: stringify!(#application_name).to_string(),
4219 app_name: env!("CARGO_PKG_NAME").to_string(),
4220 app_version: env!("CARGO_PKG_VERSION").to_string(),
4221 git_commit: #git_commit_tokens,
4222 git_dirty: #git_dirty_tokens,
4223 subsystem_id: Self::subsystem().id().map(str::to_string),
4224 subsystem_code: Self::subsystem().code(),
4225 effective_config_ron,
4226 missions: vec![#(#mission_names.to_string()),*],
4227 catalog_offset: 0,
4228 }
4229 }
4230
4231 #[doc(hidden)]
4232 fn seal_log_metadata<S: SectionStorage, L: UnifiedLogWrite<S>>(
4233 logger: &mut L, metadata: &cu29::prelude::ApplicationMetadata,
4234 ) -> CuResult<()> {
4235 #seal_metadata
4236 Ok(())
4237 }
4238
4239 #[doc(hidden)]
4240 fn prepare_log_metadata<S: SectionStorage, L: UnifiedLogWrite<S>>(
4241 logger: &Arc<Mutex<L>>, config: &CuConfig, instance_id: u32,
4242 ) -> CuResult<cu29::prelude::SectionContext> {
4243 let metadata = Self::application_log_metadata(config.serialize_ron()?);
4244 #metadata_log_lock
4245 Self::seal_log_metadata::<S, L>(&mut *logger, &metadata)?;
4246 logger.construction_context(instance_id, #mission_index)
4247 }
4248
4249 pub fn original_config() -> String {
4250 #copper_config_content.to_string()
4251 }
4252
4253 pub fn register_reflect_types(registry: &mut cu29::reflect::TypeRegistry) {
4254 #(#task_debug_state_registration_calls)*
4255 #(#reflect_type_registration_calls)*
4256 }
4257
4258 #[inline]
4260 pub fn clock(&self) -> cu29::clock::RobotClock {
4261 self.copper_runtime.clock()
4262 }
4263
4264 pub fn log_runtime_lifecycle_event(
4266 &mut self,
4267 event: RuntimeLifecycleEvent,
4268 ) -> CuResult<()> {
4269 let timestamp = self.copper_runtime.clock_ref().now();
4270 let Some(sink) = self.runtime_lifecycle_sink.as_mut() else {
4271 return Err(CuError::from("Runtime lifecycle stream is not initialized"));
4272 };
4273 sink.log(&RuntimeLifecycleRecord { timestamp, event })
4274 }
4275
4276 pub fn set_stop_reason(&mut self, reason: cu29::curuntime::RuntimeStopReason) { self.stop_reason = reason; }
4278
4279 pub fn log_shutdown_completed(&mut self) -> CuResult<()> {
4281 if self.lifecycle_running { return Err(CuError::from("Stop the application before completing shutdown")); }
4282 if self.shutdown_completed { return Ok(()); }
4283 self.copper_runtime.copperlists_manager.close_output()?;
4284 self.copper_runtime.keyframes_manager.close_output()?;
4285 self.logger_runtime.try_flush()?;
4286 self.log_runtime_lifecycle_event(RuntimeLifecycleEvent::ShutdownCompleted)?;
4287 self.shutdown_completed = true;
4288 self.runtime_lifecycle_sink.take();
4289 Ok(())
4290 }
4291
4292 #prepare_config_fn
4293 #prepare_resources_compat_fn
4294 #prepare_resources_fn
4295 #init_resources_compat_fn
4296 #new_with_resources_compat_fn
4297 #build_with_resources_fn
4298
4299 #[inline]
4301 pub fn copper_runtime_mut(&mut self) -> &mut CuRuntime<#mission_mod::#tasks_type, #mission_mod::CuBridges, #mission_mod::CuStampedDataSet, #monitor_type, #copperlist_count_tokens> {
4302 &mut self.copper_runtime
4303 }
4304 }
4305 };
4306
4307 let app_metadata_impl = quote! {
4308 impl cu29::prelude::app::CuSubsystemMetadata for #application_name {
4309 fn subsystem() -> cu29::prelude::app::Subsystem {
4310 #application_name::subsystem()
4311 }
4312 }
4313 };
4314
4315 let app_reflect_impl = quote! {
4316 impl cu29::reflect::ReflectTaskIntrospection for #application_name {
4317 fn reflect_task(&self, task_id: &str) -> Option<&dyn cu29::reflect::Reflect> {
4318 match task_id {
4319 #(#task_reflect_read_arms)*
4320 _ => None,
4321 }
4322 }
4323
4324 fn reflect_task_mut(
4325 &mut self,
4326 task_id: &str,
4327 ) -> Option<&mut dyn cu29::reflect::Reflect> {
4328 match task_id {
4329 #(#task_reflect_write_arms)*
4330 _ => None,
4331 }
4332 }
4333
4334 fn register_reflect_types(registry: &mut cu29::reflect::TypeRegistry) {
4335 #application_name::register_reflect_types(registry);
4336 }
4337
4338 fn debug_state_type_path(task_id: &str) -> Option<&'static str> {
4339 match task_id {
4340 #(#task_debug_state_type_path_arms)*
4341 _ => None,
4342 }
4343 }
4344
4345 fn with_debug_state<R>(
4346 &self,
4347 task_id: &str,
4348 f: impl FnOnce(&dyn cu29::reflect::Reflect) -> R,
4349 ) -> Option<R> {
4350 match task_id {
4351 #(#task_debug_state_read_arms)*
4352 _ => None,
4353 }
4354 }
4355 }
4356 };
4357
4358 let app_runtime_copperlist_impl = quote! {
4359 impl cu29::app::CurrentRuntimeCopperList<#mission_mod::CuStampedDataSet>
4360 for #application_name
4361 {
4362 fn current_runtime_copperlist_bytes(&self) -> Option<&[u8]> {
4363 self.copper_runtime.copperlists_manager.last_completed_encoded()
4364 }
4365
4366 fn set_current_runtime_copperlist_bytes(
4367 &mut self,
4368 snapshot: Option<Vec<u8>>,
4369 ) {
4370 self.copper_runtime
4371 .copperlists_manager
4372 .set_last_completed_encoded(snapshot);
4373 }
4374 }
4375 };
4376
4377 #[cfg(feature = "std")]
4378 #[cfg(feature = "macro_debug")]
4379 eprintln!("[build result]");
4380 let application_impl = quote! {
4381 #app_impl_decl {
4382 #simstep_type_decl
4383
4384 fn get_original_config() -> String {
4385 Self::original_config()
4386 }
4387
4388 #mission_id_method
4389
4390 #run_methods
4391 }
4392 };
4393
4394 let restore_clock_sync_method = clock_sync_enabled.then(|| quote! {
4395 fn restore_clock_sync(&mut self, record: ::cu29::clock_sync::ClockSyncRecord) -> CuResult<()> {
4396 let clock = self.copper_runtime.clock();
4397 if let Some(sync) = &mut self.replay_clock_sync {
4398 sync.restore(record.snapshot).map_err(|error| CuError::new_with_cause("Replay clock restore failed", error))?;
4399 } else {
4400 self.replay_clock_sync = Some(::cu29::clock::sync::ClockSync::from_snapshot(&clock, record.snapshot)
4401 .map_err(|error| CuError::new_with_cause("Replay clock attach failed", error))?);
4402 }
4403 Ok(())
4404 }
4405 fn set_recorded_clock_time(&mut self, clock_mock: &RobotClockMock, timestamp: CuTime) -> CuResult<()> {
4406 if let Some(sync) = &mut self.replay_clock_sync {
4407 sync.set_replay_time(clock_mock, timestamp).map_err(|error| CuError::new_with_cause("Replay clock time failed", error))?;
4408 } else { clock_mock.set_value(timestamp.as_nanos()); }
4409 Ok(())
4410 }
4411 });
4412 let replay_set_clock = if clock_sync_enabled {
4413 quote! {
4414 if let Some(sync) = &mut self.replay_clock_sync {
4415 sync.set_replay_time(clock_mock, timestamp).map_err(|error| CuError::new_with_cause("Replay clock time failed", error))?;
4416 } else { clock_mock.set_value(timestamp.as_nanos()); }
4417 }
4418 } else { quote! { clock_mock.set_value(timestamp.as_nanos()); } };
4419
4420 let recorded_replay_app_impl = if sim_mode {
4421 Some(quote! {
4422 impl<S: SectionStorage + 'static, L: UnifiedLogWrite<S> + 'static>
4423 CuRecordedReplayApplication<S, L> for #application_name
4424 {
4425 type RecordedDataSet = #mission_mod::CuStampedDataSet;
4426 #restore_clock_sync_method
4427
4428 #[allow(deprecated)] fn replay_recorded_copperlist(
4430 &mut self,
4431 clock_mock: &RobotClockMock,
4432 copperlist: &CopperList<Self::RecordedDataSet>,
4433 keyframe: Option<&KeyFrame>,
4434 ) -> CuResult<()> {
4435 cu29::continuity::validate_replay_continuity(
4436 self.copper_runtime_mut().copperlists_manager.next_cl_id(),
4437 copperlist.id,
4438 keyframe.map(|frame| frame.culistid),
4439 )?;
4440 let timestamp = if let Some(keyframe) = keyframe {
4441 if !self.copper_runtime_mut().captures_keyframe(copperlist.id) {
4442 return Err(CuError::from(format!(
4443 "CopperList {} is not configured to capture a keyframe in this runtime",
4444 copperlist.id
4445 )));
4446 }
4447 keyframe.timestamp
4448 } else {
4449 cu29::simulation::recorded_copperlist_timestamp(copperlist)
4450 .ok_or_else(|| CuError::from(format!(
4451 "Recorded copperlist {} has no process_time.start timestamps",
4452 copperlist.id
4453 )))?
4454 };
4455 self.copper_runtime_mut().copperlists_manager
4458 .prepare_recorded_replay(copperlist.id)?;
4459 self.copper_runtime_mut().keyframes_manager.finish_pending()?;
4460 if let Some(keyframe) = keyframe {
4461 self.copper_runtime_mut().set_forced_keyframe_timestamp(timestamp);
4462 self.copper_runtime_mut().lock_keyframe(keyframe);
4463 }
4464 #replay_set_clock
4465
4466 let mut sim_callback = |step: SimStep<'_>| -> SimOverride {
4467 #mission_mod::recorded_replay_step(step, copperlist)
4468 };
4469 <Self as CuSimApplication<S, L>>::run_one_iteration(self, &mut sim_callback)
4470 }
4471 }
4472 })
4473 } else {
4474 None
4475 };
4476
4477 let distributed_replay_app_impl = if sim_mode {
4478 Some(quote! {
4479 impl<S: SectionStorage + 'static, L: UnifiedLogWrite<S> + 'static>
4480 cu29::prelude::app::CuDistributedReplayApplication<S, L> for #application_name
4481 {
4482 fn build_distributed_replay(
4483 clock: cu29::clock::RobotClock,
4484 unified_logger: std::sync::Arc<std::sync::Mutex<L>>,
4485 instance_id: u32,
4486 config_override: Option<cu29::config::CuConfig>,
4487 ) -> CuResult<Self> {
4488 let mut noop =
4489 |_step: SimStep<'_>| cu29::simulation::SimOverride::ExecuteByRuntime;
4490 let builder = Self::builder()
4491 .with_logger::<S, L>(unified_logger)
4492 .with_clock(clock)
4493 .with_instance_id(instance_id);
4494 let builder = if let Some(config_override) = config_override {
4495 builder.with_config(config_override)
4496 } else {
4497 builder
4498 };
4499 builder.with_sim_callback(&mut noop).build_impl()
4500 }
4501 }
4502 })
4503 } else {
4504 None
4505 };
4506
4507 let (builder_build_thread_pools_stmt, builder_build_thread_pools_init) = if std {
4508 (
4509 quote! {
4510 let thread_pools = #mission_mod::thread_pools_instanciator(&config)?;
4511 },
4512 quote! { thread_pools, },
4513 )
4514 } else {
4515 (quote!(), quote!())
4516 };
4517
4518 let builder_prepare_config_call = if std {
4519 quote! { #application_name::prepare_config(self.instance_id, self.config_override)? }
4520 } else {
4521 quote! {{
4522 let _ = self.config_override;
4523 #application_name::prepare_config()?
4524 }}
4525 };
4526
4527 let builder_with_config_method = if std {
4528 Some(quote! {
4529 #[allow(dead_code)]
4530 pub fn with_config(mut self, config_override: CuConfig) -> Self {
4531 self.config_override = Some(config_override);
4532 self
4533 }
4534 })
4535 } else {
4536 None
4537 };
4538
4539 let builder_default_clock = if std && !clock_sync_enabled {
4540 quote! { Some(RobotClock::default()) }
4541 } else {
4542 quote! { None }
4543 };
4544
4545 let builder_clock_reference_field = clock_sync_enabled.then(|| quote! {
4546 clock_reference: Option<(Box<dyn ::cu29::clock_sync::ClockReference>, ::cu29::clock_sync::MaintenanceConfig)>,
4547 });
4548 let builder_clock_reference_none = clock_sync_enabled.then(|| quote! { clock_reference: None, });
4549 let builder_clock_reference_copy = clock_sync_enabled.then(|| quote! { clock_reference: self.clock_reference, });
4550 let builder_with_clock_reference = clock_sync_enabled.then(|| quote! {
4551 pub fn with_clock_reference<P: ::cu29::clock_sync::ClockReference>(mut self, reference: P, config: ::cu29::clock_sync::MaintenanceConfig) -> Self {
4553 self.clock_reference = Some((Box::new(reference), config)); self
4554 }
4555 });
4556 let builder_clock_setup = if clock_sync_enabled {
4557 let no_parent_clock = if std { quote! { RobotClock::new() } } else {
4558 quote! { return Err(CuError::from("Clock missing from builder and reference provider")) }
4559 };
4560 if sim_mode {
4561 quote! { let clock = self.clock.unwrap_or_else(|| RobotClock::mock().0); let clock_maintenance = None; }
4562 } else {
4563 quote! {
4564 if self.clock_reference.is_some() && config.runtime.as_ref().and_then(|r| r.clock_sync_config()).is_some() {
4565 return Err(CuError::from("Injected clock reference conflicts with runtime.clock.parent"));
4566 }
4567 let selected = match self.clock_reference {
4568 Some(parent) => Some(parent),
4569 None => #mission_mod::clock_reference_instanciator(&config, &mut resources)?,
4570 };
4571 if selected.is_some() && config.planner_kind() == ::cu29::config::PlannerKind::Pipeline {
4572 return Err(CuError::from("Clock synchronization requires the Serial planner"));
4573 }
4574 let clock = match self.clock {
4575 Some(clock) => clock,
4576 None => match &selected { Some((reference, _)) => reference.create_clock()?, None => { #no_parent_clock } },
4577 };
4578 let clock_maintenance = selected.map(|(reference, settings)| ::cu29::clock_sync::ClockMaintenance::new(reference, clock.clone(), settings)).transpose()?;
4579 }
4580 }
4581 } else {
4582 quote! { let clock = self.clock.ok_or(CuError::from("Clock missing from builder"))?; }
4583 };
4584
4585 let builder_logstream_field = logstream_enabled.then(|| {
4586 quote! {
4587 logstream: Option<(
4588 Box<dyn ::cu29::logstream::CuStreamTx>,
4589 ::cu29::logstream::LogStreamSenderConfig,
4590 )>,
4591 }
4592 });
4593 let builder_logstream_init = logstream_enabled.then(|| quote! { logstream: None, });
4594 let builder_logstream_copy =
4595 logstream_enabled.then(|| quote! { logstream: self.logstream, });
4596 let builder_with_logstream_method = logstream_enabled.then(|| {
4597 quote! {
4598 pub fn with_logstream<T>(
4604 mut self,
4605 transport: T,
4606 config: ::cu29::logstream::LogStreamSenderConfig,
4607 ) -> Self
4608 where
4609 T: ::cu29::logstream::CuStreamTx + 'static,
4610 {
4611 self.logstream = Some((Box::new(transport), config));
4612 self
4613 }
4614 }
4615 });
4616
4617 let (
4618 builder_struct,
4619 builder_impl,
4620 builder_ctor,
4621 builder_log_path_generics,
4622 builder_sim_callback_method,
4623 builder_build_sim_callback_arg,
4624 ) = if sim_mode {
4625 (
4626 quote! {
4627 #[allow(dead_code)]
4628 pub struct #builder_name<'a, F, S, L, R>
4629 where
4630 S: SectionStorage + 'static,
4631 L: UnifiedLogWrite<S> + 'static,
4632 R: FnOnce(&CuConfig) -> CuResult<ResourceManager>,
4633 F: FnMut(SimStep) -> SimOverride,
4634 {
4635 clock: Option<RobotClock>,
4636 unified_logger: Arc<Mutex<L>>,
4637 instance_id: u32,
4638 config_override: Option<CuConfig>,
4639 resources_factory: R,
4640 sim_callback: Option<&'a mut F>,
4641 #builder_logstream_field
4642 #builder_clock_reference_field
4643 _storage: core::marker::PhantomData<S>,
4644 }
4645 },
4646 quote! {
4647 impl<'a, F, S, L, R> #builder_name<'a, F, S, L, R>
4648 where
4649 S: SectionStorage + 'static,
4650 L: UnifiedLogWrite<S> + 'static,
4651 R: FnOnce(&CuConfig) -> CuResult<ResourceManager>,
4652 F: FnMut(SimStep) -> SimOverride,
4653 },
4654 quote! {
4655 #[allow(dead_code)]
4656 pub fn builder<'a, F>() -> #builder_name<'a, F, cu29::prelude::NoopSectionStorage, cu29::prelude::NoopLogger, fn(&CuConfig) -> CuResult<ResourceManager>>
4657 where
4658 F: FnMut(SimStep) -> SimOverride,
4659 {
4660 #builder_name {
4661 clock: #builder_default_clock,
4662 unified_logger: Arc::new(Mutex::new(cu29::prelude::NoopLogger::new())),
4663 instance_id: 0,
4664 config_override: None,
4665 resources_factory: #mission_mod::resources_instanciator as fn(&CuConfig) -> CuResult<ResourceManager>,
4666 sim_callback: None,
4667 #builder_logstream_init
4668 #builder_clock_reference_none
4669 _storage: core::marker::PhantomData,
4670 }
4671 }
4672 },
4673 quote! {'a, F, MmapSectionStorage, UnifiedLoggerWrite, R},
4674 Some(quote! {
4675 #[allow(dead_code)]
4676 pub fn with_sim_callback(mut self, sim_callback: &'a mut F) -> Self {
4677 self.sim_callback = Some(sim_callback);
4678 self
4679 }
4680 }),
4681 Some(quote! {
4682 self.sim_callback
4683 .ok_or(CuError::from("Sim callback missing from builder"))?,
4684 }),
4685 )
4686 } else {
4687 (
4688 quote! {
4689 #[allow(dead_code)]
4690 pub struct #builder_name<S, L, R>
4691 where
4692 S: SectionStorage + 'static,
4693 L: UnifiedLogWrite<S> + 'static,
4694 R: FnOnce(&CuConfig) -> CuResult<ResourceManager>,
4695 {
4696 clock: Option<RobotClock>,
4697 unified_logger: Arc<Mutex<L>>,
4698 instance_id: u32,
4699 config_override: Option<CuConfig>,
4700 resources_factory: R,
4701 #builder_logstream_field
4702 #builder_clock_reference_field
4703 _storage: core::marker::PhantomData<S>,
4704 }
4705 },
4706 quote! {
4707 impl<S, L, R> #builder_name<S, L, R>
4708 where
4709 S: SectionStorage + 'static,
4710 L: UnifiedLogWrite<S> + 'static,
4711 R: FnOnce(&CuConfig) -> CuResult<ResourceManager>,
4712 },
4713 quote! {
4714 #[allow(dead_code)]
4715 pub fn builder() -> #builder_name<cu29::prelude::NoopSectionStorage, cu29::prelude::NoopLogger, fn(&CuConfig) -> CuResult<ResourceManager>> {
4716 #builder_name {
4717 clock: #builder_default_clock,
4718 unified_logger: Arc::new(Mutex::new(cu29::prelude::NoopLogger::new())),
4719 instance_id: 0,
4720 config_override: None,
4721 resources_factory: #mission_mod::resources_instanciator as fn(&CuConfig) -> CuResult<ResourceManager>,
4722 #builder_logstream_init
4723 #builder_clock_reference_none
4724 _storage: core::marker::PhantomData,
4725 }
4726 }
4727 },
4728 quote! {MmapSectionStorage, UnifiedLoggerWrite, R},
4729 None,
4730 None,
4731 )
4732 };
4733
4734 let builder_with_logger_generics = if sim_mode {
4735 quote! {'a, F, S2, L2, R}
4736 } else {
4737 quote! {S2, L2, R}
4738 };
4739
4740 let builder_with_resources_generics = if sim_mode {
4741 quote! {'a, F, S, L, R2}
4742 } else {
4743 quote! {S, L, R2}
4744 };
4745
4746 let builder_sim_callback_field_copy = if sim_mode {
4747 Some(quote! {
4748 sim_callback: self.sim_callback,
4749 })
4750 } else {
4751 None
4752 };
4753
4754 let builder_with_log_path_method = if std {
4755 Some(quote! {
4756 #[allow(dead_code)]
4757 pub fn with_log_path(
4758 self,
4759 path: impl AsRef<std::path::Path>,
4760 slab_size: Option<usize>,
4761 ) -> CuResult<#builder_name<#builder_log_path_generics>> {
4762 let preallocated_size = slab_size.unwrap_or(1024 * 1024 * 10);
4763 let logger = cu29::prelude::UnifiedLoggerBuilder::new()
4764 .write(true)
4765 .create(true)
4766 .file_base_name(path.as_ref())
4767 .preallocated_size(preallocated_size)
4768 .build()
4769 .map_err(|e| CuError::new_with_cause("Failed to create unified logger", e))?;
4770 let logger = match logger {
4771 cu29::prelude::UnifiedLogger::Write(logger) => logger,
4772 cu29::prelude::UnifiedLogger::Read(_) => {
4773 return Err(CuError::from(
4774 "UnifiedLoggerBuilder did not create a write-capable logger",
4775 ));
4776 }
4777 };
4778 Ok(self.with_logger::<MmapSectionStorage, UnifiedLoggerWrite>(Arc::new(Mutex::new(
4779 logger,
4780 ))))
4781 }
4782 })
4783 } else {
4784 None
4785 };
4786
4787 let builder_with_unified_logger_method = if std {
4788 Some(quote! {
4789 #[allow(dead_code)]
4790 pub fn with_unified_logger(
4791 self,
4792 unified_logger: Arc<Mutex<UnifiedLoggerWrite>>,
4793 ) -> #builder_name<#builder_log_path_generics> {
4794 self.with_logger::<MmapSectionStorage, UnifiedLoggerWrite>(unified_logger)
4795 }
4796 })
4797 } else {
4798 None
4799 };
4800
4801 let std_application_impl = if sim_mode {
4803 Some(quote! {
4805 impl #application_name {
4806 #[deprecated(
4807 since = "1.2.0",
4808 note = "use the typed lifecycle handle returned by `build()` instead"
4809 )]
4810 #[allow(deprecated)] pub fn start_all_tasks(&mut self, sim_callback: &mut impl FnMut(SimStep) -> SimOverride) -> CuResult<()> {
4812 <Self as #app_trait<MmapSectionStorage, UnifiedLoggerWrite>>::start_all_tasks(self, sim_callback)
4813 }
4814 #[deprecated(
4815 since = "1.2.0",
4816 note = "use the typed lifecycle handle returned by `build()` instead"
4817 )]
4818 #[allow(deprecated)] pub fn run_one_iteration(&mut self, sim_callback: &mut impl FnMut(SimStep) -> SimOverride) -> CuResult<()> {
4820 <Self as #app_trait<MmapSectionStorage, UnifiedLoggerWrite>>::run_one_iteration(self, sim_callback)
4821 }
4822 #[deprecated(
4823 since = "1.2.0",
4824 note = "use the typed lifecycle handle returned by `build()` instead"
4825 )]
4826 #[allow(deprecated)] pub fn run(&mut self, sim_callback: &mut impl FnMut(SimStep) -> SimOverride) -> CuResult<()> {
4828 <Self as #app_trait<MmapSectionStorage, UnifiedLoggerWrite>>::run(self, sim_callback)
4829 }
4830 #[deprecated(
4831 since = "1.2.0",
4832 note = "use the typed lifecycle handle returned by `build()` instead"
4833 )]
4834 #[allow(deprecated)] pub fn stop_all_tasks(&mut self, sim_callback: &mut impl FnMut(SimStep) -> SimOverride) -> CuResult<()> {
4836 <Self as #app_trait<MmapSectionStorage, UnifiedLoggerWrite>>::stop_all_tasks(self, sim_callback)
4837 }
4838 pub fn replay_recorded_copperlist(
4839 &mut self,
4840 clock_mock: &RobotClockMock,
4841 copperlist: &CopperList<CuStampedDataSet>,
4842 keyframe: Option<&KeyFrame>,
4843 ) -> CuResult<()> {
4844 <Self as CuRecordedReplayApplication<MmapSectionStorage, UnifiedLoggerWrite>>::replay_recorded_copperlist(
4845 self,
4846 clock_mock,
4847 copperlist,
4848 keyframe,
4849 )
4850 }
4851 }
4852 })
4853 } else if std {
4854 Some(quote! {
4856 impl #application_name {
4857 #[deprecated(
4858 since = "1.2.0",
4859 note = "use the typed lifecycle handle returned by `build()` instead"
4860 )]
4861 #[allow(deprecated)] pub fn start_all_tasks(&mut self) -> CuResult<()> {
4863 <Self as #app_trait<MmapSectionStorage, UnifiedLoggerWrite>>::start_all_tasks(self)
4864 }
4865 #[deprecated(
4866 since = "1.2.0",
4867 note = "use the typed lifecycle handle returned by `build()` instead"
4868 )]
4869 #[allow(deprecated)] pub fn run_one_iteration(&mut self) -> CuResult<()> {
4871 <Self as #app_trait<MmapSectionStorage, UnifiedLoggerWrite>>::run_one_iteration(self)
4872 }
4873 #[deprecated(
4874 since = "1.2.0",
4875 note = "use the typed lifecycle handle returned by `build()` instead"
4876 )]
4877 #[allow(deprecated)] pub fn run(&mut self) -> CuResult<()> {
4879 <Self as #app_trait<MmapSectionStorage, UnifiedLoggerWrite>>::run(self)
4880 }
4881 #[deprecated(
4882 since = "1.2.0",
4883 note = "use the typed lifecycle handle returned by `build()` instead"
4884 )]
4885 #[allow(deprecated)] pub fn stop_all_tasks(&mut self) -> CuResult<()> {
4887 <Self as #app_trait<MmapSectionStorage, UnifiedLoggerWrite>>::stop_all_tasks(self)
4888 }
4889 }
4890 })
4891 } else {
4892 None };
4894
4895 let (builder_build_app_return, builder_build_app_wrap) = if sim_mode {
4896 (
4897 quote! { cu29::prelude::app::CuSimAppLifecycle<S, L, #application_name> },
4898 quote! { cu29::prelude::app::CuSimAppLifecycle },
4899 )
4900 } else {
4901 (
4902 quote! { cu29::prelude::app::CuAppLifecycle<S, L, #application_name> },
4903 quote! { cu29::prelude::app::CuAppLifecycle },
4904 )
4905 };
4906
4907 let application_builder = Some(quote! {
4908 #builder_struct
4909
4910 #builder_impl
4911 {
4912 #[allow(dead_code)]
4913 pub fn with_clock(mut self, clock: RobotClock) -> Self {
4914 self.clock = Some(clock);
4915 self
4916 }
4917
4918 #[allow(dead_code)]
4919 pub fn with_logger<S2, L2>(
4920 self,
4921 unified_logger: Arc<Mutex<L2>>,
4922 ) -> #builder_name<#builder_with_logger_generics>
4923 where
4924 S2: SectionStorage + 'static,
4925 L2: UnifiedLogWrite<S2> + 'static,
4926 {
4927 #builder_name {
4928 clock: self.clock,
4929 unified_logger,
4930 instance_id: self.instance_id,
4931 config_override: self.config_override,
4932 resources_factory: self.resources_factory,
4933 #builder_sim_callback_field_copy
4934 #builder_logstream_copy
4935 #builder_clock_reference_copy
4936 _storage: core::marker::PhantomData,
4937 }
4938 }
4939
4940 #builder_with_unified_logger_method
4941
4942 #[allow(dead_code)]
4943 pub fn with_instance_id(mut self, instance_id: u32) -> Self {
4944 self.instance_id = instance_id;
4945 self
4946 }
4947
4948 pub fn with_resources<R2>(self, resources_factory: R2) -> #builder_name<#builder_with_resources_generics>
4949 where
4950 R2: FnOnce(&CuConfig) -> CuResult<ResourceManager>,
4951 {
4952 #builder_name {
4953 clock: self.clock,
4954 unified_logger: self.unified_logger,
4955 instance_id: self.instance_id,
4956 config_override: self.config_override,
4957 resources_factory,
4958 #builder_sim_callback_field_copy
4959 #builder_logstream_copy
4960 #builder_clock_reference_copy
4961 _storage: core::marker::PhantomData,
4962 }
4963 }
4964
4965 #builder_with_config_method
4966 #builder_with_log_path_method
4967 #builder_sim_callback_method
4968 #builder_with_logstream_method
4969 #builder_with_clock_reference
4970
4971 #[allow(dead_code)]
4977 pub fn build(self) -> CuResult<#builder_build_app_return> {
4978 Ok(#builder_build_app_wrap::new(self.build_impl()?))
4979 }
4980
4981 fn build_impl(self) -> CuResult<#application_name> {
4982
4983 let (config, config_source) = #builder_prepare_config_call;
4984 let log_context = #application_name::prepare_log_metadata::<S, L>(&self.unified_logger, &config, self.instance_id)?;
4985 #[allow(unused_mut)]
4986 let mut resources = (self.resources_factory)(&config)?;
4987 #builder_clock_setup
4988 #builder_build_thread_pools_stmt
4989 let app_resources = AppResources {
4990 config,
4991 config_source,
4992 log_context: Some(log_context),
4993 resources,
4994 #builder_build_thread_pools_init
4995 #clock_maintenance_init
4996 };
4997 #application_name::build_with_resources(
4998 clock,
4999 self.unified_logger,
5000 app_resources,
5001 self.instance_id,
5002 #builder_logstream_arg
5003 #builder_build_sim_callback_arg
5004 )
5005 }
5006 }
5007 });
5008
5009 let app_builder_inherent_impl = quote! {
5010 impl #application_name {
5011 #builder_ctor
5012 }
5013 };
5014
5015 let sim_imports = if sim_mode {
5016 Some(quote! {
5017 use cu29::simulation::SimOverride;
5018 use cu29::simulation::CuTaskCallbackState;
5019 use cu29::simulation::CuSimSrcTask;
5020 use cu29::simulation::CuSimSrcTaskPack;
5021 use cu29::simulation::CuSimSinkTask;
5022 use cu29::simulation::CuSimBridge;
5023 use cu29::prelude::app::CuSimApplication;
5024 use cu29::prelude::app::CuRecordedReplayApplication;
5025 use cu29::cubridge::BridgeChannelSet;
5026 })
5027 } else {
5028 None
5029 };
5030
5031 let sim_tasks = if sim_mode {
5032 Some(quote! {
5033 pub type CuSimTasks = #task_types_tuple_sim;
5036 })
5037 } else {
5038 None
5039 };
5040
5041 let sim_inst_body = if task_sim_instances_init_code.is_empty() {
5042 quote! {
5043 let _ = (resources, thread_pools);
5044 Ok(())
5045 }
5046 } else {
5047 quote! { Ok(( #(#task_sim_instances_init_code),*, )) }
5048 };
5049
5050 let sim_tasks_instanciator = if sim_mode {
5051 Some(quote! {
5052 pub fn tasks_instanciator_sim<'c>(
5053 all_instances_configs: Vec<Option<&'c ComponentConfig>>,
5054 resources: &mut ResourceManager,
5055 thread_pools: &[Option<Arc<ThreadPool>>],
5056 ) -> CuResult<CuSimTasks> {
5057 #sim_inst_body
5058 }})
5059 } else {
5060 None
5061 };
5062
5063 let tasks_inst_body_std = if task_instances_init_code.is_empty() {
5064 quote! {
5065 let _ = (resources, thread_pools);
5066 Ok(())
5067 }
5068 } else {
5069 quote! { Ok(( #(#task_instances_init_code),*, )) }
5070 };
5071
5072 let tasks_inst_body_nostd = if task_instances_init_code.is_empty() {
5073 quote! {
5074 let _ = resources;
5075 Ok(())
5076 }
5077 } else {
5078 quote! { Ok(( #(#task_instances_init_code),*, )) }
5079 };
5080
5081 let tasks_instanciator = if std {
5082 quote! {
5083 pub fn tasks_instanciator<'c>(
5084 all_instances_configs: Vec<Option<&'c ComponentConfig>>,
5085 resources: &mut ResourceManager,
5086 thread_pools: &[Option<Arc<ThreadPool>>],
5087 ) -> CuResult<CuTasks> {
5088 #tasks_inst_body_std
5089 }
5090 }
5091 } else {
5092 quote! {
5094 pub fn tasks_instanciator<'c>(
5095 all_instances_configs: Vec<Option<&'c ComponentConfig>>,
5096 resources: &mut ResourceManager,
5097 ) -> CuResult<CuTasks> {
5098 #tasks_inst_body_nostd
5099 }
5100 }
5101 };
5102
5103 let thread_pools_instanciator = if std {
5108 quote! {
5109 pub fn thread_pools_instanciator(
5110 config: &CuConfig,
5111 ) -> CuResult<Vec<Option<Arc<ThreadPool>>>> {
5112 let Some(runtime) = config.runtime.as_ref() else {
5113 return Ok(Vec::new());
5114 };
5115 let mut pools: Vec<Option<Arc<ThreadPool>>> =
5116 Vec::with_capacity(runtime.thread_pools.len());
5117 for pool_spec in &runtime.thread_pools {
5118 if pool_spec.id == cu29::config::RT_POOL {
5119 pools.push(None);
5120 continue;
5121 }
5122 let pool = cu29::thread_pool::build_pool(pool_spec)?;
5123 pools.push(Some(Arc::new(pool)));
5124 }
5125 Ok(pools)
5126 }
5127 }
5128 } else {
5129 quote! {}
5130 };
5131
5132 let imports = if std {
5133 quote! {
5134 use cu29::rayon::ThreadPool;
5135 use cu29::cuasynctask::CuAsyncSrcTask;
5136 use cu29::cuasynctask::CuAsyncTask;
5137 use cu29::resource::{ResourceBindings, ResourceManager};
5138 use cu29::prelude::SectionStorage;
5139 use cu29::prelude::UnifiedLoggerWrite;
5140 use cu29::prelude::memmap::MmapSectionStorage;
5141 use cu29::__private::sync::{Arc, Mutex};
5142 use std::fmt::{Debug, Formatter};
5143 use std::fmt::Result as FmtResult;
5144 use std::mem::size_of;
5145 use std::boxed::Box;
5146 use std::sync::atomic::{AtomicBool, Ordering};
5147 }
5148 } else {
5149 quote! {
5150 use alloc::boxed::Box;
5151 use alloc::string::String;
5152 use alloc::string::ToString;
5153 use cu29::__private::sync::{Arc, Mutex};
5154 use core::sync::atomic::{AtomicBool, Ordering};
5155 use core::fmt::{Debug, Formatter};
5156 use core::fmt::Result as FmtResult;
5157 use core::mem::size_of;
5158 use cu29::prelude::SectionStorage;
5159 use cu29::resource::{ResourceBindings, ResourceManager};
5160 }
5161 };
5162
5163 let task_mapping_defs = task_resource_mappings.defs.clone();
5164 let bridge_mapping_defs = bridge_resource_mappings.defs.clone();
5165
5166 let mission_mod_tokens = quote! {
5168 mod #mission_mod {
5169 use super::*; #mission_constant_contents
5172
5173 use cu29::bincode::Encode;
5174 use cu29::bincode::enc::Encoder;
5175 use cu29::bincode::error::EncodeError;
5176 use cu29::bincode::Decode;
5177 use cu29::bincode::de::Decoder;
5178 use cu29::bincode::de::DecoderImpl;
5179 use cu29::bincode::error::DecodeError;
5180 use cu29::clock::RobotClock;
5181 use cu29::clock::RobotClockMock;
5182 use cu29::config::CuConfig;
5183 use cu29::config::ComponentConfig;
5184 use cu29::curuntime::CuRuntime;
5185 use cu29::curuntime::CuRuntimeBuilder;
5186 use cu29::curuntime::CuRuntimeParts;
5187 use cu29::curuntime::KeyFrame;
5188 use cu29::curuntime::RuntimeLifecycleConfigSource;
5189 use cu29::curuntime::RuntimeLifecycleEvent;
5190 use cu29::curuntime::RuntimeLifecycleRecord;
5191 use cu29::CuResult;
5192 use cu29::CuError;
5193 use cu29::cutask::CuSrcTask;
5194 use cu29::cutask::CuSinkTask;
5195 use cu29::cutask::CuTask;
5196 use cu29::cutask::CuStatelessTask;
5197 #[allow(unused_imports)]
5200 use cu29::cutask_anytime::CuAnytimeTask;
5201 use cu29::cutask::CuMsg;
5202 use cu29::cutask::CuMsgMetadata;
5203 use cu29::copperlist::CopperList;
5204 use cu29::monitoring::CuMonitor; use cu29::monitoring::CuComponentState;
5206 use cu29::monitoring::Decision;
5207 use cu29::prelude::app::CuApplication;
5208 use cu29::prelude::debug;
5209 use cu29::prelude::stream_write;
5210 use cu29::prelude::UnifiedLogType;
5211 use cu29::prelude::UnifiedLogWrite;
5212 use cu29::prelude::WriteStream;
5213
5214 #imports
5215
5216 #sim_imports
5217
5218 #[allow(unused_imports)]
5220 use cu29::monitoring::NoMonitor;
5221
5222 pub type CuTasks = #task_types_tuple;
5226 pub type CuBridges = #bridges_type_tokens;
5227 #sim_bridge_channel_defs
5228 #resources_module
5229 #resources_instanciator_fn
5230 #clock_reference_factory
5231 #task_mapping_defs
5232 #bridge_mapping_defs
5233 #(#autogenerated_output_warnings)*
5234
5235 #(#anytime_policy_defs)*
5238
5239 #sim_tasks
5240 #sim_support
5241 #recorded_replay_support
5242 #sim_tasks_instanciator
5243
5244 pub const TASK_IDS: &'static [&'static str] = &[#( #task_ids ),*];
5245 pub const MONITORED_COMPONENTS: &'static [cu29::monitoring::MonitorComponentMetadata] =
5246 &[#( #monitored_component_entries ),*];
5247 pub const CULIST_COMPONENT_MAPPING: &'static [cu29::monitoring::ComponentId] =
5248 &[#( cu29::monitoring::ComponentId::new(#culist_component_mapping) ),*];
5249 pub const MONITOR_LAYOUT: cu29::monitoring::CopperListLayout =
5250 cu29::monitoring::CopperListLayout::new(
5251 MONITORED_COMPONENTS,
5252 CULIST_COMPONENT_MAPPING,
5253 );
5254 #parallel_rt_metadata_defs
5255
5256 #[inline]
5257 pub fn monitor_component_label(
5258 component_id: cu29::monitoring::ComponentId,
5259 ) -> &'static str {
5260 MONITORED_COMPONENTS[component_id.index()].id()
5261 }
5262
5263 #culist_support
5264 #parallel_rt_support_tokens
5265
5266 #tasks_instanciator
5267 #thread_pools_instanciator
5268 #bridges_instanciator
5269
5270 pub fn monitor_instanciator(
5271 config: &CuConfig,
5272 metadata: ::cu29::monitoring::CuMonitoringMetadata,
5273 runtime: ::cu29::monitoring::CuMonitoringRuntime,
5274 ) -> #monitor_type {
5275 #monitor_instanciator_body
5276 }
5277
5278 #app_resources_struct
5280 pub #application_struct
5281
5282 #app_inherent_impl
5283 #app_drop_impl
5284 #app_builder_inherent_impl
5285 #app_metadata_impl
5286 #app_reflect_impl
5287 #app_runtime_copperlist_impl
5288 #application_impl
5289 #live_replay_impl
5290 #recorded_replay_app_impl
5291 #distributed_replay_app_impl
5292
5293 #std_application_impl
5294
5295 #application_builder
5296 }
5297
5298 };
5299 all_missions_tokens.push(mission_mod_tokens);
5300 }
5301
5302 let default_application_tokens = if all_missions
5303 .iter()
5304 .any(|(mission_name, _)| mission_name == "default")
5305 {
5306 let default_builder = quote! {
5307 #[allow(unused_imports)]
5308 use default::#builder_name;
5309 };
5310 quote! {
5311 #default_builder
5312
5313 #[allow(unused_imports)]
5314 use default::AppResources;
5315
5316 #[allow(unused_imports)]
5317 use default::resources as app_resources;
5318
5319 #[allow(unused_imports)]
5320 use default::#application_name;
5321 }
5322 } else {
5323 quote!() };
5325
5326 let mission_module_names = all_missions
5327 .iter()
5328 .map(|(mission, _)| {
5329 parse_str::<Ident>(mission)
5330 .expect("Could not make an identifier of the mission name")
5331 .unraw()
5332 .to_string()
5333 })
5334 .collect::<BTreeSet<_>>();
5335 let root_constant_modules = constant_modules.root_modules_except(&mission_module_names);
5336
5337 let result: proc_macro2::TokenStream = quote! {
5338 #root_constant_modules
5339 #(#all_missions_tokens)*
5340 #default_application_tokens
5341 };
5342
5343 result.into()
5344}