cu29_runtime/
cuasynctask.rs

1use crate::config::ComponentConfig;
2use crate::context::CuContext;
3use crate::cutask::{BincodeAdapter, CuMsg, CuMsgPayload, CuSrcTask, CuTask, Freezable};
4use crate::reflect::{Reflect, TypePath};
5use bincode::config::standard;
6use bincode::de::read::Reader;
7use bincode::de::{Decode, Decoder};
8use bincode::enc::write::Writer;
9use bincode::enc::{Encode, Encoder, EncoderImpl};
10use bincode::error::{DecodeError, EncodeError};
11use cu29_clock::CuTime;
12use cu29_traits::{CuError, CuResult};
13use rayon::ThreadPool;
14use std::any::Any;
15use std::cell::UnsafeCell;
16use std::sync::{Arc, Mutex};
17
18const ASYNC_IDLE_TAG: u8 = 0xA0;
19const ASYNC_WAITING_TAG: u8 = 0xA1;
20const ASYNC_FAILED_TAG: u8 = 0xA2;
21const ASYNC_PENDING_TAG: u8 = 0xA3;
22
23enum AsyncStatus {
24    Idle,
25    Waiting(CuTime),
26    Failed(String),
27    Running(u64),
28    ReplayPending(u64),
29}
30
31struct AsyncState<O: CuMsgPayload> {
32    status: AsyncStatus,
33    committed_task: Vec<u8>,
34    task_scratch: Vec<u8>,
35    committed_output: CuMsg<O>,
36}
37
38impl<O: CuMsgPayload> AsyncState<O> {
39    fn new() -> Self {
40        Self {
41            status: AsyncStatus::Idle,
42            committed_task: Vec::new(),
43            task_scratch: Vec::new(),
44            committed_output: CuMsg::default(),
45        }
46    }
47}
48
49fn commit_initial_snapshot<O: CuMsgPayload>(state: &mut AsyncState<O>, snapshot: Vec<u8>) {
50    let mut scratch = core::mem::replace(&mut state.committed_task, snapshot);
51    if scratch.capacity() < state.committed_task.len() {
52        scratch.reserve_exact(state.committed_task.len() - scratch.len());
53    }
54    state.task_scratch = scratch;
55}
56
57struct BufferWriter<'a>(&'a mut Vec<u8>);
58
59impl Writer for BufferWriter<'_> {
60    fn write(&mut self, bytes: &[u8]) -> Result<(), EncodeError> {
61        self.0.extend_from_slice(bytes);
62        Ok(())
63    }
64}
65
66fn encode_value_into(value: &impl Encode, buffer: &mut Vec<u8>) -> Result<(), EncodeError> {
67    buffer.clear();
68    let mut encoder = EncoderImpl::new(BufferWriter(buffer), standard());
69    value.encode(&mut encoder)
70}
71
72fn freeze_into(task: &impl Freezable, buffer: &mut Vec<u8>) -> Result<(), EncodeError> {
73    encode_value_into(&BincodeAdapter(task), buffer)
74}
75
76fn thaw_from(task: &mut impl Freezable, buffer: &[u8]) -> Result<(), DecodeError> {
77    let reader = bincode::de::read::SliceReader::new(buffer);
78    let mut decoder = bincode::de::DecoderImpl::new(reader, standard(), ());
79    task.thaw(&mut decoder)
80}
81
82struct DynWriter<'a>(&'a mut dyn Writer);
83
84impl Writer for DynWriter<'_> {
85    fn write(&mut self, bytes: &[u8]) -> Result<(), EncodeError> {
86        self.0.write(bytes)
87    }
88}
89
90struct DynReader<'a>(&'a mut dyn Reader);
91
92impl Reader for DynReader<'_> {
93    fn read(&mut self, bytes: &mut [u8]) -> Result<(), DecodeError> {
94        self.0.read(bytes)
95    }
96
97    fn peek_read(&mut self, length: usize) -> Option<&[u8]> {
98        self.0.peek_read(length)
99    }
100
101    fn consume(&mut self, length: usize) {
102        self.0.consume(length);
103    }
104}
105
106trait ErasedDispatch: Any + Send + Sync {
107    fn as_any(&self) -> &dyn Any;
108    fn encode_input(&self, writer: &mut dyn Writer) -> Result<(), EncodeError>;
109    fn decode_input(&self, reader: &mut dyn Reader) -> Result<(), DecodeError>;
110}
111
112struct DispatchSlot<I: CuMsgPayload> {
113    input: UnsafeCell<CuMsg<I>>,
114}
115
116// SAFETY: the slot is written only before a run is published or while thaw holds the
117// inner-task mutex. While a run is published, workers and keyframes only read it.
118unsafe impl<I: CuMsgPayload + Send + Sync> Sync for DispatchSlot<I> {}
119
120impl<I> DispatchSlot<I>
121where
122    I: CuMsgPayload + Send + Sync + 'static,
123{
124    fn new() -> Self {
125        Self {
126            input: UnsafeCell::new(CuMsg::default()),
127        }
128    }
129
130    fn replace(&self, input: CuMsg<I>) -> CuMsg<I> {
131        // SAFETY: callers write only while no worker can hold a reference to the slot.
132        unsafe { core::mem::replace(&mut *self.input.get(), input) }
133    }
134
135    fn get(&self) -> &CuMsg<I> {
136        // SAFETY: published dispatch input remains immutable until the worker finishes.
137        unsafe { &*self.input.get() }
138    }
139}
140
141impl<I> ErasedDispatch for DispatchSlot<I>
142where
143    I: CuMsgPayload + Send + Sync + 'static,
144{
145    fn as_any(&self) -> &dyn Any {
146        self
147    }
148
149    fn encode_input(&self, writer: &mut dyn Writer) -> Result<(), EncodeError> {
150        let mut encoder = EncoderImpl::new(DynWriter(writer), standard());
151        self.get().encode(&mut encoder)
152    }
153
154    fn decode_input(&self, reader: &mut dyn Reader) -> Result<(), DecodeError> {
155        let mut decoder = bincode::de::DecoderImpl::new(DynReader(reader), standard(), ());
156        drop(self.replace(CuMsg::<I>::decode(&mut decoder)?));
157        Ok(())
158    }
159}
160
161fn failure(error: CuError) -> AsyncStatus {
162    AsyncStatus::Failed(error.to_string())
163}
164
165fn encode_async_state<O, E>(state: &AsyncState<O>, encoder: &mut E) -> Result<bool, EncodeError>
166where
167    O: CuMsgPayload + Send + 'static,
168    E: Encoder,
169{
170    Encode::encode(&state.committed_task, encoder)?;
171    Encode::encode(&state.committed_output, encoder)?;
172    match &state.status {
173        AsyncStatus::Idle => {
174            ASYNC_IDLE_TAG.encode(encoder)?;
175            Ok(false)
176        }
177        AsyncStatus::Waiting(ready_at) => {
178            ASYNC_WAITING_TAG.encode(encoder)?;
179            ready_at.encode(encoder)?;
180            Ok(false)
181        }
182        AsyncStatus::Failed(snapshot) => {
183            ASYNC_FAILED_TAG.encode(encoder)?;
184            snapshot.encode(encoder)?;
185            Ok(false)
186        }
187        AsyncStatus::Running(cl_id) | AsyncStatus::ReplayPending(cl_id) => {
188            ASYNC_PENDING_TAG.encode(encoder)?;
189            cl_id.encode(encoder)?;
190            Ok(true)
191        }
192    }
193}
194
195fn decode_async_state<O, D>(decoder: &mut D) -> Result<AsyncState<O>, DecodeError>
196where
197    O: CuMsgPayload + Send + 'static,
198    D: Decoder,
199{
200    let committed_task = Decode::decode(decoder)?;
201    let committed_output = CuMsg::<O>::decode(&mut decoder.with_context(()))?;
202    let status = match u8::decode(decoder)? {
203        ASYNC_IDLE_TAG => AsyncStatus::Idle,
204        ASYNC_WAITING_TAG => AsyncStatus::Waiting(Decode::decode(decoder)?),
205        ASYNC_FAILED_TAG => {
206            let snapshot: String = Decode::decode(decoder)?;
207            AsyncStatus::Failed(snapshot)
208        }
209        ASYNC_PENDING_TAG => AsyncStatus::ReplayPending(u64::decode(decoder)?),
210        tag => {
211            return Err(DecodeError::OtherString(format!(
212                "unsupported async keyframe payload tag {tag:#04x}; expected version 1"
213            )));
214        }
215    };
216    Ok(AsyncState {
217        status,
218        committed_task,
219        task_scratch: Vec::new(),
220        committed_output,
221    })
222}
223
224type ErasedDispatchSlot = Arc<dyn ErasedDispatch>;
225
226fn dispatch_slot<I>(slot: &ErasedDispatchSlot) -> CuResult<&DispatchSlot<I>>
227where
228    I: CuMsgPayload + Send + Sync + 'static,
229{
230    slot.as_any()
231        .downcast_ref::<DispatchSlot<I>>()
232        .ok_or_else(|| CuError::from("Async task dispatch slot type did not match its input"))
233}
234
235fn record_async_error<O: CuMsgPayload>(state: &Mutex<AsyncState<O>>, error: CuError) {
236    let mut guard = match state.lock() {
237        Ok(guard) => guard,
238        Err(poison) => poison.into_inner(),
239    };
240    guard.status = failure(error);
241}
242
243#[derive(Reflect)]
244#[reflect(no_field_bounds, from_reflect = false, type_path = false)]
245pub struct CuAsyncTask<T, O>
246where
247    T: for<'m> CuTask<Output<'m> = CuMsg<O>> + Send + 'static,
248    O: CuMsgPayload + Send + 'static,
249{
250    #[reflect(ignore)]
251    task: Arc<Mutex<T>>,
252    #[reflect(ignore)]
253    state: Arc<Mutex<AsyncState<O>>>,
254    #[reflect(ignore)]
255    dispatch: Option<ErasedDispatchSlot>,
256    #[reflect(ignore)]
257    tp: Arc<ThreadPool>,
258}
259
260impl<T, O> TypePath for CuAsyncTask<T, O>
261where
262    T: for<'m> CuTask<Output<'m> = CuMsg<O>> + Send + 'static,
263    O: CuMsgPayload + Send + 'static,
264{
265    fn type_path() -> &'static str {
266        "cu29_runtime::cuasynctask::CuAsyncTask"
267    }
268
269    fn short_type_path() -> &'static str {
270        "CuAsyncTask"
271    }
272
273    fn type_ident() -> Option<&'static str> {
274        Some("CuAsyncTask")
275    }
276
277    fn crate_name() -> Option<&'static str> {
278        Some("cu29_runtime")
279    }
280
281    fn module_path() -> Option<&'static str> {
282        Some("cuasynctask")
283    }
284}
285
286/// Resource bundle required by a backgrounded task.
287pub struct CuAsyncTaskResources<'r, T: CuTask> {
288    pub inner: T::Resources<'r>,
289    pub threadpool: Arc<ThreadPool>,
290}
291
292impl<T, O> CuAsyncTask<T, O>
293where
294    T: for<'m> CuTask<Output<'m> = CuMsg<O>> + Send + 'static,
295    O: CuMsgPayload + Send + 'static,
296{
297    #[allow(unused)]
298    pub fn new(
299        config: Option<&ComponentConfig>,
300        resources: T::Resources<'_>,
301        tp: Arc<ThreadPool>,
302    ) -> CuResult<Self> {
303        let task = Arc::new(Mutex::new(T::new(config, resources)?));
304        Ok(Self {
305            task,
306            state: Arc::new(Mutex::new(AsyncState::new())),
307            dispatch: None,
308            tp,
309        })
310    }
311
312    fn initialize_dispatch<I>(&mut self) -> CuResult<()>
313    where
314        I: CuMsgPayload + Send + Sync + 'static,
315    {
316        if let Some(dispatch) = self.dispatch.as_ref() {
317            let _ = dispatch_slot::<I>(dispatch)?;
318        } else {
319            self.dispatch = Some(Arc::new(DispatchSlot::<I>::new()));
320        }
321        Ok(())
322    }
323}
324
325impl<T, O> Freezable for CuAsyncTask<T, O>
326where
327    T: for<'m> CuTask<Output<'m> = CuMsg<O>> + Send + 'static,
328    O: CuMsgPayload + Send + 'static,
329{
330    fn freeze<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
331        let state = self
332            .state
333            .lock()
334            .map_err(|_| EncodeError::OtherString("async task state mutex poisoned".to_string()))?;
335        let pending = encode_async_state(&state, encoder)?;
336        if pending {
337            self.dispatch
338                .as_ref()
339                .ok_or_else(|| {
340                    EncodeError::OtherString(
341                        "async task pending before dispatch initialization".to_string(),
342                    )
343                })?
344                .encode_input(encoder.writer())?;
345        }
346        Ok(())
347    }
348
349    fn thaw<D: Decoder>(&mut self, decoder: &mut D) -> Result<(), DecodeError> {
350        let mut task = self
351            .task
352            .lock()
353            .map_err(|_| DecodeError::OtherString("async task mutex poisoned".to_string()))?;
354        let restored_state = decode_async_state(decoder)?;
355        if matches!(restored_state.status, AsyncStatus::ReplayPending(_)) {
356            self.dispatch
357                .as_ref()
358                .ok_or_else(|| {
359                    DecodeError::OtherString(
360                        "async task restored before dispatch initialization".to_string(),
361                    )
362                })?
363                .decode_input(decoder.reader())?;
364        }
365        thaw_from(&mut *task, &restored_state.committed_task)?;
366
367        let mut state = self
368            .state
369            .lock()
370            .map_err(|_| DecodeError::OtherString("async task state mutex poisoned".to_string()))?;
371        *state = restored_state;
372        Ok(())
373    }
374}
375
376impl<T, I, O> CuTask for CuAsyncTask<T, O>
377where
378    T: for<'i, 'o> CuTask<Input<'i> = CuMsg<I>, Output<'o> = CuMsg<O>> + Send + 'static,
379    I: CuMsgPayload + Send + Sync + 'static,
380    O: CuMsgPayload + Send + 'static,
381{
382    type Resources<'r> = CuAsyncTaskResources<'r, T>;
383    type Input<'m> = T::Input<'m>;
384    type Output<'m> = T::Output<'m>;
385
386    fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
387    where
388        Self: Sized,
389    {
390        CuAsyncTask::new(config, resources.inner, resources.threadpool)
391    }
392
393    fn start(&mut self, ctx: &CuContext) -> CuResult<()> {
394        self.initialize_dispatch::<I>()?;
395        let mut task = self
396            .task
397            .lock()
398            .map_err(|_| CuError::from("Async task mutex poisoned during start"))?;
399        task.start(ctx)?;
400        let mut state = self
401            .state
402            .lock()
403            .map_err(|_| CuError::from("Async task state mutex poisoned during start"))?;
404        let mut snapshot = core::mem::take(&mut state.task_scratch);
405        freeze_into(&*task, &mut snapshot).map_err(|error| {
406            CuError::from("Failed to snapshot async task after start").with_cause(error)
407        })?;
408        commit_initial_snapshot(&mut state, snapshot);
409        Ok(())
410    }
411
412    fn process<'i, 'o>(
413        &mut self,
414        ctx: &CuContext,
415        input: &Self::Input<'i>,
416        real_output: &mut Self::Output<'o>,
417    ) -> CuResult<()> {
418        let (dispatch, dispatch_cl_id, mut task_snapshot, retired_input) = {
419            let mut state = self.state.lock().map_err(|_| {
420                CuError::from("Async task state mutex poisoned while scheduling background work")
421            })?;
422            if matches!(state.status, AsyncStatus::Failed(_)) {
423                let AsyncStatus::Failed(error) =
424                    core::mem::replace(&mut state.status, AsyncStatus::Idle)
425                else {
426                    unreachable!();
427                };
428                return Err(CuError::from(error));
429            }
430            if matches!(state.status, AsyncStatus::Running(_)) {
431                *real_output = CuMsg::default();
432                return Ok(());
433            }
434            if let AsyncStatus::Waiting(ready_at) = state.status
435                && ctx.now() < ready_at
436            {
437                *real_output = CuMsg::default();
438                return Ok(());
439            }
440
441            let dispatch = self
442                .dispatch
443                .as_ref()
444                .ok_or_else(|| CuError::from("Async task dispatch slot was not initialized"))?;
445            let typed_dispatch = dispatch_slot::<I>(dispatch)?;
446            let (dispatch_cl_id, replay_pending) =
447                if let AsyncStatus::ReplayPending(cl_id) = state.status {
448                    (cl_id, true)
449                } else {
450                    (ctx.cl_id(), false)
451                };
452            let retired_input = if replay_pending {
453                CuMsg::default()
454            } else {
455                typed_dispatch.replace((*input).clone())
456            };
457            *real_output = state.committed_output.clone();
458            state.status = AsyncStatus::Running(dispatch_cl_id);
459            (
460                dispatch.clone(),
461                dispatch_cl_id,
462                core::mem::take(&mut state.task_scratch),
463                retired_input,
464            )
465        };
466
467        self.tp.spawn_fifo({
468            let ctx = ctx.with_cl_id(dispatch_cl_id);
469            let task = self.task.clone();
470            let state = self.state.clone();
471            move || {
472                let mut worker_output = CuMsg::default();
473                let typed_dispatch =
474                    if let Some(slot) = dispatch.as_any().downcast_ref::<DispatchSlot<I>>() {
475                        slot
476                    } else {
477                        record_async_error(
478                            &state,
479                            CuError::from("Async task dispatch slot type did not match its input"),
480                        );
481                        return;
482                    };
483                let input_ref = typed_dispatch.get();
484                let mut task_guard = match task.lock() {
485                    Ok(guard) => guard,
486                    Err(poison) => {
487                        record_async_error(
488                            &state,
489                            CuError::from(format!("Async task mutex poisoned: {poison}")),
490                        );
491                        return;
492                    }
493                };
494                // Each async run starts from an empty output so a task that
495                // chooses not to publish does not leak the previous payload.
496                // Track the actual processing interval so replay can honor it.
497                if worker_output.metadata.process_time.start.is_none() {
498                    worker_output.metadata.process_time.start = ctx.now().into();
499                }
500                let task_result = task_guard.process(&ctx, input_ref, &mut worker_output);
501                let fallback_end = ctx.now();
502                let end_from_metadata: Option<CuTime> =
503                    worker_output.metadata.process_time.end.into();
504                let ready_at = end_from_metadata.unwrap_or_else(|| {
505                    worker_output.metadata.process_time.end = fallback_end.into();
506                    fallback_end
507                });
508                let snapshot_result = freeze_into(&*task_guard, &mut task_snapshot);
509                let (commit_snapshot, status) = match snapshot_result {
510                    Ok(()) => (
511                        true,
512                        match task_result {
513                            Ok(()) => AsyncStatus::Waiting(ready_at),
514                            Err(error) => failure(error),
515                        },
516                    ),
517                    Err(error) => (
518                        false,
519                        failure(
520                            CuError::from("Failed to snapshot completed async task")
521                                .with_cause(error),
522                        ),
523                    ),
524                };
525
526                let mut guard = state.lock().unwrap_or_else(|poison| poison.into_inner());
527                let retired_output = if commit_snapshot {
528                    guard.task_scratch =
529                        core::mem::replace(&mut guard.committed_task, task_snapshot);
530                    Some(core::mem::replace(
531                        &mut guard.committed_output,
532                        worker_output,
533                    ))
534                } else {
535                    guard.task_scratch = task_snapshot;
536                    None
537                };
538                guard.status = status;
539                drop(guard);
540                drop(task_guard);
541                drop(retired_output);
542                drop(retired_input);
543            }
544        });
545        Ok(())
546    }
547
548    fn stop(&mut self, ctx: &CuContext) -> CuResult<()> {
549        let mut task = self
550            .task
551            .lock()
552            .map_err(|_| CuError::from("Async task mutex poisoned during stop"))?;
553        task.stop(ctx)
554    }
555}
556
557#[derive(Reflect)]
558#[reflect(no_field_bounds, from_reflect = false, type_path = false)]
559pub struct CuAsyncSrcTask<T, O>
560where
561    T: for<'m> CuSrcTask<Output<'m> = CuMsg<O>> + Send + 'static,
562    O: CuMsgPayload + Send + 'static,
563{
564    #[reflect(ignore)]
565    task: Arc<Mutex<T>>,
566    #[reflect(ignore)]
567    state: Arc<Mutex<AsyncState<O>>>,
568    #[reflect(ignore)]
569    tp: Arc<ThreadPool>,
570}
571
572impl<T, O> TypePath for CuAsyncSrcTask<T, O>
573where
574    T: for<'m> CuSrcTask<Output<'m> = CuMsg<O>> + Send + 'static,
575    O: CuMsgPayload + Send + 'static,
576{
577    fn type_path() -> &'static str {
578        "cu29_runtime::cuasynctask::CuAsyncSrcTask"
579    }
580
581    fn short_type_path() -> &'static str {
582        "CuAsyncSrcTask"
583    }
584
585    fn type_ident() -> Option<&'static str> {
586        Some("CuAsyncSrcTask")
587    }
588
589    fn crate_name() -> Option<&'static str> {
590        Some("cu29_runtime")
591    }
592
593    fn module_path() -> Option<&'static str> {
594        Some("cuasynctask")
595    }
596}
597
598/// Resource bundle required by a backgrounded source.
599pub struct CuAsyncSrcTaskResources<'r, T: CuSrcTask> {
600    pub inner: T::Resources<'r>,
601    pub threadpool: Arc<ThreadPool>,
602}
603
604impl<T, O> CuAsyncSrcTask<T, O>
605where
606    T: for<'m> CuSrcTask<Output<'m> = CuMsg<O>> + Send + 'static,
607    O: CuMsgPayload + Send + 'static,
608{
609    #[allow(unused)]
610    pub fn new(
611        config: Option<&ComponentConfig>,
612        resources: T::Resources<'_>,
613        tp: Arc<ThreadPool>,
614    ) -> CuResult<Self> {
615        let task = Arc::new(Mutex::new(T::new(config, resources)?));
616        Ok(Self {
617            task,
618            state: Arc::new(Mutex::new(AsyncState::new())),
619            tp,
620        })
621    }
622}
623
624impl<T, O> Freezable for CuAsyncSrcTask<T, O>
625where
626    T: for<'m> CuSrcTask<Output<'m> = CuMsg<O>> + Send + 'static,
627    O: CuMsgPayload + Send + 'static,
628{
629    fn freeze<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
630        let state = self.state.lock().map_err(|_| {
631            EncodeError::OtherString("async source state mutex poisoned".to_string())
632        })?;
633        let _ = encode_async_state(&state, encoder)?;
634        Ok(())
635    }
636
637    fn thaw<D: Decoder>(&mut self, decoder: &mut D) -> Result<(), DecodeError> {
638        let mut task = self
639            .task
640            .lock()
641            .map_err(|_| DecodeError::OtherString("async source mutex poisoned".to_string()))?;
642        let restored_state = decode_async_state(decoder)?;
643        thaw_from(&mut *task, &restored_state.committed_task)?;
644
645        let mut state = self.state.lock().map_err(|_| {
646            DecodeError::OtherString("async source state mutex poisoned".to_string())
647        })?;
648        *state = restored_state;
649        Ok(())
650    }
651}
652
653impl<T, O> CuSrcTask for CuAsyncSrcTask<T, O>
654where
655    T: for<'m> CuSrcTask<Output<'m> = CuMsg<O>> + Send + 'static,
656    O: CuMsgPayload + Send + 'static,
657{
658    type Resources<'r> = CuAsyncSrcTaskResources<'r, T>;
659    type Output<'m> = T::Output<'m>;
660
661    fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
662    where
663        Self: Sized,
664    {
665        CuAsyncSrcTask::new(config, resources.inner, resources.threadpool)
666    }
667
668    fn start(&mut self, ctx: &CuContext) -> CuResult<()> {
669        let mut task = self
670            .task
671            .lock()
672            .map_err(|_| CuError::from("Async source mutex poisoned during start"))?;
673        task.start(ctx)?;
674        let mut state = self
675            .state
676            .lock()
677            .map_err(|_| CuError::from("Async source state mutex poisoned during start"))?;
678        let mut snapshot = core::mem::take(&mut state.task_scratch);
679        freeze_into(&*task, &mut snapshot).map_err(|error| {
680            CuError::from("Failed to snapshot async source after start").with_cause(error)
681        })?;
682        commit_initial_snapshot(&mut state, snapshot);
683        Ok(())
684    }
685
686    fn process<'o>(&mut self, ctx: &CuContext, real_output: &mut Self::Output<'o>) -> CuResult<()> {
687        let (dispatch_cl_id, mut task_snapshot) = {
688            let mut state = self.state.lock().map_err(|_| {
689                CuError::from("Async source state mutex poisoned while scheduling background work")
690            })?;
691            if matches!(state.status, AsyncStatus::Failed(_)) {
692                let AsyncStatus::Failed(error) =
693                    core::mem::replace(&mut state.status, AsyncStatus::Idle)
694                else {
695                    unreachable!();
696                };
697                return Err(CuError::from(error));
698            }
699            if matches!(state.status, AsyncStatus::Running(_)) {
700                *real_output = CuMsg::default();
701                return Ok(());
702            }
703            if let AsyncStatus::Waiting(ready_at) = state.status
704                && ctx.now() < ready_at
705            {
706                *real_output = CuMsg::default();
707                return Ok(());
708            }
709
710            let dispatch_cl_id = if let AsyncStatus::ReplayPending(cl_id) = state.status {
711                cl_id
712            } else {
713                ctx.cl_id()
714            };
715            *real_output = state.committed_output.clone();
716            state.status = AsyncStatus::Running(dispatch_cl_id);
717            (dispatch_cl_id, core::mem::take(&mut state.task_scratch))
718        };
719
720        self.tp.spawn_fifo({
721            let ctx = ctx.with_cl_id(dispatch_cl_id);
722            let task = self.task.clone();
723            let state = self.state.clone();
724            move || {
725                let mut worker_output = CuMsg::default();
726                let mut task_guard = match task.lock() {
727                    Ok(guard) => guard,
728                    Err(poison) => {
729                        record_async_error(
730                            &state,
731                            CuError::from(format!("Async source mutex poisoned: {poison}")),
732                        );
733                        return;
734                    }
735                };
736                if worker_output.metadata.process_time.start.is_none() {
737                    worker_output.metadata.process_time.start = ctx.now().into();
738                }
739                let task_result = task_guard.process(&ctx, &mut worker_output);
740                let fallback_end = ctx.now();
741                let end_from_metadata: Option<CuTime> =
742                    worker_output.metadata.process_time.end.into();
743                let ready_at = end_from_metadata.unwrap_or_else(|| {
744                    worker_output.metadata.process_time.end = fallback_end.into();
745                    fallback_end
746                });
747                let snapshot_result = freeze_into(&*task_guard, &mut task_snapshot);
748                let (commit_snapshot, status) = match snapshot_result {
749                    Ok(()) => (
750                        true,
751                        match task_result {
752                            Ok(()) => AsyncStatus::Waiting(ready_at),
753                            Err(error) => failure(error),
754                        },
755                    ),
756                    Err(error) => (
757                        false,
758                        failure(
759                            CuError::from("Failed to snapshot completed async source")
760                                .with_cause(error),
761                        ),
762                    ),
763                };
764
765                let mut guard = state.lock().unwrap_or_else(|poison| poison.into_inner());
766                let retired_output = if commit_snapshot {
767                    guard.task_scratch =
768                        core::mem::replace(&mut guard.committed_task, task_snapshot);
769                    Some(core::mem::replace(
770                        &mut guard.committed_output,
771                        worker_output,
772                    ))
773                } else {
774                    guard.task_scratch = task_snapshot;
775                    None
776                };
777                guard.status = status;
778                drop(guard);
779                drop(task_guard);
780                drop(retired_output);
781            }
782        });
783        Ok(())
784    }
785
786    fn stop(&mut self, ctx: &CuContext) -> CuResult<()> {
787        let mut task = self
788            .task
789            .lock()
790            .map_err(|_| CuError::from("Async source mutex poisoned during stop"))?;
791        task.stop(ctx)
792    }
793}
794
795#[cfg(test)]
796mod tests {
797    use super::*;
798    use crate::config::ComponentConfig;
799    use crate::cutask::CuMsg;
800    use crate::cutask::Freezable;
801    use crate::cutask_anytime::{
802        AnytimePolicy, AnytimeStatus, CuAnytimeRunner, CuAnytimeTask, Quality, quality_from_f32,
803    };
804    use crate::input_msg;
805    use crate::output_msg;
806    use cu29_clock::CuDuration;
807    use cu29_traits::CuResult;
808    use rayon::ThreadPoolBuilder;
809    use std::borrow::BorrowMut;
810    use std::sync::OnceLock;
811    use std::sync::mpsc;
812    use std::time::Duration;
813
814    static READY_RX: OnceLock<Arc<Mutex<mpsc::Receiver<CuTime>>>> = OnceLock::new();
815    static DONE_TX: OnceLock<mpsc::Sender<()>> = OnceLock::new();
816    #[derive(Reflect)]
817    struct TestTask {}
818
819    impl Freezable for TestTask {}
820
821    impl CuTask for TestTask {
822        type Resources<'r> = ();
823        type Input<'m> = input_msg!(u32);
824        type Output<'m> = output_msg!(u32);
825
826        fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
827        where
828            Self: Sized,
829        {
830            Ok(Self {})
831        }
832
833        fn process(
834            &mut self,
835            _ctx: &CuContext,
836            input: &Self::Input<'_>,
837            output: &mut Self::Output<'_>,
838        ) -> CuResult<()> {
839            output.borrow_mut().set_payload(*input.payload().unwrap());
840            Ok(())
841        }
842    }
843
844    #[test]
845    fn test_lifecycle() {
846        let tp = Arc::new(
847            rayon::ThreadPoolBuilder::new()
848                .num_threads(1)
849                .build()
850                .unwrap(),
851        );
852
853        let config = ComponentConfig::default();
854        let context = CuContext::new_with_clock();
855        let mut async_task: CuAsyncTask<TestTask, u32> =
856            CuAsyncTask::new(Some(&config), (), tp).unwrap();
857        async_task.start(&context).unwrap();
858        let input = CuMsg::new(Some(42u32));
859        let mut output = CuMsg::new(None);
860
861        loop {
862            {
863                let output_ref: &mut CuMsg<u32> = &mut output;
864                async_task.process(&context, &input, output_ref).unwrap();
865            }
866
867            if let Some(val) = output.payload() {
868                assert_eq!(*val, 42u32);
869                break;
870            }
871        }
872    }
873
874    #[derive(Reflect)]
875    struct ControlledTask;
876
877    impl Freezable for ControlledTask {}
878
879    impl CuTask for ControlledTask {
880        type Resources<'r> = ();
881        type Input<'m> = input_msg!(u32);
882        type Output<'m> = output_msg!(u32);
883
884        fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
885        where
886            Self: Sized,
887        {
888            Ok(Self {})
889        }
890
891        fn process(
892            &mut self,
893            ctx: &CuContext,
894            _input: &Self::Input<'_>,
895            output: &mut Self::Output<'_>,
896        ) -> CuResult<()> {
897            let rx = READY_RX
898                .get()
899                .expect("ready channel not set")
900                .lock()
901                .unwrap();
902            let ready_time = rx
903                .recv_timeout(Duration::from_secs(1))
904                .expect("timed out waiting for ready signal");
905
906            output.set_payload(ready_time.as_nanos() as u32);
907            output.metadata.process_time.start = ctx.now().into();
908            output.metadata.process_time.end = ready_time.into();
909
910            if let Some(done_tx) = DONE_TX.get() {
911                let _ = done_tx.send(());
912            }
913            Ok(())
914        }
915    }
916
917    fn wait_until_async_idle<T, O>(async_task: &CuAsyncTask<T, O>)
918    where
919        T: for<'m> CuTask<Output<'m> = CuMsg<O>> + Send + 'static,
920        O: CuMsgPayload + Send + 'static,
921    {
922        for _ in 0..100 {
923            let state = async_task.state.lock().unwrap();
924            if !matches!(state.status, AsyncStatus::Running(_)) {
925                return;
926            }
927            drop(state);
928            std::thread::sleep(Duration::from_millis(1));
929        }
930        panic!("background task never became idle");
931    }
932
933    fn wait_until_async_src_idle<T, O>(async_task: &CuAsyncSrcTask<T, O>)
934    where
935        T: for<'m> CuSrcTask<Output<'m> = CuMsg<O>> + Send + 'static,
936        O: CuMsgPayload + Send + 'static,
937    {
938        for _ in 0..100 {
939            let state = async_task.state.lock().unwrap();
940            if !matches!(state.status, AsyncStatus::Running(_)) {
941                return;
942            }
943            drop(state);
944            std::thread::sleep(Duration::from_millis(1));
945        }
946        panic!("background source never became idle");
947    }
948
949    #[derive(Clone)]
950    struct ActionTaskResources {
951        actions: Arc<Mutex<mpsc::Receiver<Option<u32>>>>,
952        done: mpsc::Sender<()>,
953    }
954
955    #[derive(Reflect)]
956    #[reflect(no_field_bounds, from_reflect = false)]
957    struct ActionTask {
958        #[reflect(ignore)]
959        actions: Arc<Mutex<mpsc::Receiver<Option<u32>>>>,
960        #[reflect(ignore)]
961        done: mpsc::Sender<()>,
962    }
963
964    impl Freezable for ActionTask {}
965
966    impl CuTask for ActionTask {
967        type Resources<'r> = ActionTaskResources;
968        type Input<'m> = input_msg!(u32);
969        type Output<'m> = output_msg!(u32);
970
971        fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
972        where
973            Self: Sized,
974        {
975            let _ = config;
976            Ok(Self {
977                actions: resources.actions,
978                done: resources.done,
979            })
980        }
981
982        fn process(
983            &mut self,
984            _ctx: &CuContext,
985            _input: &Self::Input<'_>,
986            output: &mut Self::Output<'_>,
987        ) -> CuResult<()> {
988            let action = self
989                .actions
990                .lock()
991                .unwrap()
992                .recv_timeout(Duration::from_secs(1))
993                .expect("timed out waiting for action");
994            if let Some(value) = action {
995                output.set_payload(value);
996            }
997            let _ = self.done.send(());
998            Ok(())
999        }
1000    }
1001
1002    #[derive(Reflect)]
1003    #[reflect(no_field_bounds, from_reflect = false)]
1004    struct ActionSrc {
1005        #[reflect(ignore)]
1006        actions: Arc<Mutex<mpsc::Receiver<Option<u32>>>>,
1007        #[reflect(ignore)]
1008        done: mpsc::Sender<()>,
1009    }
1010
1011    impl Freezable for ActionSrc {}
1012
1013    impl CuSrcTask for ActionSrc {
1014        type Resources<'r> = ActionTaskResources;
1015        type Output<'m> = output_msg!(u32);
1016
1017        fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
1018        where
1019            Self: Sized,
1020        {
1021            let _ = config;
1022            Ok(Self {
1023                actions: resources.actions,
1024                done: resources.done,
1025            })
1026        }
1027
1028        fn process(&mut self, _ctx: &CuContext, output: &mut Self::Output<'_>) -> CuResult<()> {
1029            let action = self
1030                .actions
1031                .lock()
1032                .unwrap()
1033                .recv_timeout(Duration::from_secs(1))
1034                .expect("timed out waiting for source action");
1035            if let Some(value) = action {
1036                output.set_payload(value);
1037            }
1038            let _ = self.done.send(());
1039            Ok(())
1040        }
1041    }
1042
1043    #[derive(Clone)]
1044    struct ControlledSrcResources {
1045        ready_times: Arc<Mutex<mpsc::Receiver<CuTime>>>,
1046        done: mpsc::Sender<()>,
1047    }
1048
1049    #[derive(Reflect)]
1050    #[reflect(no_field_bounds, from_reflect = false)]
1051    struct ControlledSrc {
1052        #[reflect(ignore)]
1053        ready_times: Arc<Mutex<mpsc::Receiver<CuTime>>>,
1054        #[reflect(ignore)]
1055        done: mpsc::Sender<()>,
1056    }
1057
1058    impl Freezable for ControlledSrc {}
1059
1060    impl CuSrcTask for ControlledSrc {
1061        type Resources<'r> = ControlledSrcResources;
1062        type Output<'m> = output_msg!(u32);
1063
1064        fn new(config: Option<&ComponentConfig>, resources: Self::Resources<'_>) -> CuResult<Self>
1065        where
1066            Self: Sized,
1067        {
1068            let _ = config;
1069            Ok(Self {
1070                ready_times: resources.ready_times,
1071                done: resources.done,
1072            })
1073        }
1074
1075        fn process(&mut self, ctx: &CuContext, output: &mut Self::Output<'_>) -> CuResult<()> {
1076            let ready_time = self
1077                .ready_times
1078                .lock()
1079                .unwrap()
1080                .recv_timeout(Duration::from_secs(1))
1081                .expect("timed out waiting for ready signal");
1082            output.set_payload(ready_time.as_nanos() as u32);
1083            output.metadata.process_time.start = ctx.now().into();
1084            output.metadata.process_time.end = ready_time.into();
1085            let _ = self.done.send(());
1086            Ok(())
1087        }
1088    }
1089
1090    #[test]
1091    fn background_clears_output_while_processing() {
1092        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1093        let context = CuContext::new_with_clock();
1094        let (action_tx, action_rx) = mpsc::channel::<Option<u32>>();
1095        let (done_tx, done_rx) = mpsc::channel::<()>();
1096        let resources = ActionTaskResources {
1097            actions: Arc::new(Mutex::new(action_rx)),
1098            done: done_tx,
1099        };
1100
1101        let mut async_task: CuAsyncTask<ActionTask, u32> =
1102            CuAsyncTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
1103        async_task.start(&context).unwrap();
1104        let input = CuMsg::new(Some(1u32));
1105        let mut output = CuMsg::new(None);
1106
1107        async_task.process(&context, &input, &mut output).unwrap();
1108        assert!(output.payload().is_none());
1109
1110        output.set_payload(999);
1111        async_task.process(&context, &input, &mut output).unwrap();
1112        assert!(
1113            output.payload().is_none(),
1114            "background poll should clear stale output while the worker is still running"
1115        );
1116
1117        action_tx.send(Some(7)).unwrap();
1118        done_rx
1119            .recv_timeout(Duration::from_secs(1))
1120            .expect("background worker never finished");
1121    }
1122
1123    #[test]
1124    fn background_empty_run_does_not_reemit_previous_payload() {
1125        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1126        let context = CuContext::new_with_clock();
1127        let (action_tx, action_rx) = mpsc::channel::<Option<u32>>();
1128        let (done_tx, done_rx) = mpsc::channel::<()>();
1129        let resources = ActionTaskResources {
1130            actions: Arc::new(Mutex::new(action_rx)),
1131            done: done_tx,
1132        };
1133
1134        let mut async_task: CuAsyncTask<ActionTask, u32> =
1135            CuAsyncTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
1136        async_task.start(&context).unwrap();
1137        let some_input = CuMsg::new(Some(1u32));
1138        let no_input = CuMsg::new(None::<u32>);
1139        let mut output = CuMsg::new(None);
1140
1141        action_tx.send(Some(42)).unwrap();
1142        async_task
1143            .process(&context, &some_input, &mut output)
1144            .expect("failed to start first background run");
1145        done_rx
1146            .recv_timeout(Duration::from_secs(1))
1147            .expect("first background run never finished");
1148        wait_until_async_idle(&async_task);
1149
1150        action_tx.send(None).unwrap();
1151        async_task
1152            .process(&context, &no_input, &mut output)
1153            .expect("failed to start empty background run");
1154        assert_eq!(output.payload(), Some(&42));
1155        done_rx
1156            .recv_timeout(Duration::from_secs(1))
1157            .expect("empty background run never finished");
1158        wait_until_async_idle(&async_task);
1159
1160        action_tx.send(None).unwrap();
1161        async_task
1162            .process(&context, &no_input, &mut output)
1163            .expect("failed to poll after empty background run");
1164        assert!(
1165            output.payload().is_none(),
1166            "background task re-emitted the previous payload after an empty run"
1167        );
1168        done_rx
1169            .recv_timeout(Duration::from_secs(1))
1170            .expect("cleanup background run never finished");
1171    }
1172
1173    #[test]
1174    fn background_source_clears_output_while_processing() {
1175        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1176        let context = CuContext::new_with_clock();
1177        let (action_tx, action_rx) = mpsc::channel::<Option<u32>>();
1178        let (done_tx, done_rx) = mpsc::channel::<()>();
1179        let resources = ActionTaskResources {
1180            actions: Arc::new(Mutex::new(action_rx)),
1181            done: done_tx,
1182        };
1183
1184        let mut async_src: CuAsyncSrcTask<ActionSrc, u32> =
1185            CuAsyncSrcTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
1186        let mut output = CuMsg::new(None);
1187
1188        async_src.process(&context, &mut output).unwrap();
1189        assert!(output.payload().is_none());
1190
1191        output.set_payload(999);
1192        async_src.process(&context, &mut output).unwrap();
1193        assert!(
1194            output.payload().is_none(),
1195            "background source poll should clear stale output while the worker is still running"
1196        );
1197
1198        action_tx.send(Some(7)).unwrap();
1199        done_rx
1200            .recv_timeout(Duration::from_secs(1))
1201            .expect("background source never finished");
1202    }
1203
1204    #[test]
1205    fn background_source_empty_run_does_not_reemit_previous_payload() {
1206        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1207        let context = CuContext::new_with_clock();
1208        let (action_tx, action_rx) = mpsc::channel::<Option<u32>>();
1209        let (done_tx, done_rx) = mpsc::channel::<()>();
1210        let resources = ActionTaskResources {
1211            actions: Arc::new(Mutex::new(action_rx)),
1212            done: done_tx,
1213        };
1214
1215        let mut async_src: CuAsyncSrcTask<ActionSrc, u32> =
1216            CuAsyncSrcTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
1217        let mut output = CuMsg::new(None);
1218
1219        action_tx.send(Some(42)).unwrap();
1220        async_src
1221            .process(&context, &mut output)
1222            .expect("failed to start first background source run");
1223        done_rx
1224            .recv_timeout(Duration::from_secs(1))
1225            .expect("first background source run never finished");
1226        wait_until_async_src_idle(&async_src);
1227
1228        action_tx.send(None).unwrap();
1229        async_src
1230            .process(&context, &mut output)
1231            .expect("failed to start empty background source run");
1232        assert_eq!(output.payload(), Some(&42));
1233        done_rx
1234            .recv_timeout(Duration::from_secs(1))
1235            .expect("empty background source run never finished");
1236        wait_until_async_src_idle(&async_src);
1237
1238        action_tx.send(None).unwrap();
1239        async_src
1240            .process(&context, &mut output)
1241            .expect("failed to poll background source after empty run");
1242        assert!(
1243            output.payload().is_none(),
1244            "background source re-emitted the previous payload after an empty run"
1245        );
1246        done_rx
1247            .recv_timeout(Duration::from_secs(1))
1248            .expect("cleanup background source run never finished");
1249    }
1250
1251    #[test]
1252    fn background_respects_recorded_ready_time() {
1253        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1254        let (context, clock_mock) = CuContext::new_mock_clock();
1255
1256        // Install the control channels for the task.
1257        let (ready_tx, ready_rx) = mpsc::channel::<CuTime>();
1258        let (done_tx, done_rx) = mpsc::channel::<()>();
1259        READY_RX
1260            .set(Arc::new(Mutex::new(ready_rx)))
1261            .expect("ready channel already set");
1262        DONE_TX
1263            .set(done_tx)
1264            .expect("completion channel already set");
1265
1266        let mut async_task: CuAsyncTask<ControlledTask, u32> =
1267            CuAsyncTask::new(Some(&ComponentConfig::default()), (), tp.clone()).unwrap();
1268        async_task.start(&context).unwrap();
1269        let input = CuMsg::new(Some(1u32));
1270        let mut output = CuMsg::new(None);
1271
1272        // Copperlist 0: kick off processing, nothing ready yet.
1273        clock_mock.set_value(0);
1274        async_task.process(&context, &input, &mut output).unwrap();
1275        assert!(output.payload().is_none());
1276
1277        // Copperlist 1 at time 10: still running in the background.
1278        clock_mock.set_value(10);
1279        async_task.process(&context, &input, &mut output).unwrap();
1280        assert!(output.payload().is_none());
1281
1282        // The background thread finishes at time 30 (recorded in metadata).
1283        clock_mock.set_value(30);
1284        ready_tx.send(CuTime::from(30u64)).unwrap();
1285        done_rx
1286            .recv_timeout(Duration::from_secs(1))
1287            .expect("background task never finished");
1288        // Wait until the async wrapper has cleared its processing flag and captured ready_at.
1289        let mut ready_at_recorded = false;
1290        for _ in 0..100 {
1291            let state = async_task.state.lock().unwrap();
1292            if matches!(state.status, AsyncStatus::Waiting(_)) {
1293                ready_at_recorded = true;
1294                break;
1295            }
1296            drop(state);
1297            std::thread::sleep(Duration::from_millis(1));
1298        }
1299        assert!(
1300            ready_at_recorded,
1301            "background task finished without recording ready_at"
1302        );
1303
1304        // Replay earlier than the recorded end time: the output should be held back.
1305        clock_mock.set_value(20);
1306        async_task.process(&context, &input, &mut output).unwrap();
1307        assert!(
1308            output.payload().is_none(),
1309            "Output surfaced before recorded ready time"
1310        );
1311
1312        // Once the mock clock reaches the recorded end time, the result is released.
1313        clock_mock.set_value(30);
1314        async_task.process(&context, &input, &mut output).unwrap();
1315        assert_eq!(output.payload(), Some(&30u32));
1316
1317        // Allow the background worker spawned by the last poll to complete so the thread pool shuts down cleanly.
1318        ready_tx.send(CuTime::from(40u64)).unwrap();
1319        let _ = done_rx.recv_timeout(Duration::from_secs(1));
1320    }
1321
1322    /// Anytime task under the wrapper: one increment per quantum, quality
1323    /// climbing toward the input target.
1324    #[derive(Reflect)]
1325    struct IncrementalPlanner {
1326        target: u32,
1327        acc: u32,
1328    }
1329
1330    impl Freezable for IncrementalPlanner {}
1331
1332    impl CuAnytimeTask for IncrementalPlanner {
1333        type Input<'m> = input_msg!(u32);
1334        type Output<'m> = output_msg!(u32);
1335        type Resources<'r> = ();
1336        type Quality = Quality;
1337
1338        fn new(_config: Option<&ComponentConfig>, _resources: ()) -> CuResult<Self> {
1339            Ok(Self { target: 0, acc: 0 })
1340        }
1341
1342        fn base(
1343            &mut self,
1344            _ctx: &CuContext,
1345            input: &Self::Input<'_>,
1346            output: &mut Self::Output<'_>,
1347        ) -> CuResult<AnytimeStatus<Quality>> {
1348            self.target = input.payload().copied().ok_or("planner: no input")?;
1349            self.acc = 0;
1350            output.set_payload(self.acc);
1351            Ok(AnytimeStatus::Improved(quality_from_f32(0.0)))
1352        }
1353
1354        fn refine(
1355            &mut self,
1356            _ctx: &CuContext,
1357            output: &mut Self::Output<'_>,
1358        ) -> CuResult<AnytimeStatus<Quality>> {
1359            self.acc += 1;
1360            output.set_payload(self.acc);
1361            Ok(AnytimeStatus::Improved(quality_from_f32(
1362                self.acc as f32 / self.target as f32,
1363            )))
1364        }
1365    }
1366
1367    /// Mirrors codegen for `anytime: (max_refines: 3)` on a background node.
1368    struct ThreeQuantaPolicy;
1369    impl<Q: Copy + PartialOrd> AnytimePolicy<Q> for ThreeQuantaPolicy {
1370        const TIME_BUDGET: Option<CuDuration> = None;
1371        const MAX_AGE: Option<CuDuration> = None;
1372        const MAX_STALL: Option<u32> = None;
1373        const MAX_REFINES: Option<u32> = Some(3);
1374    }
1375
1376    #[test]
1377    fn background_anytime_job_lands_with_its_status_stamp() {
1378        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1379        let context = CuContext::new_with_clock();
1380        let mut task: CuAsyncTask<CuAnytimeRunner<IncrementalPlanner, ThreeQuantaPolicy>, u32> =
1381            CuAsyncTask::new(Some(&ComponentConfig::default()), (), tp).unwrap();
1382        task.start(&context).unwrap();
1383
1384        let input = CuMsg::new(Some(5u32));
1385        let mut output = CuMsg::new(None);
1386
1387        // Poll until the worker's job comes back through the buffered output.
1388        for _ in 0..1000 {
1389            task.process(&context, &input, &mut output).unwrap();
1390            if output.payload().is_some() {
1391                break;
1392            }
1393            std::thread::sleep(Duration::from_millis(1));
1394        }
1395
1396        // Three quanta of a job needing five: stopped by the quanta bound, and
1397        // the stamp the runner wrote survived the buffered-output copy.
1398        assert_eq!(output.payload(), Some(&3));
1399        assert_eq!(output.metadata.status_txt.0.as_str(), "any:3it q=0.60 max");
1400    }
1401
1402    #[test]
1403    fn background_source_respects_recorded_ready_time() {
1404        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1405        let (context, clock_mock) = CuContext::new_mock_clock();
1406        let (ready_tx, ready_rx) = mpsc::channel::<CuTime>();
1407        let (done_tx, done_rx) = mpsc::channel::<()>();
1408        let resources = ControlledSrcResources {
1409            ready_times: Arc::new(Mutex::new(ready_rx)),
1410            done: done_tx,
1411        };
1412
1413        let mut async_src: CuAsyncSrcTask<ControlledSrc, u32> =
1414            CuAsyncSrcTask::new(Some(&ComponentConfig::default()), resources, tp.clone()).unwrap();
1415        let mut output = CuMsg::new(None);
1416
1417        clock_mock.set_value(0);
1418        async_src.process(&context, &mut output).unwrap();
1419        assert!(output.payload().is_none());
1420
1421        clock_mock.set_value(10);
1422        async_src.process(&context, &mut output).unwrap();
1423        assert!(output.payload().is_none());
1424
1425        clock_mock.set_value(30);
1426        ready_tx.send(CuTime::from(30u64)).unwrap();
1427        done_rx
1428            .recv_timeout(Duration::from_secs(1))
1429            .expect("background source never finished");
1430
1431        let mut ready_at_recorded = false;
1432        for _ in 0..100 {
1433            let state = async_src.state.lock().unwrap();
1434            if matches!(state.status, AsyncStatus::Waiting(_)) {
1435                ready_at_recorded = true;
1436                break;
1437            }
1438            drop(state);
1439            std::thread::sleep(Duration::from_millis(1));
1440        }
1441        assert!(
1442            ready_at_recorded,
1443            "background source finished without recording ready_at"
1444        );
1445
1446        clock_mock.set_value(20);
1447        async_src.process(&context, &mut output).unwrap();
1448        assert!(
1449            output.payload().is_none(),
1450            "background source surfaced output before recorded ready time"
1451        );
1452
1453        clock_mock.set_value(30);
1454        async_src.process(&context, &mut output).unwrap();
1455        assert_eq!(output.payload(), Some(&30u32));
1456
1457        ready_tx.send(CuTime::from(40u64)).unwrap();
1458        let _ = done_rx.recv_timeout(Duration::from_secs(1));
1459    }
1460
1461    type ReplayTaskObservation = (u32, u64, u32, u32);
1462
1463    #[derive(Clone)]
1464    struct ReplayTaskResources {
1465        release: Arc<Mutex<mpsc::Receiver<()>>>,
1466        observed: mpsc::Sender<ReplayTaskObservation>,
1467    }
1468
1469    #[derive(Reflect)]
1470    #[reflect(no_field_bounds, from_reflect = false)]
1471    struct ReplayTask {
1472        counter: u32,
1473        #[reflect(ignore)]
1474        release: Arc<Mutex<mpsc::Receiver<()>>>,
1475        #[reflect(ignore)]
1476        observed: mpsc::Sender<ReplayTaskObservation>,
1477    }
1478
1479    impl Freezable for ReplayTask {
1480        fn freeze<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
1481            self.counter.encode(encoder)
1482        }
1483
1484        fn thaw<D: Decoder>(&mut self, decoder: &mut D) -> Result<(), DecodeError> {
1485            self.counter = u32::decode(decoder)?;
1486            Ok(())
1487        }
1488    }
1489
1490    impl CuTask for ReplayTask {
1491        type Resources<'r> = ReplayTaskResources;
1492        type Input<'m> = input_msg!(u32);
1493        type Output<'m> = output_msg!(u32);
1494
1495        fn new(
1496            _config: Option<&ComponentConfig>,
1497            resources: Self::Resources<'_>,
1498        ) -> CuResult<Self> {
1499            Ok(Self {
1500                counter: 0,
1501                release: resources.release,
1502                observed: resources.observed,
1503            })
1504        }
1505
1506        fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
1507            self.counter = 10;
1508            Ok(())
1509        }
1510
1511        fn process(
1512            &mut self,
1513            ctx: &CuContext,
1514            input: &Self::Input<'_>,
1515            output: &mut Self::Output<'_>,
1516        ) -> CuResult<()> {
1517            self.release
1518                .lock()
1519                .unwrap()
1520                .recv_timeout(Duration::from_secs(1))
1521                .expect("timed out waiting to release replay task");
1522            let input = input.payload().copied().expect("replay task input");
1523            self.observed
1524                .send((input, ctx.cl_id(), ctx.instance_id(), self.counter))
1525                .expect("failed to record replay task dispatch");
1526            self.counter += 1;
1527            output.set_payload(input + self.counter);
1528            Ok(())
1529        }
1530    }
1531
1532    fn replay_task_resources() -> (
1533        ReplayTaskResources,
1534        mpsc::Sender<()>,
1535        mpsc::Receiver<ReplayTaskObservation>,
1536    ) {
1537        let (release_tx, release_rx) = mpsc::channel();
1538        let (observed_tx, observed_rx) = mpsc::channel();
1539        (
1540            ReplayTaskResources {
1541                release: Arc::new(Mutex::new(release_rx)),
1542                observed: observed_tx,
1543            },
1544            release_tx,
1545            observed_rx,
1546        )
1547    }
1548
1549    #[test]
1550    fn background_freeze_mid_run_replays_original_dispatch_from_committed_state() {
1551        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1552        let (resources, release_tx, observed_rx) = replay_task_resources();
1553        let (base_context, _) = CuContext::new_mock_clock();
1554        let mut dispatch_context = CuContext::builder(base_context.clock.clone())
1555            .cl_id(7)
1556            .instance_id(3)
1557            .task_ids(&["replay"])
1558            .build();
1559        dispatch_context.set_current_task(0);
1560        let mut original: CuAsyncTask<ReplayTask, u32> =
1561            CuAsyncTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
1562        original.start(&dispatch_context).unwrap();
1563        let original_input = CuMsg::new(Some(41u32));
1564        let mut output = CuMsg::default();
1565        original
1566            .process(&dispatch_context, &original_input, &mut output)
1567            .unwrap();
1568
1569        let frozen = bincode::encode_to_vec(BincodeAdapter(&original), standard())
1570            .expect("mid-run async freeze failed");
1571        let committed_task = bincode::encode_to_vec(10u32, standard()).unwrap();
1572        let expected = bincode::encode_to_vec(
1573            (
1574                committed_task,
1575                CuMsg::<u32>::default(),
1576                ASYNC_PENDING_TAG,
1577                7u64,
1578                original_input.clone(),
1579            ),
1580            standard(),
1581        )
1582        .unwrap();
1583        assert_eq!(
1584            frozen, expected,
1585            "pending task frames contain only committed state, CL id, and input"
1586        );
1587        assert!(
1588            matches!(
1589                original.state.lock().unwrap().status,
1590                AsyncStatus::Running(7)
1591            ),
1592            "freezing must not disturb the live worker"
1593        );
1594        release_tx.send(()).unwrap();
1595        assert_eq!(
1596            observed_rx.recv_timeout(Duration::from_secs(1)).unwrap(),
1597            (41, 7, 3, 10)
1598        );
1599
1600        let replay_tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1601        let (replay_resources, replay_release_tx, replay_observed_rx) = replay_task_resources();
1602        let mut restored: CuAsyncTask<ReplayTask, u32> = CuAsyncTask::new(
1603            Some(&ComponentConfig::default()),
1604            replay_resources,
1605            replay_tp,
1606        )
1607        .unwrap();
1608        restored.start(&dispatch_context).unwrap();
1609        let reader = bincode::de::read::SliceReader::new(&frozen);
1610        let mut decoder = bincode::de::DecoderImpl::new(reader, standard(), ());
1611        restored.thaw(&mut decoder).unwrap();
1612        assert!(matches!(
1613            restored.state.lock().unwrap().status,
1614            AsyncStatus::ReplayPending(7)
1615        ));
1616
1617        let mut current_context = CuContext::builder(base_context.clock.clone())
1618            .cl_id(99)
1619            .instance_id(8)
1620            .task_ids(&["replay"])
1621            .build();
1622        current_context.set_current_task(0);
1623        let current_input = CuMsg::new(Some(999u32));
1624        restored
1625            .process(&current_context, &current_input, &mut output)
1626            .unwrap();
1627        replay_release_tx.send(()).unwrap();
1628        assert_eq!(
1629            replay_observed_rx
1630                .recv_timeout(Duration::from_secs(1))
1631                .unwrap(),
1632            (41, 7, 8, 10),
1633            "replay must retain only the original input and CopperList id"
1634        );
1635        wait_until_async_idle(&restored);
1636
1637        restored
1638            .process(&current_context, &current_input, &mut output)
1639            .unwrap();
1640        assert_eq!(output.payload(), Some(&52));
1641        replay_release_tx.send(()).unwrap();
1642        let _ = replay_observed_rx.recv_timeout(Duration::from_secs(1));
1643    }
1644
1645    #[cfg(feature = "memory_monitoring")]
1646    #[test]
1647    fn background_freeze_mid_run_does_not_allocate() {
1648        const SNAPSHOT_CAPACITY: usize = 4 * 1024;
1649
1650        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1651        let (resources, release_tx, observed_rx) = replay_task_resources();
1652        let (context, _) = CuContext::new_mock_clock();
1653        let mut async_task: CuAsyncTask<ReplayTask, u32> =
1654            CuAsyncTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
1655        async_task.start(&context).unwrap();
1656        let input = CuMsg::new(Some(41u32));
1657        let mut output = CuMsg::default();
1658        async_task.process(&context, &input, &mut output).unwrap();
1659
1660        let mut snapshot = [0u8; SNAPSHOT_CAPACITY];
1661        let allocations = crate::monitoring::ScopedAllocCounter::new();
1662        let writer = bincode::enc::write::SliceWriter::new(&mut snapshot);
1663        let mut encoder = EncoderImpl::new(writer, standard());
1664        BincodeAdapter(&async_task).encode(&mut encoder).unwrap();
1665        assert_eq!(allocations.allocated(), 0);
1666
1667        release_tx.send(()).unwrap();
1668        let _ = observed_rx.recv_timeout(Duration::from_secs(1));
1669    }
1670
1671    #[test]
1672    fn background_freeze_thaw_preserves_pending_error() {
1673        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1674        let context = CuContext::new_with_clock();
1675        let mut original: CuAsyncTask<TestTask, u32> =
1676            CuAsyncTask::new(Some(&ComponentConfig::default()), (), tp).unwrap();
1677        original.start(&context).unwrap();
1678        {
1679            let mut state = original.state.lock().unwrap();
1680            let error = CuError::from("expected async failure");
1681            state.status = failure(error);
1682        }
1683        let frozen = bincode::encode_to_vec(BincodeAdapter(&original), standard()).unwrap();
1684
1685        let replay_tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1686        let mut restored: CuAsyncTask<TestTask, u32> =
1687            CuAsyncTask::new(Some(&ComponentConfig::default()), (), replay_tp).unwrap();
1688        restored.start(&context).unwrap();
1689        let reader = bincode::de::read::SliceReader::new(&frozen);
1690        let mut decoder = bincode::de::DecoderImpl::new(reader, standard(), ());
1691        restored.thaw(&mut decoder).unwrap();
1692
1693        let input = CuMsg::new(Some(1u32));
1694        let mut output = CuMsg::default();
1695        let error = restored.process(&context, &input, &mut output).unwrap_err();
1696        assert!(error.to_string().contains("expected async failure"));
1697    }
1698
1699    #[test]
1700    fn worker_completion_racing_freeze_is_an_atomic_committed_snapshot() {
1701        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1702        let (resources, release_tx, observed_rx) = replay_task_resources();
1703        let (context, _) = CuContext::new_mock_clock();
1704        let mut async_task: CuAsyncTask<ReplayTask, u32> =
1705            CuAsyncTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
1706        async_task.start(&context).unwrap();
1707        let input = CuMsg::new(Some(41u32));
1708        let mut output = CuMsg::default();
1709        async_task.process(&context, &input, &mut output).unwrap();
1710
1711        let mut snapshots = vec![
1712            bincode::encode_to_vec(BincodeAdapter(&async_task), standard())
1713                .expect("freeze running async task"),
1714        ];
1715        release_tx.send(()).unwrap();
1716        observed_rx
1717            .recv_timeout(Duration::from_secs(1))
1718            .expect("worker did not complete");
1719        for _ in 0..100 {
1720            snapshots.push(
1721                bincode::encode_to_vec(BincodeAdapter(&async_task), standard())
1722                    .expect("freeze async task racing completion"),
1723            );
1724            if !matches!(
1725                async_task.state.lock().unwrap().status,
1726                AsyncStatus::Running(_)
1727            ) {
1728                break;
1729            }
1730            std::thread::yield_now();
1731        }
1732        wait_until_async_idle(&async_task);
1733        snapshots.push(
1734            bincode::encode_to_vec(BincodeAdapter(&async_task), standard())
1735                .expect("freeze completed async task"),
1736        );
1737
1738        let mut saw_running = false;
1739        let mut saw_idle = false;
1740        for snapshot in snapshots {
1741            let reader = bincode::de::read::SliceReader::new(&snapshot);
1742            let mut decoder = bincode::de::DecoderImpl::new(reader, standard(), ());
1743            let dispatch = DispatchSlot::<u32>::new();
1744            let state: AsyncState<u32> =
1745                decode_async_state(&mut decoder).expect("decode raced snapshot");
1746            if matches!(state.status, AsyncStatus::ReplayPending(_)) {
1747                dispatch.decode_input(decoder.reader()).unwrap();
1748            }
1749            let (task_counter, bytes_read): (u32, usize) =
1750                bincode::decode_from_slice(&state.committed_task, standard())
1751                    .expect("decode committed task");
1752            assert_eq!(bytes_read, state.committed_task.len());
1753            match state.status {
1754                AsyncStatus::ReplayPending(0) => {
1755                    saw_running = true;
1756                    assert_eq!(task_counter, 10);
1757                    assert!(state.committed_output.payload().is_none());
1758                    assert_eq!(dispatch.get().payload(), Some(&41));
1759                }
1760                AsyncStatus::Waiting(_) => {
1761                    saw_idle = true;
1762                    assert_eq!(task_counter, 11);
1763                    assert_eq!(state.committed_output.payload(), Some(&52));
1764                }
1765                _ => panic!("decoded keyframe was neither pending nor committed"),
1766            }
1767        }
1768        assert!(saw_running, "race did not capture the pre-completion state");
1769        assert!(
1770            saw_idle,
1771            "race did not capture the committed completion state"
1772        );
1773    }
1774
1775    #[derive(Clone)]
1776    struct ReplaySrcResources {
1777        release: Arc<Mutex<mpsc::Receiver<()>>>,
1778        observed: mpsc::Sender<(u64, u32)>,
1779    }
1780
1781    #[derive(Reflect)]
1782    #[reflect(no_field_bounds, from_reflect = false)]
1783    struct ReplaySrc {
1784        counter: u32,
1785        #[reflect(ignore)]
1786        release: Arc<Mutex<mpsc::Receiver<()>>>,
1787        #[reflect(ignore)]
1788        observed: mpsc::Sender<(u64, u32)>,
1789    }
1790
1791    impl Freezable for ReplaySrc {
1792        fn freeze<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
1793            self.counter.encode(encoder)
1794        }
1795
1796        fn thaw<D: Decoder>(&mut self, decoder: &mut D) -> Result<(), DecodeError> {
1797            self.counter = u32::decode(decoder)?;
1798            Ok(())
1799        }
1800    }
1801
1802    impl CuSrcTask for ReplaySrc {
1803        type Resources<'r> = ReplaySrcResources;
1804        type Output<'m> = output_msg!(u32);
1805
1806        fn new(
1807            _config: Option<&ComponentConfig>,
1808            resources: Self::Resources<'_>,
1809        ) -> CuResult<Self> {
1810            Ok(Self {
1811                counter: 0,
1812                release: resources.release,
1813                observed: resources.observed,
1814            })
1815        }
1816
1817        fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
1818            self.counter = 20;
1819            Ok(())
1820        }
1821
1822        fn process(&mut self, ctx: &CuContext, output: &mut Self::Output<'_>) -> CuResult<()> {
1823            self.release
1824                .lock()
1825                .unwrap()
1826                .recv_timeout(Duration::from_secs(1))
1827                .expect("timed out waiting to release replay source");
1828            self.observed
1829                .send((ctx.cl_id(), self.counter))
1830                .expect("failed to record replay source dispatch");
1831            self.counter += 1;
1832            output.set_payload(self.counter);
1833            Ok(())
1834        }
1835    }
1836
1837    fn replay_src_resources() -> (
1838        ReplaySrcResources,
1839        mpsc::Sender<()>,
1840        mpsc::Receiver<(u64, u32)>,
1841    ) {
1842        let (release_tx, release_rx) = mpsc::channel();
1843        let (observed_tx, observed_rx) = mpsc::channel();
1844        (
1845            ReplaySrcResources {
1846                release: Arc::new(Mutex::new(release_rx)),
1847                observed: observed_tx,
1848            },
1849            release_tx,
1850            observed_rx,
1851        )
1852    }
1853
1854    #[test]
1855    fn background_source_freeze_mid_run_replays_original_context() {
1856        let tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1857        let (resources, release_tx, observed_rx) = replay_src_resources();
1858        let (base_context, _) = CuContext::new_mock_clock();
1859        let dispatch_context = CuContext::builder(base_context.clock.clone())
1860            .cl_id(7)
1861            .build();
1862        let mut original: CuAsyncSrcTask<ReplaySrc, u32> =
1863            CuAsyncSrcTask::new(Some(&ComponentConfig::default()), resources, tp).unwrap();
1864        original.start(&dispatch_context).unwrap();
1865        let mut output = CuMsg::default();
1866        original.process(&dispatch_context, &mut output).unwrap();
1867        let frozen = bincode::encode_to_vec(BincodeAdapter(&original), standard())
1868            .expect("mid-run async source freeze failed");
1869        let committed_task = bincode::encode_to_vec(20u32, standard()).unwrap();
1870        let expected = bincode::encode_to_vec(
1871            (
1872                committed_task,
1873                CuMsg::<u32>::default(),
1874                ASYNC_PENDING_TAG,
1875                7u64,
1876            ),
1877            standard(),
1878        )
1879        .unwrap();
1880        assert_eq!(
1881            frozen, expected,
1882            "pending source frames contain no task-only dispatch marker"
1883        );
1884        release_tx.send(()).unwrap();
1885        assert_eq!(
1886            observed_rx.recv_timeout(Duration::from_secs(1)).unwrap(),
1887            (7, 20)
1888        );
1889
1890        let replay_tp = Arc::new(ThreadPoolBuilder::new().num_threads(1).build().unwrap());
1891        let (replay_resources, replay_release_tx, replay_observed_rx) = replay_src_resources();
1892        let mut restored: CuAsyncSrcTask<ReplaySrc, u32> = CuAsyncSrcTask::new(
1893            Some(&ComponentConfig::default()),
1894            replay_resources,
1895            replay_tp,
1896        )
1897        .unwrap();
1898        restored.start(&dispatch_context).unwrap();
1899        let reader = bincode::de::read::SliceReader::new(&frozen);
1900        let mut decoder = bincode::de::DecoderImpl::new(reader, standard(), ());
1901        restored.thaw(&mut decoder).unwrap();
1902        let current_context = CuContext::builder(base_context.clock.clone())
1903            .cl_id(99)
1904            .build();
1905        restored.process(&current_context, &mut output).unwrap();
1906        replay_release_tx.send(()).unwrap();
1907        assert_eq!(
1908            replay_observed_rx
1909                .recv_timeout(Duration::from_secs(1))
1910                .unwrap(),
1911            (7, 20)
1912        );
1913    }
1914}
cu29_runtime/cuasynctask.rs

cu29_runtime/
cuasynctask.rs
