cu29_runtime/
cuasynctask.rs

1use crate::config::ComponentConfig;
2use crate::context::CuContext;
3use crate::cutask::{CuMsg, CuMsgPayload, CuSrcTask, CuTask, Freezable};
4use crate::reflect::{Reflect, TypePath};
5use bincode::de::{Decode, Decoder};
6use bincode::enc::{Encode, Encoder};
7use bincode::error::{DecodeError, EncodeError};
8use cu29_clock::CuTime;
9use cu29_traits::{CuError, CuResult};
10use rayon::ThreadPool;
11use std::sync::{Arc, Mutex};
12
13struct AsyncState {
14    processing: bool,
15    ready_at: Option<CuTime>,
16    last_error: Option<CuError>,
17}
18
19fn encode_async_state<E: Encoder>(state: &AsyncState, encoder: &mut E) -> Result<(), EncodeError> {
20    if state.processing {
21        return Err(EncodeError::OtherString(
22            "cannot freeze async task while background work is in progress".to_string(),
23        ));
24    }
25
26    Encode::encode(&state.ready_at, encoder)?;
27    let last_error = state.last_error.as_ref().map(ToString::to_string);
28    Encode::encode(&last_error, encoder)?;
29    Ok(())
30}
31
32fn decode_async_state<D: Decoder>(
33    state: &mut AsyncState,
34    decoder: &mut D,
35) -> Result<(), DecodeError> {
36    state.processing = false;
37    state.ready_at = Decode::decode(decoder)?;
38    let last_error: Option<String> = Decode::decode(decoder)?;
39    state.last_error = last_error.map(CuError::from);
40    Ok(())
41}
42
43fn encode_buffered_output<O, E>(output: &CuMsg<O>, encoder: &mut E) -> Result<(), EncodeError>
44where
45    O: CuMsgPayload + Send + 'static,
46    E: Encoder,
47{
48    let bytes = bincode::encode_to_vec(output, bincode::config::standard())?;
49    Encode::encode(&bytes, encoder)
50}
51
52fn decode_buffered_output<O, D>(decoder: &mut D) -> Result<CuMsg<O>, DecodeError>
53where
54    O: CuMsgPayload + Send + 'static,
55    D: Decoder,
56{
57    let bytes: Vec<u8> = Decode::decode(decoder)?;
58    let (output, bytes_read): (CuMsg<O>, usize) =
59        bincode::decode_from_slice(&bytes, bincode::config::standard())?;
60    if bytes_read != bytes.len() {
61        return Err(DecodeError::OtherString(
62            "async task buffered output snapshot had trailing bytes".to_string(),
63        ));
64    }
65    Ok(output)
66}
67
68fn record_async_error(state: &Mutex<AsyncState>, error: CuError) {
69    let mut guard = match state.lock() {
70        Ok(guard) => guard,
71        Err(poison) => poison.into_inner(),
72    };
73    guard.processing = false;
74    guard.ready_at = None;
75    guard.last_error = Some(error);
76}
77
78fn begin_background_poll<O>(
79    ctx: &CuContext,
80    state: &Mutex<AsyncState>,
81    buffered_output: &Mutex<CuMsg<O>>,
82    real_output: &mut CuMsg<O>,
83) -> CuResult<bool>
84where
85    O: CuMsgPayload + Send + 'static,
86{
87    {
88        let mut state = state.lock().map_err(|_| {
89            CuError::from("Async task state mutex poisoned while scheduling background work")
90        })?;
91        if let Some(error) = state.last_error.take() {
92            return Err(error);
93        }
94        if state.processing {
95            *real_output = CuMsg::default();
96            return Ok(false);
97        }
98
99        if let Some(ready_at) = state.ready_at
100            && ctx.now() < ready_at
101        {
102            *real_output = CuMsg::default();
103            return Ok(false);
104        }
105
106        state.processing = true;
107        state.ready_at = None;
108    }
109
110    let buffered_output = buffered_output.lock().map_err(|_| {
111        let error = CuError::from("Async task output mutex poisoned");
112        record_async_error(state, error.clone());
113        error
114    })?;
115    *real_output = buffered_output.clone();
116    Ok(true)
117}
118
119fn finalize_background_run<O>(
120    state: &Mutex<AsyncState>,
121    output_ref: &mut CuMsg<O>,
122    fallback_end: CuTime,
123    task_result: CuResult<()>,
124) where
125    O: CuMsgPayload + Send + 'static,
126{
127    let mut guard = state.lock().unwrap_or_else(|poison| poison.into_inner());
128    guard.processing = false;
129
130    match task_result {
131        Ok(()) => {
132            let end_from_metadata: Option<CuTime> = output_ref.metadata.process_time.end.into();
133            let end_time = end_from_metadata.unwrap_or_else(|| {
134                output_ref.metadata.process_time.end = fallback_end.into();
135                fallback_end
136            });
137            guard.ready_at = Some(end_time);
138        }
139        Err(error) => {
140            guard.ready_at = None;
141            guard.last_error = Some(error);
142        }
143    }
144}
145
146#[derive(Reflect)]
147#[reflect(no_field_bounds, from_reflect = false, type_path = false)]
148pub struct CuAsyncTask<T, O>
149where
150    T: for<'m> CuTask<Output<'m> = CuMsg<O>> + Send + 'static,
151    O: CuMsgPayload + Send + 'static,
152{
153    #[reflect(ignore)]
154    task: Arc<Mutex<T>>,
155    #[reflect(ignore)]
156    output: Arc<Mutex<CuMsg<O>>>,
157    #[reflect(ignore)]
158    state: Arc<Mutex<AsyncState>>,
159    #[reflect(ignore)]
160    tp: Arc<ThreadPool>,
161}
162
163impl<T, O> TypePath for CuAsyncTask<T, O>
164where
165    T: for<'m> CuTask<Output<'m> = CuMsg<O>> + Send + 'static,
166    O: CuMsgPayload + Send + 'static,
167{
168    fn type_path() -> &'static str {
169        "cu29_runtime::cuasynctask::CuAsyncTask"
170    }
171
172    fn short_type_path() -> &'static str {
173        "CuAsyncTask"
174    }
175
176    fn type_ident() -> Option<&'static str> {
177        Some("CuAsyncTask")
178    }
179
180    fn crate_name() -> Option<&'static str> {
181        Some("cu29_runtime")
182    }
183
184    fn module_path() -> Option<&'static str> {
185        Some("cuasynctask")
186    }
187}
188
189/// Resource bundle required by a backgrounded task.
190pub struct CuAsyncTaskResources<'r, T: CuTask> {
191    pub inner: T::Resources<'r>,
192    pub threadpool: Arc<ThreadPool>,
193}
194
195impl<T, O> CuAsyncTask<T, O>
196where
197    T: for<'m> CuTask<Output<'m> = CuMsg<O>> + Send + 'static,
198    O: CuMsgPayload + Send + 'static,
199{
200    #[allow(unused)]
201    pub fn new(
202        config: Option<&ComponentConfig>,
203        resources: T::Resources<'_>,
204        tp: Arc<ThreadPool>,
205    ) -> CuResult<Self> {
206        let task = Arc::new(Mutex::new(T::new(config, resources)?));
207        let output = Arc::new(Mutex::new(CuMsg::default()));
208        Ok(Self {
209            task,
210            output,
211            state: Arc::new(Mutex::new(AsyncState {
212                processing: false,
213                ready_at: None,
214                last_error: None,
215            })),
216            tp,
217        })
218    }
219}
220
221impl<T, O> Freezable for CuAsyncTask<T, O>
222where
223    T: for<'m> CuTask<Output<'m> = CuMsg<O>> + Send + 'static,
224    O: CuMsgPayload + Send + 'static,
225{
226    fn freeze<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
227        let state = self
228            .state
229            .lock()
230            .map_err(|_| EncodeError::OtherString("async task state mutex poisoned".to_string()))?;
231        encode_async_state(&state, encoder)?;
232
233        let task = self
234            .task
235            .lock()
236            .map_err(|_| EncodeError::OtherString("async task mutex poisoned".to_string()))?;
237        task.freeze(encoder)?;
238
239        let output = self.output.lock().map_err(|_| {
240            EncodeError::OtherString("async task output mutex poisoned".to_string())
241        })?;
242        encode_buffered_output(&output, encoder)?;
243        Ok(())
244    }
245
246    fn thaw<D: Decoder>(&mut self, decoder: &mut D) -> Result<(), DecodeError> {
247        let mut state = self
248            .state
249            .lock()
250            .map_err(|_| DecodeError::OtherString("async task state mutex poisoned".to_string()))?;
251        decode_async_state(&mut state, decoder)?;
252
253        let mut task = self
254            .task
255            .lock()
256            .map_err(|_| DecodeError::OtherString("async task mutex poisoned".to_string()))?;
257        task.thaw(decoder)?;
258
259        let mut output = self.output.lock().map_err(|_| {
260            DecodeError::OtherString("async task output mutex poisoned".to_string())
261        })?;
262        *output = decode_buffered_output(decoder)?;
263        Ok(())
264    }
265}
266
267impl<T, I, O> CuTask for CuAsyncTask<T, O>
268where
269    T: for<'i, 'o> CuTask<Input<'i> = CuMsg<I>, Output<'o> = CuMsg<O>> + Send + 'static,
270    I: CuMsgPayload + Send + Sync + 'static,
271    O: CuMsgPayload + Send + 'static,
272{
273    type Resources<'r> = CuAsyncTaskResources<'r, T>;
274    type Input<'m> = T::Input<'m>;
275    type Output<'m> = T::Output<'m>;
276
277    fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
278    where
279        Self: Sized,
280    {
281        CuAsyncTask::new(config, resources.inner, resources.threadpool)
282    }
283
284    fn start(&mut self, ctx: &CuContext) -> CuResult<()> {
285        let mut task = self
286            .task
287            .lock()
288            .map_err(|_| CuError::from("Async task mutex poisoned during start"))?;
289        task.start(ctx)
290    }
291
292    fn process<'i, 'o>(
293        &mut self,
294        ctx: &CuContext,
295        input: &Self::Input<'i>,
296        real_output: &mut Self::Output<'o>,
297    ) -> CuResult<()> {
298        if !begin_background_poll(ctx, &self.state, &self.output, real_output)? {
299            return Ok(());
300        }
301
302        // immediately requeue a task based on the new input
303        self.tp.spawn_fifo({
304            let ctx = ctx.clone();
305            let input = (*input).clone();
306            let output = self.output.clone();
307            let task = self.task.clone();
308            let state = self.state.clone();
309            move || {
310                let input_ref: &CuMsg<I> = &input;
311                let mut output_guard = match output.lock() {
312                    Ok(guard) => guard,
313                    Err(_) => {
314                        record_async_error(
315                            &state,
316                            CuError::from("Async task output mutex poisoned"),
317                        );
318                        return;
319                    }
320                };
321                let output_ref: &mut CuMsg<O> = &mut output_guard;
322
323                // Each async run starts from an empty output so a task that
324                // chooses not to publish does not leak the previous payload.
325                *output_ref = CuMsg::default();
326
327                // Track the actual processing interval so replay can honor it.
328                if output_ref.metadata.process_time.start.is_none() {
329                    output_ref.metadata.process_time.start = ctx.now().into();
330                }
331                let task_result = match task.lock() {
332                    Ok(mut task_guard) => task_guard.process(&ctx, input_ref, output_ref),
333                    Err(poison) => Err(CuError::from(format!(
334                        "Async task mutex poisoned: {poison}"
335                    ))),
336                };
337                finalize_background_run(&state, output_ref, ctx.now(), task_result);
338            }
339        });
340        Ok(())
341    }
342
343    fn stop(&mut self, ctx: &CuContext) -> CuResult<()> {
344        let mut task = self
345            .task
346            .lock()
347            .map_err(|_| CuError::from("Async task mutex poisoned during stop"))?;
348        task.stop(ctx)
349    }
350}
351
352#[derive(Reflect)]
353#[reflect(no_field_bounds, from_reflect = false, type_path = false)]
354pub struct CuAsyncSrcTask<T, O>
355where
356    T: for<'m> CuSrcTask<Output<'m> = CuMsg<O>> + Send + 'static,
357    O: CuMsgPayload + Send + 'static,
358{
359    #[reflect(ignore)]
360    task: Arc<Mutex<T>>,
361    #[reflect(ignore)]
362    output: Arc<Mutex<CuMsg<O>>>,
363    #[reflect(ignore)]
364    state: Arc<Mutex<AsyncState>>,
365    #[reflect(ignore)]
366    tp: Arc<ThreadPool>,
367}
368
369impl<T, O> TypePath for CuAsyncSrcTask<T, O>
370where
371    T: for<'m> CuSrcTask<Output<'m> = CuMsg<O>> + Send + 'static,
372    O: CuMsgPayload + Send + 'static,
373{
374    fn type_path() -> &'static str {
375        "cu29_runtime::cuasynctask::CuAsyncSrcTask"
376    }
377
378    fn short_type_path() -> &'static str {
379        "CuAsyncSrcTask"
380    }
381
382    fn type_ident() -> Option<&'static str> {
383        Some("CuAsyncSrcTask")
384    }
385
386    fn crate_name() -> Option<&'static str> {
387        Some("cu29_runtime")
388    }
389
390    fn module_path() -> Option<&'static str> {
391        Some("cuasynctask")
392    }
393}
394
395/// Resource bundle required by a backgrounded source.
396pub struct CuAsyncSrcTaskResources<'r, T: CuSrcTask> {
397    pub inner: T::Resources<'r>,
398    pub threadpool: Arc<ThreadPool>,
399}
400
401impl<T, O> CuAsyncSrcTask<T, O>
402where
403    T: for<'m> CuSrcTask<Output<'m> = CuMsg<O>> + Send + 'static,
404    O: CuMsgPayload + Send + 'static,
405{
406    #[allow(unused)]
407    pub fn new(
408        config: Option<&ComponentConfig>,
409        resources: T::Resources<'_>,
410        tp: Arc<ThreadPool>,
411    ) -> CuResult<Self> {
412        let task = Arc::new(Mutex::new(T::new(config, resources)?));
413        let output = Arc::new(Mutex::new(CuMsg::default()));
414        Ok(Self {
415            task,
416            output,
417            state: Arc::new(Mutex::new(AsyncState {
418                processing: false,
419                ready_at: None,
420                last_error: None,
421            })),
422            tp,
423        })
424    }
425}
426
427impl<T, O> Freezable for CuAsyncSrcTask<T, O>
428where
429    T: for<'m> CuSrcTask<Output<'m> = CuMsg<O>> + Send + 'static,
430    O: CuMsgPayload + Send + 'static,
431{
432    fn freeze<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
433        let state = self.state.lock().map_err(|_| {
434            EncodeError::OtherString("async source state mutex poisoned".to_string())
435        })?;
436        encode_async_state(&state, encoder)?;
437
438        let task = self
439            .task
440            .lock()
441            .map_err(|_| EncodeError::OtherString("async source mutex poisoned".to_string()))?;
442        task.freeze(encoder)?;
443
444        let output = self.output.lock().map_err(|_| {
445            EncodeError::OtherString("async source output mutex poisoned".to_string())
446        })?;
447        encode_buffered_output(&output, encoder)?;
448        Ok(())
449    }
450
451    fn thaw<D: Decoder>(&mut self, decoder: &mut D) -> Result<(), DecodeError> {
452        let mut state = self.state.lock().map_err(|_| {
453            DecodeError::OtherString("async source state mutex poisoned".to_string())
454        })?;
455        decode_async_state(&mut state, decoder)?;
456
457        let mut task = self
458            .task
459            .lock()
460            .map_err(|_| DecodeError::OtherString("async source mutex poisoned".to_string()))?;
461        task.thaw(decoder)?;
462
463        let mut output = self.output.lock().map_err(|_| {
464            DecodeError::OtherString("async source output mutex poisoned".to_string())
465        })?;
466        *output = decode_buffered_output(decoder)?;
467        Ok(())
468    }
469}
470
471impl<T, O> CuSrcTask for CuAsyncSrcTask<T, O>
472where
473    T: for<'m> CuSrcTask<Output<'m> = CuMsg<O>> + Send + 'static,
474    O: CuMsgPayload + Send + 'static,
475{
476    type Resources<'r> = CuAsyncSrcTaskResources<'r, T>;
477    type Output<'m> = T::Output<'m>;
478
479    fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
480    where
481        Self: Sized,
482    {
483        CuAsyncSrcTask::new(config, resources.inner, resources.threadpool)
484    }
485
486    fn start(&mut self, ctx: &CuContext) -> CuResult<()> {
487        let mut task = self
488            .task
489            .lock()
490            .map_err(|_| CuError::from("Async source mutex poisoned during start"))?;
491        task.start(ctx)
492    }
493
494    fn process<'o>(&mut self, ctx: &CuContext, real_output: &mut Self::Output<'o>) -> CuResult<()> {
495        if !begin_background_poll(ctx, &self.state, &self.output, real_output)? {
496            return Ok(());
497        }
498
499        self.tp.spawn_fifo({
500            let ctx = ctx.clone();
501            let output = self.output.clone();
502            let task = self.task.clone();
503            let state = self.state.clone();
504            move || {
505                let mut output_guard = match output.lock() {
506                    Ok(guard) => guard,
507                    Err(_) => {
508                        record_async_error(
509                            &state,
510                            CuError::from("Async task output mutex poisoned"),
511                        );
512                        return;
513                    }
514                };
515                let output_ref: &mut CuMsg<O> = &mut output_guard;
516
517                *output_ref = CuMsg::default();
518
519                if output_ref.metadata.process_time.start.is_none() {
520                    output_ref.metadata.process_time.start = ctx.now().into();
521                }
522                let task_result = match task.lock() {
523                    Ok(mut task_guard) => task_guard.process(&ctx, output_ref),
524                    Err(poison) => Err(CuError::from(format!(
525                        "Async source mutex poisoned: {poison}"
526                    ))),
527                };
528                finalize_background_run(&state, output_ref, ctx.now(), task_result);
529            }
530        });
531        Ok(())
532    }
533
534    fn stop(&mut self, ctx: &CuContext) -> CuResult<()> {
535        let mut task = self
536            .task
537            .lock()
538            .map_err(|_| CuError::from("Async source mutex poisoned during stop"))?;
539        task.stop(ctx)
540    }
541}
542
543#[cfg(test)]
544mod tests {
545    use super::*;
546    use crate::config::ComponentConfig;
547    use crate::cutask::CuMsg;
548    use crate::cutask::Freezable;
549    use crate::cutask_anytime::{
550        AnytimePolicy, AnytimeStatus, CuAnytimeRunner, CuAnytimeTask, Quality, quality_from_f32,
551    };
552    use crate::input_msg;
553    use crate::output_msg;
554    use cu29_clock::CuDuration;
555    use cu29_traits::CuResult;
556    use rayon::ThreadPoolBuilder;
557    use std::borrow::BorrowMut;
558    use std::sync::OnceLock;
559    use std::sync::mpsc;
560    use std::time::Duration;
561
562    static READY_RX: OnceLock<Arc<Mutex<mpsc::Receiver<CuTime>>>> = OnceLock::new();
563    static DONE_TX: OnceLock<mpsc::Sender<()>> = OnceLock::new();
564    #[derive(Reflect)]
565    struct TestTask {}
566
567    impl Freezable for TestTask {}
568
569    impl CuTask for TestTask {
570        type Resources<'r> = ();
571        type Input<'m> = input_msg!(u32);
572        type Output<'m> = output_msg!(u32);
573
574        fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
575        where
576            Self: Sized,
577        {
578            Ok(Self {})
579        }
580
581        fn process(
582            &mut self,
583            _ctx: &CuContext,
584            input: &Self::Input<'_>,
585            output: &mut Self::Output<'_>,
586        ) -> CuResult<()> {
587            output.borrow_mut().set_payload(*input.payload().unwrap());
588            Ok(())
589        }
590    }
591
592    #[test]
593    fn test_lifecycle() {
594        let tp = Arc::new(
595            rayon::ThreadPoolBuilder::new()
596                .num_threads(1)
597                .build()
598                .unwrap(),
599        );
600
601        let config = ComponentConfig::default();
602        let context = CuContext::new_with_clock();
603        let mut async_task: CuAsyncTask<TestTask, u32> =
604            CuAsyncTask::new(Some(&config), (), tp).unwrap();
605        let input = CuMsg::new(Some(42u32));
606        let mut output = CuMsg::new(None);
607
608        loop {
609            {
610                let output_ref: &mut CuMsg<u32> = &mut output;
611                async_task.process(&context, &input, output_ref).unwrap();
612            }
613
614            if let Some(val) = output.payload() {
615                assert_eq!(*val, 42u32);
616                break;
617            }
618        }
619    }
620
621    #[derive(Reflect)]
622    struct ControlledTask;
623
624    impl Freezable for ControlledTask {}
625
626    impl CuTask for ControlledTask {
627        type Resources<'r> = ();
628        type Input<'m> = input_msg!(u32);
629        type Output<'m> = output_msg!(u32);
630
631        fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
632        where
633            Self: Sized,
634        {
635            Ok(Self {})
636        }
637
638        fn process(
639            &mut self,
640            ctx: &CuContext,
641            _input: &Self::Input<'_>,
642            output: &mut Self::Output<'_>,
643        ) -> CuResult<()> {
644            let rx = READY_RX
645                .get()
646                .expect("ready channel not set")
647                .lock()
648                .unwrap();
649            let ready_time = rx
650                .recv_timeout(Duration::from_secs(1))
651                .expect("timed out waiting for ready signal");
652
653            output.set_payload(ready_time.as_nanos() as u32);
654            output.metadata.process_time.start = ctx.now().into();
655            output.metadata.process_time.end = ready_time.into();
656
657            if let Some(done_tx) = DONE_TX.get() {
658                let _ = done_tx.send(());
659            }
660            Ok(())
661        }
662    }
663
664    fn wait_until_async_idle<T, O>(async_task: &CuAsyncTask<T, O>)
665    where
666        T: for<'m> CuTask<Output<'m> = CuMsg<O>> + Send + 'static,
667        O: CuMsgPayload + Send + 'static,
668    {
669        for _ in 0..100 {
670            let state = async_task.state.lock().unwrap();
671            if !state.processing {
672                return;
673            }
674            drop(state);
675            std::thread::sleep(Duration::from_millis(1));
676        }
677        panic!("background task never became idle");
678    }
679
680    fn wait_until_async_src_idle<T, O>(async_task: &CuAsyncSrcTask<T, O>)
681    where
682        T: for<'m> CuSrcTask<Output<'m> = CuMsg<O>> + Send + 'static,
683        O: CuMsgPayload + Send + 'static,
684    {
685        for _ in 0..100 {
686            let state = async_task.state.lock().unwrap();
687            if !state.processing {
688                return;
689            }
690            drop(state);
691            std::thread::sleep(Duration::from_millis(1));
692        }
693        panic!("background source never became idle");
694    }
695
696    #[derive(Clone)]
697    struct ActionTaskResources {
698        actions: Arc<Mutex<mpsc::Receiver<Option<u32>>>>,
699        done: mpsc::Sender<()>,
700    }
701
702    #[derive(Reflect)]
703    #[reflect(no_field_bounds, from_reflect = false)]
704    struct ActionTask {
705        #[reflect(ignore)]
706        actions: Arc<Mutex<mpsc::Receiver<Option<u32>>>>,
707        #[reflect(ignore)]
708        done: mpsc::Sender<()>,
709    }
710
711    impl Freezable for ActionTask {}
712
713    impl CuTask for ActionTask {
714        type Resources<'r> = ActionTaskResources;
715        type Input<'m> = input_msg!(u32);
716        type Output<'m> = output_msg!(u32);
717
718        fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
719        where
720            Self: Sized,
721        {
722            let _ = config;
723            Ok(Self {
724                actions: resources.actions,
725                done: resources.done,
726            })
727        }
728
729        fn process(
730            &mut self,
731            _ctx: &CuContext,
732            _input: &Self::Input<'_>,
733            output: &mut Self::Output<'_>,
734        ) -> CuResult<()> {
735            let action = self
736                .actions
737                .lock()
738                .unwrap()
739                .recv_timeout(Duration::from_secs(1))
740                .expect("timed out waiting for action");
741            if let Some(value) = action {
742                output.set_payload(value);
743            }
744            let _ = self.done.send(());
745            Ok(())
746        }
747    }
748
749    #[derive(Reflect)]
750    #[reflect(no_field_bounds, from_reflect = false)]
751    struct ActionSrc {
752        #[reflect(ignore)]
753        actions: Arc<Mutex<mpsc::Receiver<Option<u32>>>>,
754        #[reflect(ignore)]
755        done: mpsc::Sender<()>,
756    }
757
758    impl Freezable for ActionSrc {}
759
760    impl CuSrcTask for ActionSrc {
761        type Resources<'r> = ActionTaskResources;
762        type Output<'m> = output_msg!(u32);
763
764        fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
765        where
766            Self: Sized,
767        {
768            let _ = config;
769            Ok(Self {
770                actions: resources.actions,
771                done: resources.done,
772            })
773        }
774
775        fn process(&mut self, _ctx: &CuContext, output: &mut Self::Output<'_>) -> CuResult<()> {
776            let action = self
777                .actions
778                .lock()
779                .unwrap()
780                .recv_timeout(Duration::from_secs(1))
781                .expect("timed out waiting for source action");
782            if let Some(value) = action {
783                output.set_payload(value);
784            }
785            let _ = self.done.send(());
786            Ok(())
787        }
788    }
789
790    #[derive(Clone)]
791    struct ControlledSrcResources {
792        ready_times: Arc<Mutex<mpsc::Receiver<CuTime>>>,
793        done: mpsc::Sender<()>,
794    }
795
796    #[derive(Reflect)]
797    #[reflect(no_field_bounds, from_reflect = false)]
798    struct ControlledSrc {
799        #[reflect(ignore)]
800        ready_times: Arc<Mutex<mpsc::Receiver<CuTime>>>,
801        #[reflect(ignore)]
802        done: mpsc::Sender<()>,
803    }
804
805    impl Freezable for ControlledSrc {}
806
807    impl CuSrcTask for ControlledSrc {
808        type Resources<'r> = ControlledSrcResources;
809        type Output<'m> = output_msg!(u32);
810
811        fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
812        where
813            Self: Sized,
814        {
815            let _ = config;
816            Ok(Self {
817                ready_times: resources.ready_times,
818                done: resources.done,
819            })
820        }
821
822        fn process(&mut self, ctx: &CuContext, output: &mut Self::Output<'_>) -> CuResult<()> {
823            let ready_time = self
824                .ready_times
825                .lock()
826                .unwrap()
827                .recv_timeout(Duration::from_secs(1))
828                .expect("timed out waiting for ready signal");
829            output.set_payload(ready_time.as_nanos() as u32);
830            output.metadata.process_time.start = ctx.now().into();
831            output.metadata.process_time.end = ready_time.into();
832            let _ = self.done.send(());
833            Ok(())
834        }
835    }
836
837    #[test]
838    fn background_clears_output_while_processing() {
839        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
840        let context = CuContext::new_with_clock();
841        let (action_tx, action_rx) = mpsc::channel::<Option<u32>>();
842        let (done_tx, done_rx) = mpsc::channel::<()>();
843        let resources = ActionTaskResources {
844            actions: Arc::new(Mutex::new(action_rx)),
845            done: done_tx,
846        };
847
848        let mut async_task: CuAsyncTask<ActionTask, u32> =
849            CuAsyncTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
850        let input = CuMsg::new(Some(1u32));
851        let mut output = CuMsg::new(None);
852
853        async_task.process(&context, &input, &mut output).unwrap();
854        assert!(output.payload().is_none());
855
856        output.set_payload(999);
857        async_task.process(&context, &input, &mut output).unwrap();
858        assert!(
859            output.payload().is_none(),
860            "background poll should clear stale output while the worker is still running"
861        );
862
863        action_tx.send(Some(7)).unwrap();
864        done_rx
865            .recv_timeout(Duration::from_secs(1))
866            .expect("background worker never finished");
867    }
868
869    #[test]
870    fn background_empty_run_does_not_reemit_previous_payload() {
871        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
872        let context = CuContext::new_with_clock();
873        let (action_tx, action_rx) = mpsc::channel::<Option<u32>>();
874        let (done_tx, done_rx) = mpsc::channel::<()>();
875        let resources = ActionTaskResources {
876            actions: Arc::new(Mutex::new(action_rx)),
877            done: done_tx,
878        };
879
880        let mut async_task: CuAsyncTask<ActionTask, u32> =
881            CuAsyncTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
882        let some_input = CuMsg::new(Some(1u32));
883        let no_input = CuMsg::new(None::<u32>);
884        let mut output = CuMsg::new(None);
885
886        action_tx.send(Some(42)).unwrap();
887        async_task
888            .process(&context, &some_input, &mut output)
889            .expect("failed to start first background run");
890        done_rx
891            .recv_timeout(Duration::from_secs(1))
892            .expect("first background run never finished");
893        wait_until_async_idle(&async_task);
894
895        action_tx.send(None).unwrap();
896        async_task
897            .process(&context, &no_input, &mut output)
898            .expect("failed to start empty background run");
899        assert_eq!(output.payload(), Some(&42));
900        done_rx
901            .recv_timeout(Duration::from_secs(1))
902            .expect("empty background run never finished");
903        wait_until_async_idle(&async_task);
904
905        action_tx.send(None).unwrap();
906        async_task
907            .process(&context, &no_input, &mut output)
908            .expect("failed to poll after empty background run");
909        assert!(
910            output.payload().is_none(),
911            "background task re-emitted the previous payload after an empty run"
912        );
913        done_rx
914            .recv_timeout(Duration::from_secs(1))
915            .expect("cleanup background run never finished");
916    }
917
918    #[test]
919    fn background_source_clears_output_while_processing() {
920        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
921        let context = CuContext::new_with_clock();
922        let (action_tx, action_rx) = mpsc::channel::<Option<u32>>();
923        let (done_tx, done_rx) = mpsc::channel::<()>();
924        let resources = ActionTaskResources {
925            actions: Arc::new(Mutex::new(action_rx)),
926            done: done_tx,
927        };
928
929        let mut async_src: CuAsyncSrcTask<ActionSrc, u32> =
930            CuAsyncSrcTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
931        let mut output = CuMsg::new(None);
932
933        async_src.process(&context, &mut output).unwrap();
934        assert!(output.payload().is_none());
935
936        output.set_payload(999);
937        async_src.process(&context, &mut output).unwrap();
938        assert!(
939            output.payload().is_none(),
940            "background source poll should clear stale output while the worker is still running"
941        );
942
943        action_tx.send(Some(7)).unwrap();
944        done_rx
945            .recv_timeout(Duration::from_secs(1))
946            .expect("background source never finished");
947    }
948
949    #[test]
950    fn background_source_empty_run_does_not_reemit_previous_payload() {
951        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
952        let context = CuContext::new_with_clock();
953        let (action_tx, action_rx) = mpsc::channel::<Option<u32>>();
954        let (done_tx, done_rx) = mpsc::channel::<()>();
955        let resources = ActionTaskResources {
956            actions: Arc::new(Mutex::new(action_rx)),
957            done: done_tx,
958        };
959
960        let mut async_src: CuAsyncSrcTask<ActionSrc, u32> =
961            CuAsyncSrcTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
962        let mut output = CuMsg::new(None);
963
964        action_tx.send(Some(42)).unwrap();
965        async_src
966            .process(&context, &mut output)
967            .expect("failed to start first background source run");
968        done_rx
969            .recv_timeout(Duration::from_secs(1))
970            .expect("first background source run never finished");
971        wait_until_async_src_idle(&async_src);
972
973        action_tx.send(None).unwrap();
974        async_src
975            .process(&context, &mut output)
976            .expect("failed to start empty background source run");
977        assert_eq!(output.payload(), Some(&42));
978        done_rx
979            .recv_timeout(Duration::from_secs(1))
980            .expect("empty background source run never finished");
981        wait_until_async_src_idle(&async_src);
982
983        action_tx.send(None).unwrap();
984        async_src
985            .process(&context, &mut output)
986            .expect("failed to poll background source after empty run");
987        assert!(
988            output.payload().is_none(),
989            "background source re-emitted the previous payload after an empty run"
990        );
991        done_rx
992            .recv_timeout(Duration::from_secs(1))
993            .expect("cleanup background source run never finished");
994    }
995
996    #[test]
997    fn background_respects_recorded_ready_time() {
998        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
999        let (context, clock_mock) = CuContext::new_mock_clock();
1000
1001        // Install the control channels for the task.
1002        let (ready_tx, ready_rx) = mpsc::channel::<CuTime>();
1003        let (done_tx, done_rx) = mpsc::channel::<()>();
1004        READY_RX
1005            .set(Arc::new(Mutex::new(ready_rx)))
1006            .expect("ready channel already set");
1007        DONE_TX
1008            .set(done_tx)
1009            .expect("completion channel already set");
1010
1011        let mut async_task: CuAsyncTask<ControlledTask, u32> =
1012            CuAsyncTask::new(Some(&ComponentConfig::default()), (), tp.clone()).unwrap();
1013        let input = CuMsg::new(Some(1u32));
1014        let mut output = CuMsg::new(None);
1015
1016        // Copperlist 0: kick off processing, nothing ready yet.
1017        clock_mock.set_value(0);
1018        async_task.process(&context, &input, &mut output).unwrap();
1019        assert!(output.payload().is_none());
1020
1021        // Copperlist 1 at time 10: still running in the background.
1022        clock_mock.set_value(10);
1023        async_task.process(&context, &input, &mut output).unwrap();
1024        assert!(output.payload().is_none());
1025
1026        // The background thread finishes at time 30 (recorded in metadata).
1027        clock_mock.set_value(30);
1028        ready_tx.send(CuTime::from(30u64)).unwrap();
1029        done_rx
1030            .recv_timeout(Duration::from_secs(1))
1031            .expect("background task never finished");
1032        // Wait until the async wrapper has cleared its processing flag and captured ready_at.
1033        let mut ready_at_recorded = None;
1034        for _ in 0..100 {
1035            let state = async_task.state.lock().unwrap();
1036            if !state.processing {
1037                ready_at_recorded = state.ready_at;
1038                if ready_at_recorded.is_some() {
1039                    break;
1040                }
1041            }
1042            drop(state);
1043            std::thread::sleep(Duration::from_millis(1));
1044        }
1045        assert!(
1046            ready_at_recorded.is_some(),
1047            "background task finished without recording ready_at"
1048        );
1049
1050        // Replay earlier than the recorded end time: the output should be held back.
1051        clock_mock.set_value(20);
1052        async_task.process(&context, &input, &mut output).unwrap();
1053        assert!(
1054            output.payload().is_none(),
1055            "Output surfaced before recorded ready time"
1056        );
1057
1058        // Once the mock clock reaches the recorded end time, the result is released.
1059        clock_mock.set_value(30);
1060        async_task.process(&context, &input, &mut output).unwrap();
1061        assert_eq!(output.payload(), Some(&30u32));
1062
1063        // Allow the background worker spawned by the last poll to complete so the thread pool shuts down cleanly.
1064        ready_tx.send(CuTime::from(40u64)).unwrap();
1065        let _ = done_rx.recv_timeout(Duration::from_secs(1));
1066    }
1067
1068    /// Anytime task under the wrapper: one increment per quantum, quality
1069    /// climbing toward the input target.
1070    #[derive(Reflect)]
1071    struct IncrementalPlanner {
1072        target: u32,
1073        acc: u32,
1074    }
1075
1076    impl Freezable for IncrementalPlanner {}
1077
1078    impl CuAnytimeTask for IncrementalPlanner {
1079        type Input<'m> = input_msg!(u32);
1080        type Output<'m> = output_msg!(u32);
1081        type Resources<'r> = ();
1082        type Quality = Quality;
1083
1084        fn new(_config: Option<&ComponentConfig>, _resources: ()) -> CuResult<Self> {
1085            Ok(Self { target: 0, acc: 0 })
1086        }
1087
1088        fn base(
1089            &mut self,
1090            _ctx: &CuContext,
1091            input: &Self::Input<'_>,
1092            output: &mut Self::Output<'_>,
1093        ) -> CuResult<AnytimeStatus<Quality>> {
1094            self.target = input.payload().copied().ok_or("planner: no input")?;
1095            self.acc = 0;
1096            output.set_payload(self.acc);
1097            Ok(AnytimeStatus::Improved(quality_from_f32(0.0)))
1098        }
1099
1100        fn refine(
1101            &mut self,
1102            _ctx: &CuContext,
1103            output: &mut Self::Output<'_>,
1104        ) -> CuResult<AnytimeStatus<Quality>> {
1105            self.acc += 1;
1106            output.set_payload(self.acc);
1107            Ok(AnytimeStatus::Improved(quality_from_f32(
1108                self.acc as f32 / self.target as f32,
1109            )))
1110        }
1111    }
1112
1113    /// Mirrors codegen for `anytime: (max_refines: 3)` on a background node.
1114    struct ThreeQuantaPolicy;
1115    impl<Q: Copy + PartialOrd> AnytimePolicy<Q> for ThreeQuantaPolicy {
1116        const TIME_BUDGET: Option<CuDuration> = None;
1117        const MAX_AGE: Option<CuDuration> = None;
1118        const MAX_STALL: Option<u32> = None;
1119        const MAX_REFINES: Option<u32> = Some(3);
1120    }
1121
1122    #[test]
1123    fn background_anytime_job_lands_with_its_status_stamp() {
1124        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1125        let context = CuContext::new_with_clock();
1126        let mut task: CuAsyncTask<CuAnytimeRunner<IncrementalPlanner, ThreeQuantaPolicy>, u32> =
1127            CuAsyncTask::new(Some(&ComponentConfig::default()), (), tp).unwrap();
1128
1129        let input = CuMsg::new(Some(5u32));
1130        let mut output = CuMsg::new(None);
1131
1132        // Poll until the worker's job comes back through the buffered output.
1133        for _ in 0..1000 {
1134            task.process(&context, &input, &mut output).unwrap();
1135            if output.payload().is_some() {
1136                break;
1137            }
1138            std::thread::sleep(Duration::from_millis(1));
1139        }
1140
1141        // Three quanta of a job needing five: stopped by the quanta bound, and
1142        // the stamp the runner wrote survived the buffered-output copy.
1143        assert_eq!(output.payload(), Some(&3));
1144        assert_eq!(output.metadata.status_txt.0.as_str(), "any:3it q=0.60 max");
1145    }
1146
1147    #[test]
1148    fn background_source_respects_recorded_ready_time() {
1149        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1150        let (context, clock_mock) = CuContext::new_mock_clock();
1151        let (ready_tx, ready_rx) = mpsc::channel::<CuTime>();
1152        let (done_tx, done_rx) = mpsc::channel::<()>();
1153        let resources = ControlledSrcResources {
1154            ready_times: Arc::new(Mutex::new(ready_rx)),
1155            done: done_tx,
1156        };
1157
1158        let mut async_src: CuAsyncSrcTask<ControlledSrc, u32> =
1159            CuAsyncSrcTask::new(Some(&ComponentConfig::default()), resources, tp.clone()).unwrap();
1160        let mut output = CuMsg::new(None);
1161
1162        clock_mock.set_value(0);
1163        async_src.process(&context, &mut output).unwrap();
1164        assert!(output.payload().is_none());
1165
1166        clock_mock.set_value(10);
1167        async_src.process(&context, &mut output).unwrap();
1168        assert!(output.payload().is_none());
1169
1170        clock_mock.set_value(30);
1171        ready_tx.send(CuTime::from(30u64)).unwrap();
1172        done_rx
1173            .recv_timeout(Duration::from_secs(1))
1174            .expect("background source never finished");
1175
1176        let mut ready_at_recorded = None;
1177        for _ in 0..100 {
1178            let state = async_src.state.lock().unwrap();
1179            if !state.processing {
1180                ready_at_recorded = state.ready_at;
1181                if ready_at_recorded.is_some() {
1182                    break;
1183                }
1184            }
1185            drop(state);
1186            std::thread::sleep(Duration::from_millis(1));
1187        }
1188        assert!(
1189            ready_at_recorded.is_some(),
1190            "background source finished without recording ready_at"
1191        );
1192
1193        clock_mock.set_value(20);
1194        async_src.process(&context, &mut output).unwrap();
1195        assert!(
1196            output.payload().is_none(),
1197            "background source surfaced output before recorded ready time"
1198        );
1199
1200        clock_mock.set_value(30);
1201        async_src.process(&context, &mut output).unwrap();
1202        assert_eq!(output.payload(), Some(&30u32));
1203
1204        ready_tx.send(CuTime::from(40u64)).unwrap();
1205        let _ = done_rx.recv_timeout(Duration::from_secs(1));
1206    }
1207}
cu29_runtime/cuasynctask.rs

cu29_runtime/
cuasynctask.rs
