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

cu29_runtime/
cuasynctask.rs
