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::input_msg;
550    use crate::output_msg;
551    use cu29_traits::CuResult;
552    use rayon::ThreadPoolBuilder;
553    use std::borrow::BorrowMut;
554    use std::sync::OnceLock;
555    use std::sync::mpsc;
556    use std::time::Duration;
557
558    static READY_RX: OnceLock<Arc<Mutex<mpsc::Receiver<CuTime>>>> = OnceLock::new();
559    static DONE_TX: OnceLock<mpsc::Sender<()>> = OnceLock::new();
560    #[derive(Reflect)]
561    struct TestTask {}
562
563    impl Freezable for TestTask {}
564
565    impl CuTask for TestTask {
566        type Resources<'r> = ();
567        type Input<'m> = input_msg!(u32);
568        type Output<'m> = output_msg!(u32);
569
570        fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
571        where
572            Self: Sized,
573        {
574            Ok(Self {})
575        }
576
577        fn process(
578            &mut self,
579            _ctx: &CuContext,
580            input: &Self::Input<'_>,
581            output: &mut Self::Output<'_>,
582        ) -> CuResult<()> {
583            output.borrow_mut().set_payload(*input.payload().unwrap());
584            Ok(())
585        }
586    }
587
588    #[test]
589    fn test_lifecycle() {
590        let tp = Arc::new(
591            rayon::ThreadPoolBuilder::new()
592                .num_threads(1)
593                .build()
594                .unwrap(),
595        );
596
597        let config = ComponentConfig::default();
598        let context = CuContext::new_with_clock();
599        let mut async_task: CuAsyncTask<TestTask, u32> =
600            CuAsyncTask::new(Some(&config), (), tp).unwrap();
601        let input = CuMsg::new(Some(42u32));
602        let mut output = CuMsg::new(None);
603
604        loop {
605            {
606                let output_ref: &mut CuMsg<u32> = &mut output;
607                async_task.process(&context, &input, output_ref).unwrap();
608            }
609
610            if let Some(val) = output.payload() {
611                assert_eq!(*val, 42u32);
612                break;
613            }
614        }
615    }
616
617    #[derive(Reflect)]
618    struct ControlledTask;
619
620    impl Freezable for ControlledTask {}
621
622    impl CuTask for ControlledTask {
623        type Resources<'r> = ();
624        type Input<'m> = input_msg!(u32);
625        type Output<'m> = output_msg!(u32);
626
627        fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
628        where
629            Self: Sized,
630        {
631            Ok(Self {})
632        }
633
634        fn process(
635            &mut self,
636            ctx: &CuContext,
637            _input: &Self::Input<'_>,
638            output: &mut Self::Output<'_>,
639        ) -> CuResult<()> {
640            let rx = READY_RX
641                .get()
642                .expect("ready channel not set")
643                .lock()
644                .unwrap();
645            let ready_time = rx
646                .recv_timeout(Duration::from_secs(1))
647                .expect("timed out waiting for ready signal");
648
649            output.set_payload(ready_time.as_nanos() as u32);
650            output.metadata.process_time.start = ctx.now().into();
651            output.metadata.process_time.end = ready_time.into();
652
653            if let Some(done_tx) = DONE_TX.get() {
654                let _ = done_tx.send(());
655            }
656            Ok(())
657        }
658    }
659
660    fn wait_until_async_idle<T, O>(async_task: &CuAsyncTask<T, O>)
661    where
662        T: for<'m> CuTask<Output<'m> = CuMsg<O>> + Send + 'static,
663        O: CuMsgPayload + Send + 'static,
664    {
665        for _ in 0..100 {
666            let state = async_task.state.lock().unwrap();
667            if !state.processing {
668                return;
669            }
670            drop(state);
671            std::thread::sleep(Duration::from_millis(1));
672        }
673        panic!("background task never became idle");
674    }
675
676    fn wait_until_async_src_idle<T, O>(async_task: &CuAsyncSrcTask<T, O>)
677    where
678        T: for<'m> CuSrcTask<Output<'m> = CuMsg<O>> + Send + 'static,
679        O: CuMsgPayload + Send + 'static,
680    {
681        for _ in 0..100 {
682            let state = async_task.state.lock().unwrap();
683            if !state.processing {
684                return;
685            }
686            drop(state);
687            std::thread::sleep(Duration::from_millis(1));
688        }
689        panic!("background source never became idle");
690    }
691
692    #[derive(Clone)]
693    struct ActionTaskResources {
694        actions: Arc<Mutex<mpsc::Receiver<Option<u32>>>>,
695        done: mpsc::Sender<()>,
696    }
697
698    #[derive(Reflect)]
699    #[reflect(no_field_bounds, from_reflect = false)]
700    struct ActionTask {
701        #[reflect(ignore)]
702        actions: Arc<Mutex<mpsc::Receiver<Option<u32>>>>,
703        #[reflect(ignore)]
704        done: mpsc::Sender<()>,
705    }
706
707    impl Freezable for ActionTask {}
708
709    impl CuTask for ActionTask {
710        type Resources<'r> = ActionTaskResources;
711        type Input<'m> = input_msg!(u32);
712        type Output<'m> = output_msg!(u32);
713
714        fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
715        where
716            Self: Sized,
717        {
718            let _ = config;
719            Ok(Self {
720                actions: resources.actions,
721                done: resources.done,
722            })
723        }
724
725        fn process(
726            &mut self,
727            _ctx: &CuContext,
728            _input: &Self::Input<'_>,
729            output: &mut Self::Output<'_>,
730        ) -> CuResult<()> {
731            let action = self
732                .actions
733                .lock()
734                .unwrap()
735                .recv_timeout(Duration::from_secs(1))
736                .expect("timed out waiting for action");
737            if let Some(value) = action {
738                output.set_payload(value);
739            }
740            let _ = self.done.send(());
741            Ok(())
742        }
743    }
744
745    #[derive(Reflect)]
746    #[reflect(no_field_bounds, from_reflect = false)]
747    struct ActionSrc {
748        #[reflect(ignore)]
749        actions: Arc<Mutex<mpsc::Receiver<Option<u32>>>>,
750        #[reflect(ignore)]
751        done: mpsc::Sender<()>,
752    }
753
754    impl Freezable for ActionSrc {}
755
756    impl CuSrcTask for ActionSrc {
757        type Resources<'r> = ActionTaskResources;
758        type Output<'m> = output_msg!(u32);
759
760        fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
761        where
762            Self: Sized,
763        {
764            let _ = config;
765            Ok(Self {
766                actions: resources.actions,
767                done: resources.done,
768            })
769        }
770
771        fn process(&mut self, _ctx: &CuContext, output: &mut Self::Output<'_>) -> CuResult<()> {
772            let action = self
773                .actions
774                .lock()
775                .unwrap()
776                .recv_timeout(Duration::from_secs(1))
777                .expect("timed out waiting for source action");
778            if let Some(value) = action {
779                output.set_payload(value);
780            }
781            let _ = self.done.send(());
782            Ok(())
783        }
784    }
785
786    #[derive(Clone)]
787    struct ControlledSrcResources {
788        ready_times: Arc<Mutex<mpsc::Receiver<CuTime>>>,
789        done: mpsc::Sender<()>,
790    }
791
792    #[derive(Reflect)]
793    #[reflect(no_field_bounds, from_reflect = false)]
794    struct ControlledSrc {
795        #[reflect(ignore)]
796        ready_times: Arc<Mutex<mpsc::Receiver<CuTime>>>,
797        #[reflect(ignore)]
798        done: mpsc::Sender<()>,
799    }
800
801    impl Freezable for ControlledSrc {}
802
803    impl CuSrcTask for ControlledSrc {
804        type Resources<'r> = ControlledSrcResources;
805        type Output<'m> = output_msg!(u32);
806
807        fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
808        where
809            Self: Sized,
810        {
811            let _ = config;
812            Ok(Self {
813                ready_times: resources.ready_times,
814                done: resources.done,
815            })
816        }
817
818        fn process(&mut self, ctx: &CuContext, output: &mut Self::Output<'_>) -> CuResult<()> {
819            let ready_time = self
820                .ready_times
821                .lock()
822                .unwrap()
823                .recv_timeout(Duration::from_secs(1))
824                .expect("timed out waiting for ready signal");
825            output.set_payload(ready_time.as_nanos() as u32);
826            output.metadata.process_time.start = ctx.now().into();
827            output.metadata.process_time.end = ready_time.into();
828            let _ = self.done.send(());
829            Ok(())
830        }
831    }
832
833    #[test]
834    fn background_clears_output_while_processing() {
835        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
836        let context = CuContext::new_with_clock();
837        let (action_tx, action_rx) = mpsc::channel::<Option<u32>>();
838        let (done_tx, done_rx) = mpsc::channel::<()>();
839        let resources = ActionTaskResources {
840            actions: Arc::new(Mutex::new(action_rx)),
841            done: done_tx,
842        };
843
844        let mut async_task: CuAsyncTask<ActionTask, u32> =
845            CuAsyncTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
846        let input = CuMsg::new(Some(1u32));
847        let mut output = CuMsg::new(None);
848
849        async_task.process(&context, &input, &mut output).unwrap();
850        assert!(output.payload().is_none());
851
852        output.set_payload(999);
853        async_task.process(&context, &input, &mut output).unwrap();
854        assert!(
855            output.payload().is_none(),
856            "background poll should clear stale output while the worker is still running"
857        );
858
859        action_tx.send(Some(7)).unwrap();
860        done_rx
861            .recv_timeout(Duration::from_secs(1))
862            .expect("background worker never finished");
863    }
864
865    #[test]
866    fn background_empty_run_does_not_reemit_previous_payload() {
867        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
868        let context = CuContext::new_with_clock();
869        let (action_tx, action_rx) = mpsc::channel::<Option<u32>>();
870        let (done_tx, done_rx) = mpsc::channel::<()>();
871        let resources = ActionTaskResources {
872            actions: Arc::new(Mutex::new(action_rx)),
873            done: done_tx,
874        };
875
876        let mut async_task: CuAsyncTask<ActionTask, u32> =
877            CuAsyncTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
878        let some_input = CuMsg::new(Some(1u32));
879        let no_input = CuMsg::new(None::<u32>);
880        let mut output = CuMsg::new(None);
881
882        action_tx.send(Some(42)).unwrap();
883        async_task
884            .process(&context, &some_input, &mut output)
885            .expect("failed to start first background run");
886        done_rx
887            .recv_timeout(Duration::from_secs(1))
888            .expect("first background run never finished");
889        wait_until_async_idle(&async_task);
890
891        action_tx.send(None).unwrap();
892        async_task
893            .process(&context, &no_input, &mut output)
894            .expect("failed to start empty background run");
895        assert_eq!(output.payload(), Some(&42));
896        done_rx
897            .recv_timeout(Duration::from_secs(1))
898            .expect("empty background run never finished");
899        wait_until_async_idle(&async_task);
900
901        action_tx.send(None).unwrap();
902        async_task
903            .process(&context, &no_input, &mut output)
904            .expect("failed to poll after empty background run");
905        assert!(
906            output.payload().is_none(),
907            "background task re-emitted the previous payload after an empty run"
908        );
909        done_rx
910            .recv_timeout(Duration::from_secs(1))
911            .expect("cleanup background run never finished");
912    }
913
914    #[test]
915    fn background_source_clears_output_while_processing() {
916        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
917        let context = CuContext::new_with_clock();
918        let (action_tx, action_rx) = mpsc::channel::<Option<u32>>();
919        let (done_tx, done_rx) = mpsc::channel::<()>();
920        let resources = ActionTaskResources {
921            actions: Arc::new(Mutex::new(action_rx)),
922            done: done_tx,
923        };
924
925        let mut async_src: CuAsyncSrcTask<ActionSrc, u32> =
926            CuAsyncSrcTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
927        let mut output = CuMsg::new(None);
928
929        async_src.process(&context, &mut output).unwrap();
930        assert!(output.payload().is_none());
931
932        output.set_payload(999);
933        async_src.process(&context, &mut output).unwrap();
934        assert!(
935            output.payload().is_none(),
936            "background source poll should clear stale output while the worker is still running"
937        );
938
939        action_tx.send(Some(7)).unwrap();
940        done_rx
941            .recv_timeout(Duration::from_secs(1))
942            .expect("background source never finished");
943    }
944
945    #[test]
946    fn background_source_empty_run_does_not_reemit_previous_payload() {
947        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
948        let context = CuContext::new_with_clock();
949        let (action_tx, action_rx) = mpsc::channel::<Option<u32>>();
950        let (done_tx, done_rx) = mpsc::channel::<()>();
951        let resources = ActionTaskResources {
952            actions: Arc::new(Mutex::new(action_rx)),
953            done: done_tx,
954        };
955
956        let mut async_src: CuAsyncSrcTask<ActionSrc, u32> =
957            CuAsyncSrcTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
958        let mut output = CuMsg::new(None);
959
960        action_tx.send(Some(42)).unwrap();
961        async_src
962            .process(&context, &mut output)
963            .expect("failed to start first background source run");
964        done_rx
965            .recv_timeout(Duration::from_secs(1))
966            .expect("first background source run never finished");
967        wait_until_async_src_idle(&async_src);
968
969        action_tx.send(None).unwrap();
970        async_src
971            .process(&context, &mut output)
972            .expect("failed to start empty background source run");
973        assert_eq!(output.payload(), Some(&42));
974        done_rx
975            .recv_timeout(Duration::from_secs(1))
976            .expect("empty background source run never finished");
977        wait_until_async_src_idle(&async_src);
978
979        action_tx.send(None).unwrap();
980        async_src
981            .process(&context, &mut output)
982            .expect("failed to poll background source after empty run");
983        assert!(
984            output.payload().is_none(),
985            "background source re-emitted the previous payload after an empty run"
986        );
987        done_rx
988            .recv_timeout(Duration::from_secs(1))
989            .expect("cleanup background source run never finished");
990    }
991
992    #[test]
993    fn background_respects_recorded_ready_time() {
994        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
995        let (context, clock_mock) = CuContext::new_mock_clock();
996
997        // Install the control channels for the task.
998        let (ready_tx, ready_rx) = mpsc::channel::<CuTime>();
999        let (done_tx, done_rx) = mpsc::channel::<()>();
1000        READY_RX
1001            .set(Arc::new(Mutex::new(ready_rx)))
1002            .expect("ready channel already set");
1003        DONE_TX
1004            .set(done_tx)
1005            .expect("completion channel already set");
1006
1007        let mut async_task: CuAsyncTask<ControlledTask, u32> =
1008            CuAsyncTask::new(Some(&ComponentConfig::default()), (), tp.clone()).unwrap();
1009        let input = CuMsg::new(Some(1u32));
1010        let mut output = CuMsg::new(None);
1011
1012        // Copperlist 0: kick off processing, nothing ready yet.
1013        clock_mock.set_value(0);
1014        async_task.process(&context, &input, &mut output).unwrap();
1015        assert!(output.payload().is_none());
1016
1017        // Copperlist 1 at time 10: still running in the background.
1018        clock_mock.set_value(10);
1019        async_task.process(&context, &input, &mut output).unwrap();
1020        assert!(output.payload().is_none());
1021
1022        // The background thread finishes at time 30 (recorded in metadata).
1023        clock_mock.set_value(30);
1024        ready_tx.send(CuTime::from(30u64)).unwrap();
1025        done_rx
1026            .recv_timeout(Duration::from_secs(1))
1027            .expect("background task never finished");
1028        // Wait until the async wrapper has cleared its processing flag and captured ready_at.
1029        let mut ready_at_recorded = None;
1030        for _ in 0..100 {
1031            let state = async_task.state.lock().unwrap();
1032            if !state.processing {
1033                ready_at_recorded = state.ready_at;
1034                if ready_at_recorded.is_some() {
1035                    break;
1036                }
1037            }
1038            drop(state);
1039            std::thread::sleep(Duration::from_millis(1));
1040        }
1041        assert!(
1042            ready_at_recorded.is_some(),
1043            "background task finished without recording ready_at"
1044        );
1045
1046        // Replay earlier than the recorded end time: the output should be held back.
1047        clock_mock.set_value(20);
1048        async_task.process(&context, &input, &mut output).unwrap();
1049        assert!(
1050            output.payload().is_none(),
1051            "Output surfaced before recorded ready time"
1052        );
1053
1054        // Once the mock clock reaches the recorded end time, the result is released.
1055        clock_mock.set_value(30);
1056        async_task.process(&context, &input, &mut output).unwrap();
1057        assert_eq!(output.payload(), Some(&30u32));
1058
1059        // Allow the background worker spawned by the last poll to complete so the thread pool shuts down cleanly.
1060        ready_tx.send(CuTime::from(40u64)).unwrap();
1061        let _ = done_rx.recv_timeout(Duration::from_secs(1));
1062    }
1063
1064    #[test]
1065    fn background_source_respects_recorded_ready_time() {
1066        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1067        let (context, clock_mock) = CuContext::new_mock_clock();
1068        let (ready_tx, ready_rx) = mpsc::channel::<CuTime>();
1069        let (done_tx, done_rx) = mpsc::channel::<()>();
1070        let resources = ControlledSrcResources {
1071            ready_times: Arc::new(Mutex::new(ready_rx)),
1072            done: done_tx,
1073        };
1074
1075        let mut async_src: CuAsyncSrcTask<ControlledSrc, u32> =
1076            CuAsyncSrcTask::new(Some(&ComponentConfig::default()), resources, tp.clone()).unwrap();
1077        let mut output = CuMsg::new(None);
1078
1079        clock_mock.set_value(0);
1080        async_src.process(&context, &mut output).unwrap();
1081        assert!(output.payload().is_none());
1082
1083        clock_mock.set_value(10);
1084        async_src.process(&context, &mut output).unwrap();
1085        assert!(output.payload().is_none());
1086
1087        clock_mock.set_value(30);
1088        ready_tx.send(CuTime::from(30u64)).unwrap();
1089        done_rx
1090            .recv_timeout(Duration::from_secs(1))
1091            .expect("background source never finished");
1092
1093        let mut ready_at_recorded = None;
1094        for _ in 0..100 {
1095            let state = async_src.state.lock().unwrap();
1096            if !state.processing {
1097                ready_at_recorded = state.ready_at;
1098                if ready_at_recorded.is_some() {
1099                    break;
1100                }
1101            }
1102            drop(state);
1103            std::thread::sleep(Duration::from_millis(1));
1104        }
1105        assert!(
1106            ready_at_recorded.is_some(),
1107            "background source finished without recording ready_at"
1108        );
1109
1110        clock_mock.set_value(20);
1111        async_src.process(&context, &mut output).unwrap();
1112        assert!(
1113            output.payload().is_none(),
1114            "background source surfaced output before recorded ready time"
1115        );
1116
1117        clock_mock.set_value(30);
1118        async_src.process(&context, &mut output).unwrap();
1119        assert_eq!(output.payload(), Some(&30u32));
1120
1121        ready_tx.send(CuTime::from(40u64)).unwrap();
1122        let _ = done_rx.recv_timeout(Duration::from_secs(1));
1123    }
1124}
cu29_runtime/cuasynctask.rs

cu29_runtime/
cuasynctask.rs
