1use crate::config::ComponentConfig;
11use crate::context::CuContext;
12use crate::cutask::{CuMsg, CuMsgPack, CuMsgPayload, CuTask, Freezable};
13use crate::reflect::{GetTypeRegistration, Reflect, TypePath, TypeRegistry};
14use bincode::de::Decoder;
15use bincode::enc::Encoder;
16use bincode::error::{DecodeError, EncodeError};
17use compact_str::format_compact;
18use core::fmt::{Debug, Formatter, Result as FmtResult};
19use core::marker::PhantomData;
20use cu29_clock::{CuDuration, CuTime, Tov};
21use cu29_traits::{CuCompactString, CuResult};
22use cu29_units::si::f32::Ratio;
23use cu29_units::si::ratio::ratio;
24
25pub type Quality = Ratio;
30
31#[derive(Debug, Clone, Copy, PartialEq)]
43pub enum AnytimeStatus<Q> {
44 Improved(Q),
46 Converged(Q),
49 Aborted,
55}
56
57pub trait CuAnytimeTask: Freezable + Reflect {
64 type Input<'m>: CuMsgPack;
65 type Output<'m>: CuMsgPayload;
66 type Resources<'r>;
68 type Quality: AnytimeQuality;
73
74 fn register_debug_state_types(registry: &mut TypeRegistry)
80 where
81 Self: GetTypeRegistration + Sized,
82 {
83 registry.register::<Self>();
84 }
85
86 fn debug_state_type_path() -> &'static str
88 where
89 Self: TypePath + Sized,
90 {
91 Self::type_path()
92 }
93
94 fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
99 where
100 Self: Sized,
101 {
102 f(self)
103 }
104
105 fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
108 where
109 Self: Sized;
110
111 fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
114 Ok(())
115 }
116
117 fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
121 Ok(())
122 }
123
124 fn base<'i, 'o>(
136 &mut self,
137 ctx: &CuContext,
138 input: &Self::Input<'i>,
139 output: &mut Self::Output<'o>,
140 ) -> CuResult<AnytimeStatus<Self::Quality>>;
141
142 fn refine<'o>(
156 &mut self,
157 ctx: &CuContext,
158 output: &mut Self::Output<'o>,
159 ) -> CuResult<AnytimeStatus<Self::Quality>>;
160
161 fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
166 Ok(())
167 }
168
169 fn stop(&mut self, _ctx: &CuContext) -> CuResult<()> {
172 Ok(())
173 }
174}
175
176#[inline(always)]
178pub fn quality_from_f32(value: f32) -> Quality {
179 Quality::new::<ratio>(value)
180}
181
182#[inline(always)]
184pub fn quality_to_f32(quality: Quality) -> f32 {
185 quality.get::<ratio>()
186}
187
188pub trait AnytimeQuality: Copy + PartialOrd {
193 #[inline(always)]
195 fn ratio(self) -> Option<f32> {
196 None
197 }
198}
199
200impl AnytimeQuality for Quality {
201 #[inline(always)]
202 fn ratio(self) -> Option<f32> {
203 Some(quality_to_f32(self))
204 }
205}
206
207impl AnytimeQuality for () {}
208
209#[doc(hidden)]
215#[diagnostic::on_unimplemented(
216 message = "the anytime policy `{Self}` is pinned to the shared quality scale, but this task's `Quality` is `{Q}`",
217 note = "quality knobs (quality_target/quality_floor/max_stall) require `type Quality = cu29::cutask_anytime::Quality` on the task; remove the knob or score the task's results"
218)]
219pub trait AnytimePolicy<Q> {
220 const TIME_BUDGET: Option<CuDuration>;
222 const MAX_AGE: Option<CuDuration>;
225 const MAX_STALL: Option<u32>;
227 const MAX_REFINES: Option<u32>;
231
232 #[inline(always)]
234 fn target_met(_q: Q) -> bool {
235 false
236 }
237 #[inline(always)]
241 fn below_floor(_q: Q) -> bool {
242 false
243 }
244}
245
246#[doc(hidden)]
248#[derive(Debug, Clone, Copy, PartialEq, Eq)]
249pub enum AnytimeStopCause {
250 Converged,
252 TargetMet,
254 BudgetExhausted,
256 AgeExceeded,
259 SkippedStale,
261 MaxRefines,
263 Stalled,
265 Aborted,
267}
268
269impl AnytimeStopCause {
270 pub fn label(self) -> &'static str {
272 match self {
273 AnytimeStopCause::Converged => "conv",
274 AnytimeStopCause::TargetMet => "tgt",
275 AnytimeStopCause::BudgetExhausted => "bdgt",
276 AnytimeStopCause::AgeExceeded => "age",
277 AnytimeStopCause::SkippedStale => "stale",
278 AnytimeStopCause::MaxRefines => "max",
279 AnytimeStopCause::Stalled => "stall",
280 AnytimeStopCause::Aborted => "abort",
281 }
282 }
283}
284
285#[doc(hidden)]
287#[derive(Debug, Clone, Copy, PartialEq, Eq)]
288pub struct AnytimeOutcome {
289 pub iterations: u32,
291 pub elapsed: CuDuration,
293 pub stop: AnytimeStopCause,
295 pub published: bool,
298}
299
300#[doc(hidden)]
307pub struct AnytimeJob<Q, P> {
308 t0: CuTime,
310 anchor: CuTime,
312 best: Q,
314 stall: u32,
316 _policy: PhantomData<P>,
317}
318
319impl<Q: Copy, P> Clone for AnytimeJob<Q, P> {
322 fn clone(&self) -> Self {
323 Self {
324 t0: self.t0,
325 anchor: self.anchor,
326 best: self.best,
327 stall: self.stall,
328 _policy: PhantomData,
329 }
330 }
331}
332impl<Q: Debug, P> Debug for AnytimeJob<Q, P> {
333 fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
334 f.debug_struct("AnytimeJob")
335 .field("t0", &self.t0)
336 .field("anchor", &self.anchor)
337 .field("best", &self.best)
338 .field("stall", &self.stall)
339 .finish()
340 }
341}
342
343impl<Q: AnytimeQuality, P: AnytimePolicy<Q>> AnytimeJob<Q, P> {
344 pub fn new(t0: CuTime, anchor: CuTime, quality: Q) -> Self {
346 Self {
347 t0,
348 anchor,
349 best: quality,
350 stall: 0,
351 _policy: PhantomData,
352 }
353 }
354
355 pub fn record(&mut self, quality: Q) {
360 if quality > self.best || self.best.partial_cmp(&self.best).is_none() {
361 self.best = quality;
362 self.stall = 0;
363 } else if P::MAX_STALL.is_some() {
364 self.stall += 1;
365 }
366 }
367
368 pub fn check(&self, now: CuTime) -> Option<AnytimeStopCause> {
371 if P::target_met(self.best) {
372 return Some(AnytimeStopCause::TargetMet);
373 }
374 if let Some(budget) = P::TIME_BUDGET
375 && now >= self.t0 + budget
376 {
377 return Some(AnytimeStopCause::BudgetExhausted);
378 }
379 if let Some(age) = P::MAX_AGE
380 && now >= self.anchor + age
381 {
382 return Some(AnytimeStopCause::AgeExceeded);
383 }
384 if let Some(max_stall) = P::MAX_STALL
385 && self.stall >= max_stall
386 {
387 return Some(AnytimeStopCause::Stalled);
388 }
389 None
390 }
391
392 pub fn finish<O: CuMsgPayload>(
396 self,
397 now: CuTime,
398 cause: AnytimeStopCause,
399 iterations: u32,
400 output: &mut CuMsg<O>,
401 ) -> AnytimeOutcome {
402 let published = if P::below_floor(self.best) {
403 output.clear_payload();
404 false
405 } else {
406 output.payload().is_some()
407 };
408 stamp(output, iterations, self.best.ratio(), cause, published);
409 AnytimeOutcome {
410 iterations,
411 elapsed: now - self.t0,
412 stop: cause,
413 published,
414 }
415 }
416}
417
418fn stamp<O: CuMsgPayload>(
426 output: &mut CuMsg<O>,
427 iterations: u32,
428 quality: Option<f32>,
429 cause: AnytimeStopCause,
430 published: bool,
431) {
432 let not_published = if published { "" } else { "!" };
433 output.metadata.status_txt = CuCompactString(match quality {
434 Some(q) => format_compact!(
435 "any:{}it q={:.2} {}{}",
436 iterations,
437 q,
438 cause.label(),
439 not_published
440 ),
441 None => format_compact!("any:{}it {}{}", iterations, cause.label(), not_published),
442 });
443}
444
445#[doc(hidden)]
450#[inline(always)]
451pub fn anchor_from_tov(tov: Tov, now: CuTime) -> CuTime {
452 match tov {
453 Tov::Time(time) => time,
454 Tov::Range(range) => range.start,
455 Tov::None => now,
456 }
457}
458
459#[doc(hidden)]
462pub fn skip_stale<O: CuMsgPayload>(output: &mut CuMsg<O>) -> AnytimeOutcome {
463 output.clear_payload();
464 stamp(output, 0, None, AnytimeStopCause::SkippedStale, false);
465 AnytimeOutcome {
466 iterations: 0,
467 elapsed: CuDuration::default(),
468 stop: AnytimeStopCause::SkippedStale,
469 published: false,
470 }
471}
472
473#[doc(hidden)]
478pub fn abort_at_base<O: CuMsgPayload>(
479 t0: CuTime,
480 now: CuTime,
481 output: &mut CuMsg<O>,
482) -> AnytimeOutcome {
483 let published = output.payload().is_some();
484 stamp(output, 0, None, AnytimeStopCause::Aborted, published);
485 AnytimeOutcome {
486 iterations: 0,
487 elapsed: now - t0,
488 stop: AnytimeStopCause::Aborted,
489 published,
490 }
491}
492
493#[doc(hidden)]
506#[derive(Reflect)]
507#[reflect(no_field_bounds, from_reflect = false, type_path = false)]
508pub struct CuAnytimeRunner<T, P>
509where
510 T: Send + Sync + 'static,
511 P: Send + Sync + 'static,
512{
513 #[reflect(ignore)]
514 task: T,
515 #[reflect(ignore)]
516 _policy: PhantomData<P>,
517}
518
519impl<T, P> TypePath for CuAnytimeRunner<T, P>
520where
521 T: Send + Sync + 'static,
522 P: Send + Sync + 'static,
523{
524 fn type_path() -> &'static str {
525 "cu29_runtime::cutask_anytime::CuAnytimeRunner"
526 }
527
528 fn short_type_path() -> &'static str {
529 "CuAnytimeRunner"
530 }
531
532 fn type_ident() -> Option<&'static str> {
533 Some("CuAnytimeRunner")
534 }
535
536 fn crate_name() -> Option<&'static str> {
537 Some("cu29_runtime")
538 }
539
540 fn module_path() -> Option<&'static str> {
541 Some("cutask_anytime")
542 }
543}
544
545impl<T, P> Freezable for CuAnytimeRunner<T, P>
546where
547 T: Freezable + Send + Sync + 'static,
548 P: Send + Sync + 'static,
549{
550 fn freeze<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
551 self.task.freeze(encoder)
552 }
553
554 fn thaw<D: Decoder>(&mut self, decoder: &mut D) -> Result<(), DecodeError> {
555 self.task.thaw(decoder)
556 }
557}
558
559impl<T, I, O, P> CuTask for CuAnytimeRunner<T, P>
560where
561 T: for<'i, 'o> CuAnytimeTask<Input<'i> = CuMsg<I>, Output<'o> = CuMsg<O>>
562 + GetTypeRegistration
563 + TypePath
564 + Send
565 + Sync
566 + 'static,
567 I: CuMsgPayload,
568 O: CuMsgPayload,
569 P: AnytimePolicy<T::Quality> + Send + Sync + 'static,
570{
571 type Resources<'r> = T::Resources<'r>;
572 type Input<'m> = T::Input<'m>;
573 type Output<'m> = T::Output<'m>;
574
575 fn register_debug_state_types(registry: &mut TypeRegistry)
578 where
579 Self: GetTypeRegistration + Sized,
580 {
581 T::register_debug_state_types(registry);
582 }
583
584 fn debug_state_type_path() -> &'static str
585 where
586 Self: TypePath + Sized,
587 {
588 T::debug_state_type_path()
589 }
590
591 fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
592 where
593 Self: Sized,
594 {
595 self.task.with_debug_state(f)
596 }
597
598 fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
599 where
600 Self: Sized,
601 {
602 Ok(Self {
603 task: T::new(config, resources)?,
604 _policy: PhantomData,
605 })
606 }
607
608 fn start(&mut self, ctx: &CuContext) -> CuResult<()> {
609 self.task.start(ctx)
610 }
611
612 fn process<'i, 'o>(
613 &mut self,
614 ctx: &CuContext,
615 input: &Self::Input<'i>,
616 output: &mut Self::Output<'o>,
617 ) -> CuResult<()> {
618 self.task.preprocess(ctx)?;
622 let job = run_job::<T, I, O, P>(&mut self.task, ctx, input, output);
623 let post = self.task.postprocess(ctx);
624 job.and(post)
625 }
626
627 fn stop(&mut self, ctx: &CuContext) -> CuResult<()> {
628 self.task.stop(ctx)
629 }
630}
631
632fn run_job<T, I, O, P>(
636 task: &mut T,
637 ctx: &CuContext,
638 input: &CuMsg<I>,
639 output: &mut CuMsg<O>,
640) -> CuResult<()>
641where
642 T: for<'i, 'o> CuAnytimeTask<Input<'i> = CuMsg<I>, Output<'o> = CuMsg<O>>,
643 I: CuMsgPayload,
644 O: CuMsgPayload,
645 P: AnytimePolicy<T::Quality>,
646{
647 let start = if P::TIME_BUDGET.is_some() || P::MAX_AGE.is_some() {
652 ctx.now()
653 } else {
654 CuTime::default()
655 };
656 let anchor = if P::MAX_AGE.is_some() {
657 anchor_from_tov(input.tov, start)
658 } else {
659 start
660 };
661 if let Some(max_age) = P::MAX_AGE
662 && start >= anchor + max_age
663 {
664 skip_stale(output);
665 return Ok(());
666 }
667
668 let mut job = match task.base(ctx, input, output)? {
671 AnytimeStatus::Improved(quality) => AnytimeJob::<_, P>::new(start, anchor, quality),
672 AnytimeStatus::Converged(quality) => {
673 AnytimeJob::<_, P>::new(start, anchor, quality).finish(
674 start,
675 AnytimeStopCause::Converged,
676 0,
677 output,
678 );
679 return Ok(());
680 }
681 AnytimeStatus::Aborted => {
682 abort_at_base(start, start, output);
683 return Ok(());
684 }
685 };
686
687 let mut ran = 0u32;
688 loop {
689 let now = if P::TIME_BUDGET.is_some() || P::MAX_AGE.is_some() {
693 ctx.now()
694 } else {
695 CuTime::default()
696 };
697 if let Some(cause) = job.check(now) {
698 job.finish(now, cause, ran, output);
699 return Ok(());
700 }
701 let status = task.refine(ctx, output)?;
704 ran += 1;
705 match status {
706 AnytimeStatus::Improved(quality) => {
707 job.record(quality);
708 if let Some(max_refines) = P::MAX_REFINES
709 && ran >= max_refines
710 {
711 job.finish(now, AnytimeStopCause::MaxRefines, ran, output);
712 return Ok(());
713 }
714 }
715 AnytimeStatus::Converged(quality) => {
716 job.record(quality);
717 job.finish(now, AnytimeStopCause::Converged, ran, output);
718 return Ok(());
719 }
720 AnytimeStatus::Aborted => {
721 job.finish(now, AnytimeStopCause::Aborted, ran, output);
722 return Ok(());
723 }
724 }
725 }
726}
727
728#[cfg(test)]
729mod tests {
730 use super::*;
731 use crate::cutask::CuMsg;
732 use crate::input_msg;
733 use crate::output_msg;
734 use alloc::sync::Arc;
735 use core::sync::atomic::{AtomicU32, Ordering};
736 use cu29_clock::RobotClockMock;
737
738 fn q(v: f32) -> Quality {
739 quality_from_f32(v)
740 }
741
742 #[derive(Reflect)]
745 struct IncrementalSum {
746 target: u32,
747 acc: u32,
748 }
749
750 impl Freezable for IncrementalSum {}
751
752 impl CuAnytimeTask for IncrementalSum {
753 type Input<'m> = input_msg!(u32);
754 type Output<'m> = output_msg!(u32);
755 type Resources<'r> = ();
756 type Quality = Quality;
757
758 fn new(
759 _config: Option<&ComponentConfig>,
760 _resources: Self::Resources<'_>,
761 ) -> CuResult<Self> {
762 Ok(Self { target: 0, acc: 0 })
763 }
764
765 fn base<'i, 'o>(
766 &mut self,
767 _ctx: &CuContext,
768 input: &Self::Input<'i>,
769 output: &mut Self::Output<'o>,
770 ) -> CuResult<AnytimeStatus<Quality>> {
771 self.target = *input.payload().ok_or("no input")?;
772 self.acc = 0;
773 output.set_payload(self.acc);
774 Ok(AnytimeStatus::Improved(q(0.0)))
775 }
776
777 fn refine<'o>(
778 &mut self,
779 _ctx: &CuContext,
780 output: &mut Self::Output<'o>,
781 ) -> CuResult<AnytimeStatus<Quality>> {
782 if self.acc == self.target {
783 return Ok(AnytimeStatus::Converged(q(1.0)));
784 }
785 self.acc += 1;
786 output.set_payload(self.acc);
787 Ok(AnytimeStatus::Improved(q(
788 self.acc as f32 / self.target as f32
789 )))
790 }
791 }
792
793 #[test]
794 fn base_then_refine_until_converged() {
795 let ctx = CuContext::new_with_clock();
796 let mut task = IncrementalSum::new(None, ()).unwrap();
797 let input = CuMsg::new(Some(3u32));
798 let mut output = CuMsg::new(None);
799
800 task.start(&ctx).unwrap();
801 task.preprocess(&ctx).unwrap();
802 let status = task.base(&ctx, &input, &mut output).unwrap();
803 assert!(matches!(status, AnytimeStatus::Improved(_)));
804 assert_eq!(output.payload(), Some(&0));
805
806 let mut best_quality = q(0.0);
807 for _ in 0..8 {
808 match task.refine(&ctx, &mut output).unwrap() {
809 AnytimeStatus::Improved(quality) => best_quality = quality,
810 AnytimeStatus::Converged(quality) => {
811 best_quality = quality;
812 break;
813 }
814 status => panic!("unexpected status: {status:?}"),
815 }
816 }
817 assert_eq!(output.payload(), Some(&3));
818 assert_eq!(quality_to_f32(best_quality), 1.0);
819 task.postprocess(&ctx).unwrap();
820 task.stop(&ctx).unwrap();
821 }
822
823 struct FullPolicy;
826 impl AnytimePolicy<Quality> for FullPolicy {
827 const TIME_BUDGET: Option<CuDuration> = Some(CuDuration(1_000_000));
828 const MAX_AGE: Option<CuDuration> = Some(CuDuration(2_000_000));
829 const MAX_STALL: Option<u32> = Some(2);
830 const MAX_REFINES: Option<u32> = Some(8);
831
832 fn target_met(q: Quality) -> bool {
833 q >= quality_from_f32(0.9)
834 }
835 fn below_floor(q: Quality) -> bool {
836 q.partial_cmp(&quality_from_f32(0.3))
837 .is_none_or(core::cmp::Ordering::is_lt)
838 }
839 }
840
841 struct NoKnobPolicy;
843 impl<Q: Copy + PartialOrd> AnytimePolicy<Q> for NoKnobPolicy {
844 const TIME_BUDGET: Option<CuDuration> = None;
845 const MAX_AGE: Option<CuDuration> = None;
846 const MAX_STALL: Option<u32> = None;
847 const MAX_REFINES: Option<u32> = None;
848 }
849
850 struct BarePolicy;
852 impl AnytimePolicy<()> for BarePolicy {
853 const TIME_BUDGET: Option<CuDuration> = None;
854 const MAX_AGE: Option<CuDuration> = None;
855 const MAX_STALL: Option<u32> = None;
856 const MAX_REFINES: Option<u32> = None;
857 }
858
859 #[test]
860 fn check_attribution_order_is_target_budget_age_stall() {
861 let t0 = CuTime::from_millis(10);
862 let job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.95));
864 assert_eq!(
865 job.check(t0 + CuDuration::from_millis(5)),
866 Some(AnytimeStopCause::TargetMet)
867 );
868 let job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.5));
870 assert_eq!(
871 job.check(t0 + CuDuration::from_millis(5)),
872 Some(AnytimeStopCause::BudgetExhausted)
873 );
874 let anchor = t0 - CuDuration::from_millis(2);
876 let job = AnytimeJob::<Quality, FullPolicy>::new(t0, anchor, q(0.5));
877 assert_eq!(
878 job.check(t0 + CuDuration::from_nanos(1)),
879 Some(AnytimeStopCause::AgeExceeded)
880 );
881 let job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.5));
883 assert_eq!(job.check(t0), None);
884 }
885
886 #[test]
887 fn stall_counts_quanta_without_improvement() {
888 let t0 = CuTime::from_millis(1);
889 let mut job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.5));
890 job.record(q(0.5)); assert_eq!(job.check(t0), None);
892 job.record(q(0.6)); assert_eq!(job.check(t0), None);
894 job.record(q(0.6));
895 job.record(q(0.6)); assert_eq!(job.check(t0), Some(AnytimeStopCause::Stalled));
897 }
898
899 #[test]
900 fn finish_gates_on_floor_and_stamps_status() {
901 let t0 = CuTime::from_millis(1);
902 let now = t0 + CuDuration::from_micros(250);
903
904 let mut output: CuMsg<u32> = CuMsg::new(Some(42));
906 let job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.5));
907 let outcome = job.finish(now, AnytimeStopCause::BudgetExhausted, 3, &mut output);
908 assert!(outcome.published);
909 assert_eq!(outcome.iterations, 3);
910 assert_eq!(outcome.elapsed, CuDuration::from_micros(250));
911 assert_eq!(output.payload(), Some(&42));
912 assert_eq!(output.metadata.status_txt.0.as_str(), "any:3it q=0.50 bdgt");
913
914 let mut output: CuMsg<u32> = CuMsg::new(Some(42));
916 let job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.1));
917 let outcome = job.finish(now, AnytimeStopCause::MaxRefines, 2, &mut output);
918 assert!(!outcome.published);
919 assert_eq!(output.payload(), None);
920 assert_eq!(output.metadata.status_txt.0.as_str(), "any:2it q=0.10 max!");
921 }
922
923 #[test]
927 fn stamp_stays_inline() {
928 const CAUSES: [AnytimeStopCause; 8] = [
929 AnytimeStopCause::Converged,
930 AnytimeStopCause::TargetMet,
931 AnytimeStopCause::BudgetExhausted,
932 AnytimeStopCause::AgeExceeded,
933 AnytimeStopCause::SkippedStale,
934 AnytimeStopCause::MaxRefines,
935 AnytimeStopCause::Stalled,
936 AnytimeStopCause::Aborted,
937 ];
938
939 for cause in CAUSES {
940 for published in [true, false] {
941 for quality in [None, Some(0.0), Some(1.0)] {
943 let mut output: CuMsg<u32> = CuMsg::new(Some(1));
944 stamp(&mut output, 9999, quality, cause, published);
945 let stamped = &output.metadata.status_txt.0;
946 assert!(
947 !stamped.is_heap_allocated(),
948 "stamp allocates on the real-time path: {stamped:?} ({} bytes)",
949 stamped.len()
950 );
951 }
952 }
953 }
954 }
955
956 #[test]
957 fn nan_quality_fails_closed() {
958 let t0 = CuTime::from_millis(1);
959
960 let mut output: CuMsg<u32> = CuMsg::new(Some(1));
962 let job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(f32::NAN));
963 let outcome = job.finish(t0, AnytimeStopCause::MaxRefines, 1, &mut output);
964 assert!(!outcome.published);
965 assert_eq!(output.payload(), None);
966
967 let mut job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(f32::NAN));
969 job.record(q(0.4));
970 let mut output: CuMsg<u32> = CuMsg::new(Some(1));
971 let outcome = job.finish(t0, AnytimeStopCause::MaxRefines, 1, &mut output);
972 assert!(outcome.published);
973
974 let mut job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.5));
976 job.record(q(f32::NAN));
977 let mut output: CuMsg<u32> = CuMsg::new(Some(1));
978 let outcome = job.finish(t0, AnytimeStopCause::MaxRefines, 1, &mut output);
979 assert!(outcome.published);
980 assert_eq!(output.metadata.status_txt.0.as_str(), "any:1it q=0.50 max");
981 }
982
983 #[test]
984 fn quality_reaches_stamp_without_quality_knobs() {
985 let t0 = CuTime::from_millis(1);
988 let mut output: CuMsg<u32> = CuMsg::new(Some(7));
989 let job = AnytimeJob::<Quality, NoKnobPolicy>::new(t0, t0, q(0.75));
990 let outcome = job.finish(t0, AnytimeStopCause::BudgetExhausted, 3, &mut output);
991 assert!(outcome.published);
992 assert_eq!(output.metadata.status_txt.0.as_str(), "any:3it q=0.75 bdgt");
993 }
994
995 #[test]
996 fn quality_less_job_has_no_quality_in_stamp() {
997 let t0 = CuTime::from_millis(1);
998 let mut output: CuMsg<u32> = CuMsg::new(Some(7));
999 let job = AnytimeJob::<(), BarePolicy>::new(t0, t0, ());
1000 assert_eq!(job.check(t0 + CuDuration::from_secs(1)), None); let outcome = job.finish(t0, AnytimeStopCause::MaxRefines, 4, &mut output);
1002 assert!(outcome.published);
1003 assert_eq!(output.metadata.status_txt.0.as_str(), "any:4it max");
1004 }
1005
1006 #[test]
1007 fn base_site_terminal_outcomes() {
1008 let t0 = CuTime::from_millis(1);
1009 let now = t0 + CuDuration::from_micros(80);
1010
1011 let mut output: CuMsg<u32> = CuMsg::new(Some(9));
1012 let outcome = skip_stale(&mut output);
1013 assert_eq!(outcome.stop, AnytimeStopCause::SkippedStale);
1014 assert!(!outcome.published);
1015 assert_eq!(outcome.elapsed, CuDuration::default());
1016 assert_eq!(output.payload(), None);
1017 assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it stale!");
1018
1019 let mut output: CuMsg<u32> = CuMsg::new(Some(9));
1021 let outcome = abort_at_base(t0, now, &mut output);
1022 assert!(outcome.published);
1023 assert_eq!(outcome.elapsed, CuDuration::from_micros(80));
1024 assert_eq!(output.payload(), Some(&9));
1025
1026 let mut output: CuMsg<u32> = CuMsg::new(None);
1028 let outcome = abort_at_base(t0, now, &mut output);
1029 assert!(!outcome.published);
1030 assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it abort!");
1031 }
1032
1033 struct MaxRefinesPolicy;
1037 impl<Q: Copy + PartialOrd> AnytimePolicy<Q> for MaxRefinesPolicy {
1038 const TIME_BUDGET: Option<CuDuration> = None;
1039 const MAX_AGE: Option<CuDuration> = None;
1040 const MAX_STALL: Option<u32> = None;
1041 const MAX_REFINES: Option<u32> = Some(2);
1042 }
1043
1044 struct BudgetOnlyPolicy;
1047 impl<Q: Copy + PartialOrd> AnytimePolicy<Q> for BudgetOnlyPolicy {
1048 const TIME_BUDGET: Option<CuDuration> = Some(CuDuration(1_000_000));
1049 const MAX_AGE: Option<CuDuration> = None;
1050 const MAX_STALL: Option<u32> = None;
1051 const MAX_REFINES: Option<u32> = None;
1052 }
1053
1054 #[derive(Reflect)]
1057 #[reflect(no_field_bounds, from_reflect = false)]
1058 struct TickingTask {
1059 #[reflect(ignore)]
1060 clock: RobotClockMock,
1061 step: CuDuration,
1062 elapsed: CuDuration,
1063 }
1064
1065 impl TickingTask {
1066 fn tick(&mut self) {
1067 self.elapsed += self.step;
1068 self.clock.set_value(self.elapsed.0);
1069 }
1070 }
1071
1072 impl Freezable for TickingTask {}
1073
1074 impl CuAnytimeTask for TickingTask {
1075 type Input<'m> = input_msg!(u32);
1076 type Output<'m> = output_msg!(u32);
1077 type Resources<'r> = RobotClockMock;
1078 type Quality = Quality;
1079
1080 fn new(_config: Option<&ComponentConfig>, clock: RobotClockMock) -> CuResult<Self> {
1081 Ok(Self {
1082 clock,
1083 step: CuDuration::from_millis(1),
1084 elapsed: CuDuration::default(),
1085 })
1086 }
1087
1088 fn base<'i, 'o>(
1089 &mut self,
1090 _ctx: &CuContext,
1091 _input: &Self::Input<'i>,
1092 output: &mut Self::Output<'o>,
1093 ) -> CuResult<AnytimeStatus<Quality>> {
1094 self.tick();
1095 output.set_payload(0);
1096 Ok(AnytimeStatus::Improved(q(0.5)))
1097 }
1098
1099 fn refine<'o>(
1100 &mut self,
1101 _ctx: &CuContext,
1102 output: &mut Self::Output<'o>,
1103 ) -> CuResult<AnytimeStatus<Quality>> {
1104 self.tick();
1105 output.set_payload(output.payload().copied().unwrap_or(0) + 1);
1106 Ok(AnytimeStatus::Improved(q(0.5)))
1107 }
1108 }
1109
1110 #[derive(Reflect)]
1112 struct AbortingTask;
1113
1114 impl Freezable for AbortingTask {}
1115
1116 impl CuAnytimeTask for AbortingTask {
1117 type Input<'m> = input_msg!(u32);
1118 type Output<'m> = output_msg!(u32);
1119 type Resources<'r> = ();
1120 type Quality = Quality;
1121
1122 fn new(_config: Option<&ComponentConfig>, _resources: ()) -> CuResult<Self> {
1123 Ok(Self)
1124 }
1125
1126 fn base<'i, 'o>(
1127 &mut self,
1128 _ctx: &CuContext,
1129 _input: &Self::Input<'i>,
1130 output: &mut Self::Output<'o>,
1131 ) -> CuResult<AnytimeStatus<Quality>> {
1132 output.clear_payload();
1133 Ok(AnytimeStatus::Aborted)
1134 }
1135
1136 fn refine<'o>(
1137 &mut self,
1138 _ctx: &CuContext,
1139 _output: &mut Self::Output<'o>,
1140 ) -> CuResult<AnytimeStatus<Quality>> {
1141 unreachable!("refine after an abort at base")
1142 }
1143 }
1144
1145 #[derive(Reflect)]
1148 #[reflect(no_field_bounds, from_reflect = false)]
1149 struct HookOrderTask {
1150 #[reflect(ignore)]
1151 seq: Arc<AtomicU32>,
1152 }
1153
1154 impl HookOrderTask {
1155 fn tag(&self, digit: u32) {
1156 let seq = self.seq.load(Ordering::SeqCst);
1157 self.seq.store(seq * 10 + digit, Ordering::SeqCst);
1158 }
1159 }
1160
1161 impl Freezable for HookOrderTask {}
1162
1163 impl CuAnytimeTask for HookOrderTask {
1164 type Input<'m> = input_msg!(u32);
1165 type Output<'m> = output_msg!(u32);
1166 type Resources<'r> = Arc<AtomicU32>;
1167 type Quality = Quality;
1168
1169 fn new(_config: Option<&ComponentConfig>, seq: Arc<AtomicU32>) -> CuResult<Self> {
1170 Ok(Self { seq })
1171 }
1172
1173 fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
1174 self.tag(1);
1175 Ok(())
1176 }
1177
1178 fn base<'i, 'o>(
1179 &mut self,
1180 _ctx: &CuContext,
1181 _input: &Self::Input<'i>,
1182 output: &mut Self::Output<'o>,
1183 ) -> CuResult<AnytimeStatus<Quality>> {
1184 self.tag(2);
1185 output.set_payload(0);
1186 Ok(AnytimeStatus::Improved(q(0.5)))
1187 }
1188
1189 fn refine<'o>(
1190 &mut self,
1191 _ctx: &CuContext,
1192 _output: &mut Self::Output<'o>,
1193 ) -> CuResult<AnytimeStatus<Quality>> {
1194 self.tag(3);
1195 Ok(AnytimeStatus::Converged(q(1.0)))
1196 }
1197
1198 fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
1199 self.tag(4);
1200 Ok(())
1201 }
1202 }
1203
1204 #[test]
1205 fn runner_drives_the_per_job_hooks_in_order() {
1206 let ctx = CuContext::new_mock_clock().0;
1207 let seq = Arc::new(AtomicU32::new(0));
1208 let mut runner: CuAnytimeRunner<HookOrderTask, NoKnobPolicy> =
1209 CuAnytimeRunner::new(None, seq.clone()).unwrap();
1210
1211 process_job(&mut runner, &ctx, Tov::None);
1212 assert_eq!(seq.load(Ordering::SeqCst), 1234, "pre, base, refine, post");
1213
1214 process_job(&mut runner, &ctx, Tov::None);
1216 assert_eq!(seq.load(Ordering::SeqCst), 12_341_234);
1217 }
1218
1219 #[test]
1220 fn per_job_hooks_bracket_even_a_skipped_job() {
1221 let (ctx, clock) = CuContext::new_mock_clock();
1222 clock.set_value(CuDuration::from_millis(5).0);
1223 let seq = Arc::new(AtomicU32::new(0));
1224 let mut runner: CuAnytimeRunner<HookOrderTask, FullPolicy> =
1225 CuAnytimeRunner::new(None, seq.clone()).unwrap();
1226
1227 let output = process_job(&mut runner, &ctx, Tov::Time(CuTime::default()));
1230 assert_eq!(output.payload(), None);
1231 assert_eq!(seq.load(Ordering::SeqCst), 14, "pre, post only");
1232 }
1233
1234 #[test]
1235 fn runner_debug_state_forwards_to_the_wrapped_task() {
1236 let seq = Arc::new(AtomicU32::new(0));
1237 let runner: CuAnytimeRunner<HookOrderTask, NoKnobPolicy> =
1238 CuAnytimeRunner::new(None, seq).unwrap();
1239
1240 assert_eq!(
1243 <CuAnytimeRunner<HookOrderTask, NoKnobPolicy> as CuTask>::debug_state_type_path(),
1244 HookOrderTask::type_path()
1245 );
1246 let task_addr = core::ptr::from_ref(&runner.task).cast::<()>();
1247 let view_addr = runner.with_debug_state(|state| (state as *const dyn Reflect).cast::<()>());
1248 assert_eq!(view_addr, task_addr);
1249 }
1250
1251 fn process_job<T, P>(
1253 runner: &mut CuAnytimeRunner<T, P>,
1254 ctx: &CuContext,
1255 tov: Tov,
1256 ) -> CuMsg<u32>
1257 where
1258 T: for<'i, 'o> CuAnytimeTask<Input<'i> = CuMsg<u32>, Output<'o> = CuMsg<u32>>
1259 + GetTypeRegistration
1260 + TypePath
1261 + Send
1262 + Sync
1263 + 'static,
1264 P: AnytimePolicy<T::Quality> + Send + Sync + 'static,
1265 {
1266 let mut input = CuMsg::new(Some(3u32));
1267 input.tov = tov;
1268 let mut output = CuMsg::new(None);
1269 runner.process(ctx, &input, &mut output).unwrap();
1270 output
1271 }
1272
1273 #[test]
1274 fn runner_stops_at_the_quanta_bound() {
1275 let ctx = CuContext::new_mock_clock().0;
1276 let mut runner: CuAnytimeRunner<IncrementalSum, MaxRefinesPolicy> =
1277 CuAnytimeRunner::new(None, ()).unwrap();
1278
1279 let output = process_job(&mut runner, &ctx, Tov::None);
1280 assert_eq!(output.payload(), Some(&2));
1282 assert_eq!(output.metadata.status_txt.0.as_str(), "any:2it q=0.67 max");
1283 }
1284
1285 #[test]
1286 fn runner_stops_when_the_task_converges() {
1287 let ctx = CuContext::new_mock_clock().0;
1288 let mut runner: CuAnytimeRunner<IncrementalSum, FullPolicy> =
1289 CuAnytimeRunner::new(None, ()).unwrap();
1290
1291 let output = process_job(&mut runner, &ctx, Tov::None);
1294 assert_eq!(output.payload(), Some(&3));
1295 assert_eq!(output.metadata.status_txt.0.as_str(), "any:3it q=1.00 tgt");
1296 }
1297
1298 #[test]
1299 fn runner_stops_when_the_budget_is_exhausted() {
1300 let (ctx, clock) = CuContext::new_mock_clock();
1301 let mut runner: CuAnytimeRunner<TickingTask, BudgetOnlyPolicy> =
1302 CuAnytimeRunner::new(None, clock).unwrap();
1303
1304 let output = process_job(&mut runner, &ctx, Tov::None);
1306 assert_eq!(output.payload(), Some(&0));
1307 assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it q=0.50 bdgt");
1308 }
1309
1310 #[test]
1311 fn runner_skips_a_dead_on_arrival_input() {
1312 let (ctx, clock) = CuContext::new_mock_clock();
1313 clock.set_value(CuDuration::from_millis(5).0);
1314 let mut runner: CuAnytimeRunner<IncrementalSum, FullPolicy> =
1315 CuAnytimeRunner::new(None, ()).unwrap();
1316
1317 let output = process_job(&mut runner, &ctx, Tov::Time(CuTime::default()));
1319 assert_eq!(output.payload(), None);
1320 assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it stale!");
1321 }
1322
1323 #[test]
1324 fn runner_reports_an_abort_at_base() {
1325 let ctx = CuContext::new_mock_clock().0;
1326 let mut runner: CuAnytimeRunner<AbortingTask, FullPolicy> =
1327 CuAnytimeRunner::new(None, ()).unwrap();
1328
1329 let output = process_job(&mut runner, &ctx, Tov::None);
1330 assert_eq!(output.payload(), None);
1331 assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it abort!");
1332 }
1333
1334 #[test]
1335 fn runner_drops_a_result_below_the_quality_floor() {
1336 let (ctx, clock) = CuContext::new_mock_clock();
1337 let mut runner: CuAnytimeRunner<TickingTask, FloorPolicy> =
1340 CuAnytimeRunner::new(None, clock).unwrap();
1341
1342 let output = process_job(&mut runner, &ctx, Tov::None);
1343 assert_eq!(output.payload(), None, "below the floor: nothing published");
1344 assert_eq!(
1345 output.metadata.status_txt.0.as_str(),
1346 "any:0it q=0.50 bdgt!"
1347 );
1348 }
1349
1350 struct FloorPolicy;
1352 impl AnytimePolicy<Quality> for FloorPolicy {
1353 const TIME_BUDGET: Option<CuDuration> = Some(CuDuration(1_000_000));
1354 const MAX_AGE: Option<CuDuration> = None;
1355 const MAX_STALL: Option<u32> = None;
1356 const MAX_REFINES: Option<u32> = None;
1357
1358 fn below_floor(q: Quality) -> bool {
1359 q.partial_cmp(&quality_from_f32(0.8))
1360 .is_none_or(core::cmp::Ordering::is_lt)
1361 }
1362 }
1363}