use crate::config::ComponentConfig;
use crate::context::CuContext;
use crate::cutask::{CuMsg, CuMsgPack, CuMsgPayload, CuTask, Freezable};
use crate::reflect::{GetTypeRegistration, Reflect, TypePath, TypeRegistry};
use bincode::de::Decoder;
use bincode::enc::Encoder;
use bincode::error::{DecodeError, EncodeError};
use compact_str::format_compact;
use core::fmt::{Debug, Formatter, Result as FmtResult};
use core::marker::PhantomData;
use cu29_clock::{CuDuration, CuTime, Tov};
use cu29_traits::{CuCompactString, CuResult};
use cu29_units::si::f32::Ratio;
use cu29_units::si::ratio::ratio;
pub type Quality = Ratio;
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum AnytimeStatus<Q> {
Improved(Q),
Converged(Q),
Aborted,
}
pub trait CuAnytimeTask: Freezable + Reflect {
type Input<'m>: CuMsgPack;
type Output<'m>: CuMsgPayload;
type Resources<'r>;
type Quality: AnytimeQuality;
fn register_debug_state_types(registry: &mut TypeRegistry)
where
Self: GetTypeRegistration + Sized,
{
registry.register::<Self>();
}
fn debug_state_type_path() -> &'static str
where
Self: TypePath + Sized,
{
Self::type_path()
}
fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
where
Self: Sized,
{
f(self)
}
fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
where
Self: Sized;
fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
Ok(())
}
fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
Ok(())
}
fn base<'i, 'o>(
&mut self,
ctx: &CuContext,
input: &Self::Input<'i>,
output: &mut Self::Output<'o>,
) -> CuResult<AnytimeStatus<Self::Quality>>;
fn refine<'o>(
&mut self,
ctx: &CuContext,
output: &mut Self::Output<'o>,
) -> CuResult<AnytimeStatus<Self::Quality>>;
fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
Ok(())
}
fn stop(&mut self, _ctx: &CuContext) -> CuResult<()> {
Ok(())
}
}
#[inline(always)]
pub fn quality_from_f32(value: f32) -> Quality {
Quality::new::<ratio>(value)
}
#[inline(always)]
pub fn quality_to_f32(quality: Quality) -> f32 {
quality.get::<ratio>()
}
pub trait AnytimeQuality: Copy + PartialOrd {
#[inline(always)]
fn ratio(self) -> Option<f32> {
None
}
}
impl AnytimeQuality for Quality {
#[inline(always)]
fn ratio(self) -> Option<f32> {
Some(quality_to_f32(self))
}
}
impl AnytimeQuality for () {}
#[doc(hidden)]
#[diagnostic::on_unimplemented(
message = "the anytime policy `{Self}` is pinned to the shared quality scale, but this task's `Quality` is `{Q}`",
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"
)]
pub trait AnytimePolicy<Q> {
const TIME_BUDGET: Option<CuDuration>;
const MAX_AGE: Option<CuDuration>;
const MAX_STALL: Option<u32>;
const MAX_REFINES: Option<u32>;
#[inline(always)]
fn target_met(_q: Q) -> bool {
false
}
#[inline(always)]
fn below_floor(_q: Q) -> bool {
false
}
}
#[doc(hidden)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AnytimeStopCause {
Converged,
TargetMet,
BudgetExhausted,
AgeExceeded,
SkippedStale,
MaxRefines,
Stalled,
Aborted,
}
impl AnytimeStopCause {
pub fn label(self) -> &'static str {
match self {
AnytimeStopCause::Converged => "conv",
AnytimeStopCause::TargetMet => "tgt",
AnytimeStopCause::BudgetExhausted => "bdgt",
AnytimeStopCause::AgeExceeded => "age",
AnytimeStopCause::SkippedStale => "stale",
AnytimeStopCause::MaxRefines => "max",
AnytimeStopCause::Stalled => "stall",
AnytimeStopCause::Aborted => "abort",
}
}
}
#[doc(hidden)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct AnytimeOutcome {
pub iterations: u32,
pub elapsed: CuDuration,
pub stop: AnytimeStopCause,
pub published: bool,
}
#[doc(hidden)]
pub struct AnytimeJob<Q, P> {
t0: CuTime,
anchor: CuTime,
best: Q,
stall: u32,
_policy: PhantomData<P>,
}
impl<Q: Copy, P> Clone for AnytimeJob<Q, P> {
fn clone(&self) -> Self {
Self {
t0: self.t0,
anchor: self.anchor,
best: self.best,
stall: self.stall,
_policy: PhantomData,
}
}
}
impl<Q: Debug, P> Debug for AnytimeJob<Q, P> {
fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
f.debug_struct("AnytimeJob")
.field("t0", &self.t0)
.field("anchor", &self.anchor)
.field("best", &self.best)
.field("stall", &self.stall)
.finish()
}
}
impl<Q: AnytimeQuality, P: AnytimePolicy<Q>> AnytimeJob<Q, P> {
pub fn new(t0: CuTime, anchor: CuTime, quality: Q) -> Self {
Self {
t0,
anchor,
best: quality,
stall: 0,
_policy: PhantomData,
}
}
pub fn record(&mut self, quality: Q) {
if quality > self.best || self.best.partial_cmp(&self.best).is_none() {
self.best = quality;
self.stall = 0;
} else if P::MAX_STALL.is_some() {
self.stall += 1;
}
}
pub fn check(&self, now: CuTime) -> Option<AnytimeStopCause> {
if P::target_met(self.best) {
return Some(AnytimeStopCause::TargetMet);
}
if let Some(budget) = P::TIME_BUDGET
&& now >= self.t0 + budget
{
return Some(AnytimeStopCause::BudgetExhausted);
}
if let Some(age) = P::MAX_AGE
&& now >= self.anchor + age
{
return Some(AnytimeStopCause::AgeExceeded);
}
if let Some(max_stall) = P::MAX_STALL
&& self.stall >= max_stall
{
return Some(AnytimeStopCause::Stalled);
}
None
}
pub fn finish<O: CuMsgPayload>(
self,
now: CuTime,
cause: AnytimeStopCause,
iterations: u32,
output: &mut CuMsg<O>,
) -> AnytimeOutcome {
let published = if P::below_floor(self.best) {
output.clear_payload();
false
} else {
output.payload().is_some()
};
stamp(output, iterations, self.best.ratio(), cause, published);
AnytimeOutcome {
iterations,
elapsed: now - self.t0,
stop: cause,
published,
}
}
}
fn stamp<O: CuMsgPayload>(
output: &mut CuMsg<O>,
iterations: u32,
quality: Option<f32>,
cause: AnytimeStopCause,
published: bool,
) {
let not_published = if published { "" } else { "!" };
output.metadata.status_txt = CuCompactString(match quality {
Some(q) => format_compact!(
"any:{}it q={:.2} {}{}",
iterations,
q,
cause.label(),
not_published
),
None => format_compact!("any:{}it {}{}", iterations, cause.label(), not_published),
});
}
#[doc(hidden)]
#[inline(always)]
pub fn anchor_from_tov(tov: Tov, now: CuTime) -> CuTime {
match tov {
Tov::Time(time) => time,
Tov::Range(range) => range.start,
Tov::None => now,
}
}
#[doc(hidden)]
pub fn skip_stale<O: CuMsgPayload>(output: &mut CuMsg<O>) -> AnytimeOutcome {
output.clear_payload();
stamp(output, 0, None, AnytimeStopCause::SkippedStale, false);
AnytimeOutcome {
iterations: 0,
elapsed: CuDuration::default(),
stop: AnytimeStopCause::SkippedStale,
published: false,
}
}
#[doc(hidden)]
pub fn abort_at_base<O: CuMsgPayload>(
t0: CuTime,
now: CuTime,
output: &mut CuMsg<O>,
) -> AnytimeOutcome {
let published = output.payload().is_some();
stamp(output, 0, None, AnytimeStopCause::Aborted, published);
AnytimeOutcome {
iterations: 0,
elapsed: now - t0,
stop: AnytimeStopCause::Aborted,
published,
}
}
#[doc(hidden)]
#[derive(Reflect)]
#[reflect(no_field_bounds, from_reflect = false, type_path = false)]
pub struct CuAnytimeRunner<T, P>
where
T: Send + Sync + 'static,
P: Send + Sync + 'static,
{
#[reflect(ignore)]
task: T,
#[reflect(ignore)]
_policy: PhantomData<P>,
}
impl<T, P> TypePath for CuAnytimeRunner<T, P>
where
T: Send + Sync + 'static,
P: Send + Sync + 'static,
{
fn type_path() -> &'static str {
"cu29_runtime::cutask_anytime::CuAnytimeRunner"
}
fn short_type_path() -> &'static str {
"CuAnytimeRunner"
}
fn type_ident() -> Option<&'static str> {
Some("CuAnytimeRunner")
}
fn crate_name() -> Option<&'static str> {
Some("cu29_runtime")
}
fn module_path() -> Option<&'static str> {
Some("cutask_anytime")
}
}
impl<T, P> Freezable for CuAnytimeRunner<T, P>
where
T: Freezable + Send + Sync + 'static,
P: Send + Sync + 'static,
{
fn freeze<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
self.task.freeze(encoder)
}
fn thaw<D: Decoder>(&mut self, decoder: &mut D) -> Result<(), DecodeError> {
self.task.thaw(decoder)
}
}
impl<T, I, O, P> CuTask for CuAnytimeRunner<T, P>
where
T: for<'i, 'o> CuAnytimeTask<Input<'i> = CuMsg<I>, Output<'o> = CuMsg<O>>
+ GetTypeRegistration
+ TypePath
+ Send
+ Sync
+ 'static,
I: CuMsgPayload,
O: CuMsgPayload,
P: AnytimePolicy<T::Quality> + Send + Sync + 'static,
{
type Resources<'r> = T::Resources<'r>;
type Input<'m> = T::Input<'m>;
type Output<'m> = T::Output<'m>;
fn register_debug_state_types(registry: &mut TypeRegistry)
where
Self: GetTypeRegistration + Sized,
{
T::register_debug_state_types(registry);
}
fn debug_state_type_path() -> &'static str
where
Self: TypePath + Sized,
{
T::debug_state_type_path()
}
fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
where
Self: Sized,
{
self.task.with_debug_state(f)
}
fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
where
Self: Sized,
{
Ok(Self {
task: T::new(config, resources)?,
_policy: PhantomData,
})
}
fn start(&mut self, ctx: &CuContext) -> CuResult<()> {
self.task.start(ctx)
}
fn process<'i, 'o>(
&mut self,
ctx: &CuContext,
input: &Self::Input<'i>,
output: &mut Self::Output<'o>,
) -> CuResult<()> {
self.task.preprocess(ctx)?;
let job = run_job::<T, I, O, P>(&mut self.task, ctx, input, output);
let post = self.task.postprocess(ctx);
job.and(post)
}
fn stop(&mut self, ctx: &CuContext) -> CuResult<()> {
self.task.stop(ctx)
}
}
fn run_job<T, I, O, P>(
task: &mut T,
ctx: &CuContext,
input: &CuMsg<I>,
output: &mut CuMsg<O>,
) -> CuResult<()>
where
T: for<'i, 'o> CuAnytimeTask<Input<'i> = CuMsg<I>, Output<'o> = CuMsg<O>>,
I: CuMsgPayload,
O: CuMsgPayload,
P: AnytimePolicy<T::Quality>,
{
let start = if P::TIME_BUDGET.is_some() || P::MAX_AGE.is_some() {
ctx.now()
} else {
CuTime::default()
};
let anchor = if P::MAX_AGE.is_some() {
anchor_from_tov(input.tov, start)
} else {
start
};
if let Some(max_age) = P::MAX_AGE
&& start >= anchor + max_age
{
skip_stale(output);
return Ok(());
}
let mut job = match task.base(ctx, input, output)? {
AnytimeStatus::Improved(quality) => AnytimeJob::<_, P>::new(start, anchor, quality),
AnytimeStatus::Converged(quality) => {
AnytimeJob::<_, P>::new(start, anchor, quality).finish(
start,
AnytimeStopCause::Converged,
0,
output,
);
return Ok(());
}
AnytimeStatus::Aborted => {
abort_at_base(start, start, output);
return Ok(());
}
};
let mut ran = 0u32;
loop {
let now = if P::TIME_BUDGET.is_some() || P::MAX_AGE.is_some() {
ctx.now()
} else {
CuTime::default()
};
if let Some(cause) = job.check(now) {
job.finish(now, cause, ran, output);
return Ok(());
}
let status = task.refine(ctx, output)?;
ran += 1;
match status {
AnytimeStatus::Improved(quality) => {
job.record(quality);
if let Some(max_refines) = P::MAX_REFINES
&& ran >= max_refines
{
job.finish(now, AnytimeStopCause::MaxRefines, ran, output);
return Ok(());
}
}
AnytimeStatus::Converged(quality) => {
job.record(quality);
job.finish(now, AnytimeStopCause::Converged, ran, output);
return Ok(());
}
AnytimeStatus::Aborted => {
job.finish(now, AnytimeStopCause::Aborted, ran, output);
return Ok(());
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::cutask::CuMsg;
use crate::input_msg;
use crate::output_msg;
use alloc::sync::Arc;
use core::sync::atomic::{AtomicU32, Ordering};
use cu29_clock::RobotClockMock;
fn q(v: f32) -> Quality {
quality_from_f32(v)
}
#[derive(Reflect)]
struct IncrementalSum {
target: u32,
acc: u32,
}
impl Freezable for IncrementalSum {}
impl CuAnytimeTask for IncrementalSum {
type Input<'m> = input_msg!(u32);
type Output<'m> = output_msg!(u32);
type Resources<'r> = ();
type Quality = Quality;
fn new(
_config: Option<&ComponentConfig>,
_resources: Self::Resources<'_>,
) -> CuResult<Self> {
Ok(Self { target: 0, acc: 0 })
}
fn base<'i, 'o>(
&mut self,
_ctx: &CuContext,
input: &Self::Input<'i>,
output: &mut Self::Output<'o>,
) -> CuResult<AnytimeStatus<Quality>> {
self.target = *input.payload().ok_or("no input")?;
self.acc = 0;
output.set_payload(self.acc);
Ok(AnytimeStatus::Improved(q(0.0)))
}
fn refine<'o>(
&mut self,
_ctx: &CuContext,
output: &mut Self::Output<'o>,
) -> CuResult<AnytimeStatus<Quality>> {
if self.acc == self.target {
return Ok(AnytimeStatus::Converged(q(1.0)));
}
self.acc += 1;
output.set_payload(self.acc);
Ok(AnytimeStatus::Improved(q(
self.acc as f32 / self.target as f32
)))
}
}
#[test]
fn base_then_refine_until_converged() {
let ctx = CuContext::new_with_clock();
let mut task = IncrementalSum::new(None, ()).unwrap();
let input = CuMsg::new(Some(3u32));
let mut output = CuMsg::new(None);
task.start(&ctx).unwrap();
task.preprocess(&ctx).unwrap();
let status = task.base(&ctx, &input, &mut output).unwrap();
assert!(matches!(status, AnytimeStatus::Improved(_)));
assert_eq!(output.payload(), Some(&0));
let mut best_quality = q(0.0);
for _ in 0..8 {
match task.refine(&ctx, &mut output).unwrap() {
AnytimeStatus::Improved(quality) => best_quality = quality,
AnytimeStatus::Converged(quality) => {
best_quality = quality;
break;
}
status => panic!("unexpected status: {status:?}"),
}
}
assert_eq!(output.payload(), Some(&3));
assert_eq!(quality_to_f32(best_quality), 1.0);
task.postprocess(&ctx).unwrap();
task.stop(&ctx).unwrap();
}
struct FullPolicy;
impl AnytimePolicy<Quality> for FullPolicy {
const TIME_BUDGET: Option<CuDuration> = Some(CuDuration(1_000_000));
const MAX_AGE: Option<CuDuration> = Some(CuDuration(2_000_000));
const MAX_STALL: Option<u32> = Some(2);
const MAX_REFINES: Option<u32> = Some(8);
fn target_met(q: Quality) -> bool {
q >= quality_from_f32(0.9)
}
fn below_floor(q: Quality) -> bool {
q.partial_cmp(&quality_from_f32(0.3))
.is_none_or(core::cmp::Ordering::is_lt)
}
}
struct NoKnobPolicy;
impl<Q: Copy + PartialOrd> AnytimePolicy<Q> for NoKnobPolicy {
const TIME_BUDGET: Option<CuDuration> = None;
const MAX_AGE: Option<CuDuration> = None;
const MAX_STALL: Option<u32> = None;
const MAX_REFINES: Option<u32> = None;
}
struct BarePolicy;
impl AnytimePolicy<()> for BarePolicy {
const TIME_BUDGET: Option<CuDuration> = None;
const MAX_AGE: Option<CuDuration> = None;
const MAX_STALL: Option<u32> = None;
const MAX_REFINES: Option<u32> = None;
}
#[test]
fn check_attribution_order_is_target_budget_age_stall() {
let t0 = CuTime::from_millis(10);
let job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.95));
assert_eq!(
job.check(t0 + CuDuration::from_millis(5)),
Some(AnytimeStopCause::TargetMet)
);
let job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.5));
assert_eq!(
job.check(t0 + CuDuration::from_millis(5)),
Some(AnytimeStopCause::BudgetExhausted)
);
let anchor = t0 - CuDuration::from_millis(2);
let job = AnytimeJob::<Quality, FullPolicy>::new(t0, anchor, q(0.5));
assert_eq!(
job.check(t0 + CuDuration::from_nanos(1)),
Some(AnytimeStopCause::AgeExceeded)
);
let job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.5));
assert_eq!(job.check(t0), None);
}
#[test]
fn stall_counts_quanta_without_improvement() {
let t0 = CuTime::from_millis(1);
let mut job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.5));
job.record(q(0.5)); assert_eq!(job.check(t0), None);
job.record(q(0.6)); assert_eq!(job.check(t0), None);
job.record(q(0.6));
job.record(q(0.6)); assert_eq!(job.check(t0), Some(AnytimeStopCause::Stalled));
}
#[test]
fn finish_gates_on_floor_and_stamps_status() {
let t0 = CuTime::from_millis(1);
let now = t0 + CuDuration::from_micros(250);
let mut output: CuMsg<u32> = CuMsg::new(Some(42));
let job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.5));
let outcome = job.finish(now, AnytimeStopCause::BudgetExhausted, 3, &mut output);
assert!(outcome.published);
assert_eq!(outcome.iterations, 3);
assert_eq!(outcome.elapsed, CuDuration::from_micros(250));
assert_eq!(output.payload(), Some(&42));
assert_eq!(output.metadata.status_txt.0.as_str(), "any:3it q=0.50 bdgt");
let mut output: CuMsg<u32> = CuMsg::new(Some(42));
let job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.1));
let outcome = job.finish(now, AnytimeStopCause::MaxRefines, 2, &mut output);
assert!(!outcome.published);
assert_eq!(output.payload(), None);
assert_eq!(output.metadata.status_txt.0.as_str(), "any:2it q=0.10 max!");
}
#[test]
fn stamp_stays_inline() {
const CAUSES: [AnytimeStopCause; 8] = [
AnytimeStopCause::Converged,
AnytimeStopCause::TargetMet,
AnytimeStopCause::BudgetExhausted,
AnytimeStopCause::AgeExceeded,
AnytimeStopCause::SkippedStale,
AnytimeStopCause::MaxRefines,
AnytimeStopCause::Stalled,
AnytimeStopCause::Aborted,
];
for cause in CAUSES {
for published in [true, false] {
for quality in [None, Some(0.0), Some(1.0)] {
let mut output: CuMsg<u32> = CuMsg::new(Some(1));
stamp(&mut output, 9999, quality, cause, published);
let stamped = &output.metadata.status_txt.0;
assert!(
!stamped.is_heap_allocated(),
"stamp allocates on the real-time path: {stamped:?} ({} bytes)",
stamped.len()
);
}
}
}
}
#[test]
fn nan_quality_fails_closed() {
let t0 = CuTime::from_millis(1);
let mut output: CuMsg<u32> = CuMsg::new(Some(1));
let job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(f32::NAN));
let outcome = job.finish(t0, AnytimeStopCause::MaxRefines, 1, &mut output);
assert!(!outcome.published);
assert_eq!(output.payload(), None);
let mut job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(f32::NAN));
job.record(q(0.4));
let mut output: CuMsg<u32> = CuMsg::new(Some(1));
let outcome = job.finish(t0, AnytimeStopCause::MaxRefines, 1, &mut output);
assert!(outcome.published);
let mut job = AnytimeJob::<Quality, FullPolicy>::new(t0, t0, q(0.5));
job.record(q(f32::NAN));
let mut output: CuMsg<u32> = CuMsg::new(Some(1));
let outcome = job.finish(t0, AnytimeStopCause::MaxRefines, 1, &mut output);
assert!(outcome.published);
assert_eq!(output.metadata.status_txt.0.as_str(), "any:1it q=0.50 max");
}
#[test]
fn quality_reaches_stamp_without_quality_knobs() {
let t0 = CuTime::from_millis(1);
let mut output: CuMsg<u32> = CuMsg::new(Some(7));
let job = AnytimeJob::<Quality, NoKnobPolicy>::new(t0, t0, q(0.75));
let outcome = job.finish(t0, AnytimeStopCause::BudgetExhausted, 3, &mut output);
assert!(outcome.published);
assert_eq!(output.metadata.status_txt.0.as_str(), "any:3it q=0.75 bdgt");
}
#[test]
fn quality_less_job_has_no_quality_in_stamp() {
let t0 = CuTime::from_millis(1);
let mut output: CuMsg<u32> = CuMsg::new(Some(7));
let job = AnytimeJob::<(), BarePolicy>::new(t0, t0, ());
assert_eq!(job.check(t0 + CuDuration::from_secs(1)), None); let outcome = job.finish(t0, AnytimeStopCause::MaxRefines, 4, &mut output);
assert!(outcome.published);
assert_eq!(output.metadata.status_txt.0.as_str(), "any:4it max");
}
#[test]
fn base_site_terminal_outcomes() {
let t0 = CuTime::from_millis(1);
let now = t0 + CuDuration::from_micros(80);
let mut output: CuMsg<u32> = CuMsg::new(Some(9));
let outcome = skip_stale(&mut output);
assert_eq!(outcome.stop, AnytimeStopCause::SkippedStale);
assert!(!outcome.published);
assert_eq!(outcome.elapsed, CuDuration::default());
assert_eq!(output.payload(), None);
assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it stale!");
let mut output: CuMsg<u32> = CuMsg::new(Some(9));
let outcome = abort_at_base(t0, now, &mut output);
assert!(outcome.published);
assert_eq!(outcome.elapsed, CuDuration::from_micros(80));
assert_eq!(output.payload(), Some(&9));
let mut output: CuMsg<u32> = CuMsg::new(None);
let outcome = abort_at_base(t0, now, &mut output);
assert!(!outcome.published);
assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it abort!");
}
struct MaxRefinesPolicy;
impl<Q: Copy + PartialOrd> AnytimePolicy<Q> for MaxRefinesPolicy {
const TIME_BUDGET: Option<CuDuration> = None;
const MAX_AGE: Option<CuDuration> = None;
const MAX_STALL: Option<u32> = None;
const MAX_REFINES: Option<u32> = Some(2);
}
struct BudgetOnlyPolicy;
impl<Q: Copy + PartialOrd> AnytimePolicy<Q> for BudgetOnlyPolicy {
const TIME_BUDGET: Option<CuDuration> = Some(CuDuration(1_000_000));
const MAX_AGE: Option<CuDuration> = None;
const MAX_STALL: Option<u32> = None;
const MAX_REFINES: Option<u32> = None;
}
#[derive(Reflect)]
#[reflect(no_field_bounds, from_reflect = false)]
struct TickingTask {
#[reflect(ignore)]
clock: RobotClockMock,
step: CuDuration,
elapsed: CuDuration,
}
impl TickingTask {
fn tick(&mut self) {
self.elapsed += self.step;
self.clock.set_value(self.elapsed.0);
}
}
impl Freezable for TickingTask {}
impl CuAnytimeTask for TickingTask {
type Input<'m> = input_msg!(u32);
type Output<'m> = output_msg!(u32);
type Resources<'r> = RobotClockMock;
type Quality = Quality;
fn new(_config: Option<&ComponentConfig>, clock: RobotClockMock) -> CuResult<Self> {
Ok(Self {
clock,
step: CuDuration::from_millis(1),
elapsed: CuDuration::default(),
})
}
fn base<'i, 'o>(
&mut self,
_ctx: &CuContext,
_input: &Self::Input<'i>,
output: &mut Self::Output<'o>,
) -> CuResult<AnytimeStatus<Quality>> {
self.tick();
output.set_payload(0);
Ok(AnytimeStatus::Improved(q(0.5)))
}
fn refine<'o>(
&mut self,
_ctx: &CuContext,
output: &mut Self::Output<'o>,
) -> CuResult<AnytimeStatus<Quality>> {
self.tick();
output.set_payload(output.payload().copied().unwrap_or(0) + 1);
Ok(AnytimeStatus::Improved(q(0.5)))
}
}
#[derive(Reflect)]
struct AbortingTask;
impl Freezable for AbortingTask {}
impl CuAnytimeTask for AbortingTask {
type Input<'m> = input_msg!(u32);
type Output<'m> = output_msg!(u32);
type Resources<'r> = ();
type Quality = Quality;
fn new(_config: Option<&ComponentConfig>, _resources: ()) -> CuResult<Self> {
Ok(Self)
}
fn base<'i, 'o>(
&mut self,
_ctx: &CuContext,
_input: &Self::Input<'i>,
output: &mut Self::Output<'o>,
) -> CuResult<AnytimeStatus<Quality>> {
output.clear_payload();
Ok(AnytimeStatus::Aborted)
}
fn refine<'o>(
&mut self,
_ctx: &CuContext,
_output: &mut Self::Output<'o>,
) -> CuResult<AnytimeStatus<Quality>> {
unreachable!("refine after an abort at base")
}
}
#[derive(Reflect)]
#[reflect(no_field_bounds, from_reflect = false)]
struct HookOrderTask {
#[reflect(ignore)]
seq: Arc<AtomicU32>,
}
impl HookOrderTask {
fn tag(&self, digit: u32) {
let seq = self.seq.load(Ordering::SeqCst);
self.seq.store(seq * 10 + digit, Ordering::SeqCst);
}
}
impl Freezable for HookOrderTask {}
impl CuAnytimeTask for HookOrderTask {
type Input<'m> = input_msg!(u32);
type Output<'m> = output_msg!(u32);
type Resources<'r> = Arc<AtomicU32>;
type Quality = Quality;
fn new(_config: Option<&ComponentConfig>, seq: Arc<AtomicU32>) -> CuResult<Self> {
Ok(Self { seq })
}
fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
self.tag(1);
Ok(())
}
fn base<'i, 'o>(
&mut self,
_ctx: &CuContext,
_input: &Self::Input<'i>,
output: &mut Self::Output<'o>,
) -> CuResult<AnytimeStatus<Quality>> {
self.tag(2);
output.set_payload(0);
Ok(AnytimeStatus::Improved(q(0.5)))
}
fn refine<'o>(
&mut self,
_ctx: &CuContext,
_output: &mut Self::Output<'o>,
) -> CuResult<AnytimeStatus<Quality>> {
self.tag(3);
Ok(AnytimeStatus::Converged(q(1.0)))
}
fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
self.tag(4);
Ok(())
}
}
#[test]
fn runner_drives_the_per_job_hooks_in_order() {
let ctx = CuContext::new_mock_clock().0;
let seq = Arc::new(AtomicU32::new(0));
let mut runner: CuAnytimeRunner<HookOrderTask, NoKnobPolicy> =
CuAnytimeRunner::new(None, seq.clone()).unwrap();
process_job(&mut runner, &ctx, Tov::None);
assert_eq!(seq.load(Ordering::SeqCst), 1234, "pre, base, refine, post");
process_job(&mut runner, &ctx, Tov::None);
assert_eq!(seq.load(Ordering::SeqCst), 12_341_234);
}
#[test]
fn per_job_hooks_bracket_even_a_skipped_job() {
let (ctx, clock) = CuContext::new_mock_clock();
clock.set_value(CuDuration::from_millis(5).0);
let seq = Arc::new(AtomicU32::new(0));
let mut runner: CuAnytimeRunner<HookOrderTask, FullPolicy> =
CuAnytimeRunner::new(None, seq.clone()).unwrap();
let output = process_job(&mut runner, &ctx, Tov::Time(CuTime::default()));
assert_eq!(output.payload(), None);
assert_eq!(seq.load(Ordering::SeqCst), 14, "pre, post only");
}
#[test]
fn runner_debug_state_forwards_to_the_wrapped_task() {
let seq = Arc::new(AtomicU32::new(0));
let runner: CuAnytimeRunner<HookOrderTask, NoKnobPolicy> =
CuAnytimeRunner::new(None, seq).unwrap();
assert_eq!(
<CuAnytimeRunner<HookOrderTask, NoKnobPolicy> as CuTask>::debug_state_type_path(),
HookOrderTask::type_path()
);
let task_addr = core::ptr::from_ref(&runner.task).cast::<()>();
let view_addr = runner.with_debug_state(|state| (state as *const dyn Reflect).cast::<()>());
assert_eq!(view_addr, task_addr);
}
fn process_job<T, P>(
runner: &mut CuAnytimeRunner<T, P>,
ctx: &CuContext,
tov: Tov,
) -> CuMsg<u32>
where
T: for<'i, 'o> CuAnytimeTask<Input<'i> = CuMsg<u32>, Output<'o> = CuMsg<u32>>
+ GetTypeRegistration
+ TypePath
+ Send
+ Sync
+ 'static,
P: AnytimePolicy<T::Quality> + Send + Sync + 'static,
{
let mut input = CuMsg::new(Some(3u32));
input.tov = tov;
let mut output = CuMsg::new(None);
runner.process(ctx, &input, &mut output).unwrap();
output
}
#[test]
fn runner_stops_at_the_quanta_bound() {
let ctx = CuContext::new_mock_clock().0;
let mut runner: CuAnytimeRunner<IncrementalSum, MaxRefinesPolicy> =
CuAnytimeRunner::new(None, ()).unwrap();
let output = process_job(&mut runner, &ctx, Tov::None);
assert_eq!(output.payload(), Some(&2));
assert_eq!(output.metadata.status_txt.0.as_str(), "any:2it q=0.67 max");
}
#[test]
fn runner_stops_when_the_task_converges() {
let ctx = CuContext::new_mock_clock().0;
let mut runner: CuAnytimeRunner<IncrementalSum, FullPolicy> =
CuAnytimeRunner::new(None, ()).unwrap();
let output = process_job(&mut runner, &ctx, Tov::None);
assert_eq!(output.payload(), Some(&3));
assert_eq!(output.metadata.status_txt.0.as_str(), "any:3it q=1.00 tgt");
}
#[test]
fn runner_stops_when_the_budget_is_exhausted() {
let (ctx, clock) = CuContext::new_mock_clock();
let mut runner: CuAnytimeRunner<TickingTask, BudgetOnlyPolicy> =
CuAnytimeRunner::new(None, clock).unwrap();
let output = process_job(&mut runner, &ctx, Tov::None);
assert_eq!(output.payload(), Some(&0));
assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it q=0.50 bdgt");
}
#[test]
fn runner_skips_a_dead_on_arrival_input() {
let (ctx, clock) = CuContext::new_mock_clock();
clock.set_value(CuDuration::from_millis(5).0);
let mut runner: CuAnytimeRunner<IncrementalSum, FullPolicy> =
CuAnytimeRunner::new(None, ()).unwrap();
let output = process_job(&mut runner, &ctx, Tov::Time(CuTime::default()));
assert_eq!(output.payload(), None);
assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it stale!");
}
#[test]
fn runner_reports_an_abort_at_base() {
let ctx = CuContext::new_mock_clock().0;
let mut runner: CuAnytimeRunner<AbortingTask, FullPolicy> =
CuAnytimeRunner::new(None, ()).unwrap();
let output = process_job(&mut runner, &ctx, Tov::None);
assert_eq!(output.payload(), None);
assert_eq!(output.metadata.status_txt.0.as_str(), "any:0it abort!");
}
#[test]
fn runner_drops_a_result_below_the_quality_floor() {
let (ctx, clock) = CuContext::new_mock_clock();
let mut runner: CuAnytimeRunner<TickingTask, FloorPolicy> =
CuAnytimeRunner::new(None, clock).unwrap();
let output = process_job(&mut runner, &ctx, Tov::None);
assert_eq!(output.payload(), None, "below the floor: nothing published");
assert_eq!(
output.metadata.status_txt.0.as_str(),
"any:0it q=0.50 bdgt!"
);
}
struct FloorPolicy;
impl AnytimePolicy<Quality> for FloorPolicy {
const TIME_BUDGET: Option<CuDuration> = Some(CuDuration(1_000_000));
const MAX_AGE: Option<CuDuration> = None;
const MAX_STALL: Option<u32> = None;
const MAX_REFINES: Option<u32> = None;
fn below_floor(q: Quality) -> bool {
q.partial_cmp(&quality_from_f32(0.8))
.is_none_or(core::cmp::Ordering::is_lt)
}
}
}