hydro_lang/live_collections/sliced/
mod.rs

1//! Utilities for transforming live collections via slicing.
2
3pub mod style;
4
5use super::boundedness::{Bounded, Unbounded};
6use super::stream::{Ordering, Retries};
7use crate::location::{Location, Tick};
8
9#[doc(hidden)]
10#[macro_export]
11macro_rules! __sliced_parse_uses__ {
12    // Parse immutable use statements with style: let name = use::style(args...);
13    (
14        @uses [$($uses:tt)*]
15        @states [$($states:tt)*]
16        let $name:ident = use:: $invocation:expr; $($rest:tt)*
17    ) => {
18        $crate::__sliced_parse_uses__!(
19            @uses [$($uses)* { $name, $invocation, $invocation }]
20            @states [$($states)*]
21            $($rest)*
22        )
23    };
24
25    // Parse immutable use statements without style: let name = use(args...);
26    // This form is deprecated; the `default` style function it expands to is marked
27    // `#[deprecated]`, which surfaces a deprecation warning on the user's `use(...)` tokens
28    // (via `copy_span!`).
29    (
30        @uses [$($uses:tt)*]
31        @states [$($states:tt)*]
32        let $name:ident = use($($args:expr),* $(,)?); $($rest:tt)*
33    ) => {
34        $crate::__sliced_parse_uses__!(
35            @uses [$($uses)* { $name, $crate::macro_support::copy_span::copy_span!($($args,)* default)($($args),*), $($args),* }]
36            @states [$($states)*]
37            $($rest)*
38        )
39    };
40
41    // Parse mutable state statements: let mut name = use::style::<Type>(args);
42    (
43        @uses [$($uses:tt)*]
44        @states [$($states:tt)*]
45        let mut $name:ident = use:: $style:ident $(::<$ty:ty>)? ($($args:expr)?); $($rest:tt)*
46    ) => {
47        $crate::__sliced_parse_uses__!(
48            @uses [$($uses)*]
49            @states [$($states)* { $name, $style, (($($ty)?), ($($args)?)) }]
50            $($rest)*
51        )
52    };
53
54    // Terminal case: no uses, only states
55    (
56        @uses []
57        @states [$({ $state_name:ident, $state_style:ident, $state_arg:tt })+]
58        $($body:tt)*
59    ) => {
60        {
61            // We need at least one use to get a tick, so panic if there are none
62            compile_error!("sliced! requires at least one `let name = use(...)` statement to determine the tick")
63        }
64    };
65
66    // Terminal case: uses with optional states
67    (
68        @uses [$({ $use_name:ident, $invocation:expr, $($invocation_spans:expr),* })+]
69        @states [$({ $state_name:ident, $state_style:ident, (($($state_ty:ty)?), ($($state_arg:expr)?)) })*]
70        $($body:tt)*
71    ) => {
72        {
73            use $crate::live_collections::sliced::style::*;
74            let __styled = (
75                $($invocation,)+
76            );
77
78            let __tick = $crate::live_collections::sliced::Slicable::create_tick(&__styled.0);
79            let __backtraces = {
80                use $crate::compile::ir::backtrace::__macro_get_backtrace;
81                (
82                    $($crate::macro_support::copy_span::copy_span!($($invocation_spans,)* {
83                        __macro_get_backtrace(1)
84                    }),)+
85                )
86            };
87            let __sliced = $crate::live_collections::sliced::Slicable::slice(__styled, &__tick, __backtraces);
88            let (
89                $($use_name,)+
90            ) = __sliced;
91
92            // Create all cycles and pack handles/values into tuples.
93            //
94            // The `copy_span!` wrapper re-spans the macro-generated tokens (the style function
95            // path and the `build` call) to the user's tokens, so that errors arising from the
96            // state creation (e.g. an initializer that is not general enough over lifetimes) are
97            // attributed to the originating `use` statement instead of the entire `sliced!`
98            // invocation. The `$state_ty` and `$state_arg` fragments are passed through untouched
99            // (as interpolated fragments), preserving the precise spans of errors within the
100            // user-provided argument.
101            //
102            // Each state borrows its own clone of the tick (created inside the `copy_span!`
103            // target and spanned to the style name, e.g. `state`), so lifetime errors caused
104            // by a bad initializer point at the originating `use` statement rather than the
105            // shared `__tick` local, whose span covers the entire macro invocation.
106            let (__handles, __states) = $crate::live_collections::sliced::unzip_cycles((
107                $($crate::macro_support::copy_span::copy_span!($state_style, {
108                    $crate::live_collections::sliced::style::$state_style$(::<$state_ty, _>)?(& __tick.clone()).build($($state_arg)?)
109                }),)*
110            ));
111
112            // Unpack mutable state values
113            let (
114                $(mut $state_name,)*
115            ) = __states;
116
117            // Execute the body
118            let __body_result = {
119                $($body)*
120            };
121
122            // Re-pack the final state values and complete cycles
123            let __final_states = (
124                $($state_name,)*
125            );
126            $crate::live_collections::sliced::complete_cycles(__handles, __final_states);
127
128            // Unslice the result
129            $crate::live_collections::sliced::Unslicable::unslice(__body_result)
130        }
131    };
132}
133
134#[macro_export]
135/// Transforms a live collection with a computation relying on a slice of another live collection.
136/// This is useful for reading a snapshot of an asynchronously updated collection while processing another
137/// collection, such as joining a stream with the latest values from a singleton.
138///
139/// # Syntax
140/// The `sliced!` macro takes in a closure-like syntax specifying the live collections to be sliced
141/// and the body of the transformation. Each `use` statement indicates a live collection to be sliced,
142/// along with a non-determinism explanation. The style specifies how the live collection is sliced:
143/// `use::batch` for stream-like collections (such as [`Stream`](crate::live_collections::Stream)),
144/// `use::snapshot` for singleton-like collections (such as
145/// [`Singleton`](crate::live_collections::Singleton)), and `use::atomic` for atomically-processed
146/// collections. All `use` statements must appear before the body.
147///
148/// ```rust,ignore
149/// let stream = sliced! {
150///     let name1 = use::batch(stream1, nondet!(/** explanation */));
151///     let name2 = use::snapshot(singleton2, nondet!(/** explanation */));
152///     let name3 = use::atomic(collection3, nondet!(/** explanation */));
153///
154///     // arbitrary statements can follow
155///     let intermediate = name1.map(...);
156///     intermediate.cross_singleton(name2)
157/// };
158/// ```
159///
160/// The style-less form `use(collection, nondet!(...))`, which picks between batching and
161/// snapshotting automatically based on the collection type, is deprecated; use the explicit
162/// `use::batch` or `use::snapshot` styles instead.
163///
164/// # Stateful Computations
165/// The `sliced!` macro also supports stateful computations across iterations using `let mut` bindings
166/// with `use::state` or `use::state_null`. These create cycles that persist values between iterations.
167///
168/// - `use::state(|l| initial)`: Creates a cycle with an initial value. The closure receives
169///   the slice location and returns the initial state for the first iteration.
170/// - `use::state_null::<Type>()`: Creates a cycle that starts as null/empty on the first iteration.
171///
172/// The mutable binding can be reassigned in the body, and the final value will be passed to the
173/// next iteration.
174///
175/// ```rust,ignore
176/// let counter_stream = sliced! {
177///     let batch = use::batch(input_stream, nondet!(/** explanation */));
178///     let mut counter = use::state(|l| l.singleton(q!(0)));
179///
180///     // Increment counter by the number of items in this batch
181///     let new_count = counter.clone().zip(batch.count())
182///         .map(q!(|(old, add)| old + add));
183///     counter = new_count.clone();
184///     new_count.into_stream()
185/// };
186/// ```
187macro_rules! __sliced__ {
188    ($($tt:tt)*) => {
189        $crate::__sliced_parse_uses__!(
190            @uses []
191            @states []
192            $($tt)*
193        )
194    };
195}
196
197pub use crate::__sliced__ as sliced;
198
199/// Marks this live collection as atomically-yielded, which means that the output outside
200/// `sliced` will be at an atomic location that is synchronous with respect to the body
201/// of the slice.
202pub fn yield_atomic<T>(t: T) -> style::Atomic<T> {
203    style::Atomic {
204        collection: t,
205        // yield_atomic doesn't need a nondet since it's for output, not input
206        nondet: crate::nondet::NonDet,
207    }
208}
209
210/// A trait for live collections which can be sliced into bounded versions at a tick.
211pub trait Slicable<'a, L: Location<'a>> {
212    /// The sliced version of this live collection.
213    type Slice;
214
215    /// The type of backtrace associated with this slice.
216    type Backtrace;
217
218    /// Gets the location associated with this live collection.
219    fn get_location(&self) -> L;
220
221    /// Creates a tick that is appropriate for the collection's location.
222    fn create_tick(&self) -> Tick<L> {
223        self.get_location().try_tick().unwrap()
224    }
225
226    /// Slices this live collection at the given tick.
227    ///
228    /// # Non-Determinism
229    /// Slicing a live collection may involve non-determinism, such as choosing which messages
230    /// to include in a batch.
231    fn slice(self, tick: &Tick<L>, backtrace: Self::Backtrace) -> Self::Slice;
232}
233
234/// A trait for live collections which can be yielded out of a slice back into their original form.
235pub trait Unslicable {
236    /// The unsliced version of this live collection.
237    type Unsliced;
238
239    /// Unslices a sliced live collection back into its original form.
240    fn unslice(self) -> Self::Unsliced;
241}
242
243/// A trait for unzipping a tuple of (handle, state) pairs into separate tuples.
244#[doc(hidden)]
245pub trait UnzipCycles {
246    /// The tuple of cycle handles.
247    type Handles;
248    /// The tuple of state values.
249    type States;
250
251    /// Unzips the cycles into handles and states.
252    fn unzip(self) -> (Self::Handles, Self::States);
253}
254
255/// Unzips a tuple of cycles into handles and states.
256#[doc(hidden)]
257pub fn unzip_cycles<T: UnzipCycles>(cycles: T) -> (T::Handles, T::States) {
258    cycles.unzip()
259}
260
261/// A trait for completing a tuple of cycle handles with their final state values.
262#[doc(hidden)]
263pub trait CompleteCycles<States> {
264    /// Completes all cycles with the provided state values.
265    fn complete(self, states: States);
266}
267
268/// Completes a tuple of cycle handles with their final state values.
269#[doc(hidden)]
270pub fn complete_cycles<H: CompleteCycles<S>, S>(handles: H, states: S) {
271    handles.complete(states);
272}
273
274impl<'a, L: Location<'a>> Slicable<'a, L> for () {
275    type Slice = ();
276    type Backtrace = ();
277
278    fn get_location(&self) -> L {
279        unreachable!()
280    }
281
282    fn slice(self, _tick: &Tick<L>, _backtrace: Self::Backtrace) -> Self::Slice {}
283}
284
285impl Unslicable for () {
286    type Unsliced = ();
287
288    fn unslice(self) -> Self::Unsliced {}
289}
290
291macro_rules! impl_slicable_for_tuple {
292    ($($T:ident, $T_bt:ident, $idx:tt),+) => {
293        impl<'a, L: Location<'a>, $($T: Slicable<'a, L>),+> Slicable<'a, L> for ($($T,)+) {
294            type Slice = ($($T::Slice,)+);
295            type Backtrace = ($($T::Backtrace,)+);
296
297            fn get_location(&self) -> L {
298                self.0.get_location()
299            }
300
301            #[expect(non_snake_case, reason = "macro codegen")]
302            fn slice(self, tick: &Tick<L>, backtrace: Self::Backtrace) -> Self::Slice {
303                let ($($T,)+) = self;
304                let ($($T_bt,)+) = backtrace;
305                ($($T.slice(tick, $T_bt),)+)
306            }
307        }
308
309        impl<$($T: Unslicable),+> Unslicable for ($($T,)+) {
310            type Unsliced = ($($T::Unsliced,)+);
311
312            #[expect(non_snake_case, reason = "macro codegen")]
313            fn unslice(self) -> Self::Unsliced {
314                let ($($T,)+) = self;
315                ($($T.unslice(),)+)
316            }
317        }
318    };
319}
320
321#[cfg(stageleft_runtime)]
322impl_slicable_for_tuple!(S1, S1_bt, 0);
323#[cfg(stageleft_runtime)]
324impl_slicable_for_tuple!(S1, S1_bt, 0, S2, S2_bt, 1);
325#[cfg(stageleft_runtime)]
326impl_slicable_for_tuple!(S1, S1_bt, 0, S2, S2_bt, 1, S3, S3_bt, 2);
327#[cfg(stageleft_runtime)]
328impl_slicable_for_tuple!(S1, S1_bt, 0, S2, S2_bt, 1, S3, S3_bt, 2, S4, S4_bt, 3);
329#[cfg(stageleft_runtime)]
330impl_slicable_for_tuple!(
331    S1, S1_bt, 0, S2, S2_bt, 1, S3, S3_bt, 2, S4, S4_bt, 3, S5, S5_bt, 4
332);
333#[cfg(stageleft_runtime)]
334impl_slicable_for_tuple!(
335    S1, S1_bt, 0, S2, S2_bt, 1, S3, S3_bt, 2, S4, S4_bt, 3, S5, S5_bt, 4, S6, S6_bt, 5
336);
337#[cfg(stageleft_runtime)]
338impl_slicable_for_tuple!(
339    S1, S1_bt, 0, S2, S2_bt, 1, S3, S3_bt, 2, S4, S4_bt, 3, S5, S5_bt, 4, S6, S6_bt, 5, S7, S7_bt,
340    6
341);
342#[cfg(stageleft_runtime)]
343impl_slicable_for_tuple!(
344    S1, S1_bt, 0, S2, S2_bt, 1, S3, S3_bt, 2, S4, S4_bt, 3, S5, S5_bt, 4, S6, S6_bt, 5, S7, S7_bt,
345    6, S8, S8_bt, 7
346);
347#[cfg(stageleft_runtime)]
348impl_slicable_for_tuple!(
349    S1, S1_bt, 0, S2, S2_bt, 1, S3, S3_bt, 2, S4, S4_bt, 3, S5, S5_bt, 4, S6, S6_bt, 5, S7, S7_bt,
350    6, S8, S8_bt, 7, S9, S9_bt, 8
351);
352#[cfg(stageleft_runtime)]
353impl_slicable_for_tuple!(
354    S1, S1_bt, 0, S2, S2_bt, 1, S3, S3_bt, 2, S4, S4_bt, 3, S5, S5_bt, 4, S6, S6_bt, 5, S7, S7_bt,
355    6, S8, S8_bt, 7, S9, S9_bt, 8, S10, S10_bt, 9
356);
357#[cfg(stageleft_runtime)]
358impl_slicable_for_tuple!(
359    S1, S1_bt, 0, S2, S2_bt, 1, S3, S3_bt, 2, S4, S4_bt, 3, S5, S5_bt, 4, S6, S6_bt, 5, S7, S7_bt,
360    6, S8, S8_bt, 7, S9, S9_bt, 8, S10, S10_bt, 9, S11, S11_bt, 10
361);
362#[cfg(stageleft_runtime)]
363impl_slicable_for_tuple!(
364    S1, S1_bt, 0, S2, S2_bt, 1, S3, S3_bt, 2, S4, S4_bt, 3, S5, S5_bt, 4, S6, S6_bt, 5, S7, S7_bt,
365    6, S8, S8_bt, 7, S9, S9_bt, 8, S10, S10_bt, 9, S11, S11_bt, 10, S12, S12_bt, 11
366);
367
368macro_rules! impl_cycles_for_tuple {
369    ($($H:ident, $S:ident, $idx:tt),*) => {
370        impl<$($H, $S),*> UnzipCycles for ($(($H, $S),)*) {
371            type Handles = ($($H,)*);
372            type States = ($($S,)*);
373
374            #[expect(clippy::allow_attributes, reason = "macro codegen")]
375            #[allow(non_snake_case, reason = "macro codegen")]
376            fn unzip(self) -> (Self::Handles, Self::States) {
377                let ($($H,)*) = self;
378                (
379                    ($($H.0,)*),
380                    ($($H.1,)*),
381                )
382            }
383        }
384
385        impl<$($H: crate::forward_handle::CompleteCycle<$S>, $S),*> CompleteCycles<($($S,)*)> for ($($H,)*) {
386            #[expect(clippy::allow_attributes, reason = "macro codegen")]
387            #[allow(non_snake_case, reason = "macro codegen")]
388            fn complete(self, states: ($($S,)*)) {
389                let ($($H,)*) = self;
390                let ($($S,)*) = states;
391                $($H.complete_next_tick($S);)*
392            }
393        }
394    };
395}
396
397#[cfg(stageleft_runtime)]
398impl_cycles_for_tuple!();
399#[cfg(stageleft_runtime)]
400impl_cycles_for_tuple!(H1, S1, 0);
401#[cfg(stageleft_runtime)]
402impl_cycles_for_tuple!(H1, S1, 0, H2, S2, 1);
403#[cfg(stageleft_runtime)]
404impl_cycles_for_tuple!(H1, S1, 0, H2, S2, 1, H3, S3, 2);
405#[cfg(stageleft_runtime)]
406impl_cycles_for_tuple!(H1, S1, 0, H2, S2, 1, H3, S3, 2, H4, S4, 3);
407#[cfg(stageleft_runtime)]
408impl_cycles_for_tuple!(H1, S1, 0, H2, S2, 1, H3, S3, 2, H4, S4, 3, H5, S5, 4);
409#[cfg(stageleft_runtime)]
410impl_cycles_for_tuple!(
411    H1, S1, 0, H2, S2, 1, H3, S3, 2, H4, S4, 3, H5, S5, 4, H6, S6, 5
412);
413#[cfg(stageleft_runtime)]
414impl_cycles_for_tuple!(
415    H1, S1, 0, H2, S2, 1, H3, S3, 2, H4, S4, 3, H5, S5, 4, H6, S6, 5, H7, S7, 6
416);
417#[cfg(stageleft_runtime)]
418impl_cycles_for_tuple!(
419    H1, S1, 0, H2, S2, 1, H3, S3, 2, H4, S4, 3, H5, S5, 4, H6, S6, 5, H7, S7, 6, H8, S8, 7
420);
421#[cfg(stageleft_runtime)]
422impl_cycles_for_tuple!(
423    H1, S1, 0, H2, S2, 1, H3, S3, 2, H4, S4, 3, H5, S5, 4, H6, S6, 5, H7, S7, 6, H8, S8, 7, H9, S9,
424    8
425);
426#[cfg(stageleft_runtime)]
427impl_cycles_for_tuple!(
428    H1, S1, 0, H2, S2, 1, H3, S3, 2, H4, S4, 3, H5, S5, 4, H6, S6, 5, H7, S7, 6, H8, S8, 7, H9, S9,
429    8, H10, S10, 9
430);
431#[cfg(stageleft_runtime)]
432impl_cycles_for_tuple!(
433    H1, S1, 0, H2, S2, 1, H3, S3, 2, H4, S4, 3, H5, S5, 4, H6, S6, 5, H7, S7, 6, H8, S8, 7, H9, S9,
434    8, H10, S10, 9, H11, S11, 10
435);
436#[cfg(stageleft_runtime)]
437impl_cycles_for_tuple!(
438    H1, S1, 0, H2, S2, 1, H3, S3, 2, H4, S4, 3, H5, S5, 4, H6, S6, 5, H7, S7, 6, H8, S8, 7, H9, S9,
439    8, H10, S10, 9, H11, S11, 10, H12, S12, 11
440);
441
442// Unslicable implementations for plain collections (used when returning from sliced! body)
443impl<'a, T, L: Location<'a>, O: Ordering, R: Retries> Unslicable
444    for super::Stream<T, Tick<L>, Bounded, O, R>
445{
446    type Unsliced = super::Stream<T, L, Unbounded, O, R>;
447
448    fn unslice(self) -> Self::Unsliced {
449        self.all_ticks()
450    }
451}
452
453impl<'a, T, L: Location<'a>> Unslicable for super::Singleton<T, Tick<L>, Bounded> {
454    type Unsliced = super::Singleton<T, L, Unbounded>;
455
456    fn unslice(self) -> Self::Unsliced {
457        self.latest()
458    }
459}
460
461impl<'a, T, L: Location<'a>> Unslicable for super::Optional<T, Tick<L>, Bounded> {
462    type Unsliced = super::Optional<T, L, Unbounded>;
463
464    fn unslice(self) -> Self::Unsliced {
465        self.latest()
466    }
467}
468
469impl<'a, K, V, L: Location<'a>, O: Ordering, R: Retries> Unslicable
470    for super::KeyedStream<K, V, Tick<L>, Bounded, O, R>
471{
472    type Unsliced = super::KeyedStream<K, V, L, Unbounded, O, R>;
473
474    fn unslice(self) -> Self::Unsliced {
475        self.all_ticks()
476    }
477}
478
479// Unslicable implementations for Atomic-wrapped bounded collections
480impl<'a, T, L: Location<'a>, O: Ordering, R: Retries> Unslicable
481    for style::Atomic<super::Stream<T, Tick<L>, Bounded, O, R>>
482{
483    type Unsliced = super::Stream<T, crate::location::Atomic<L>, Unbounded, O, R>;
484
485    fn unslice(self) -> Self::Unsliced {
486        self.collection.all_ticks_atomic()
487    }
488}
489
490impl<'a, T, L: Location<'a>> Unslicable for style::Atomic<super::Singleton<T, Tick<L>, Bounded>> {
491    type Unsliced = super::Singleton<T, crate::location::Atomic<L>, Unbounded>;
492
493    fn unslice(self) -> Self::Unsliced {
494        self.collection.latest_atomic()
495    }
496}
497
498impl<'a, T, L: Location<'a>> Unslicable for style::Atomic<super::Optional<T, Tick<L>, Bounded>> {
499    type Unsliced = super::Optional<T, crate::location::Atomic<L>, Unbounded>;
500
501    fn unslice(self) -> Self::Unsliced {
502        self.collection.latest_atomic()
503    }
504}
505
506impl<'a, K, V, L: Location<'a>, O: Ordering, R: Retries> Unslicable
507    for style::Atomic<super::KeyedStream<K, V, Tick<L>, Bounded, O, R>>
508{
509    type Unsliced = super::KeyedStream<K, V, crate::location::Atomic<L>, Unbounded, O, R>;
510
511    fn unslice(self) -> Self::Unsliced {
512        self.collection.all_ticks_atomic()
513    }
514}
515
516#[cfg(feature = "sim")]
517#[cfg(test)]
518mod tests {
519    use stageleft::q;
520
521    use super::sliced;
522    use crate::location::Location;
523    use crate::nondet::nondet;
524    use crate::prelude::FlowBuilder;
525
526    /// Test a counter using `use::state` with an initial singleton value.
527    /// Each input increments the counter, and we verify the output after each tick.
528    #[test]
529    fn sim_state_counter() {
530        let mut flow = FlowBuilder::new();
531        let node = flow.process::<()>();
532
533        let (input_send, input) = node.sim_input::<i32, _, _>();
534
535        let out_recv = sliced! {
536            let batch = use::batch(input, nondet!(/** test */));
537            let mut counter = use::state(|l| l.singleton(q!(0)));
538
539            let new_count = counter.clone().zip(batch.count())
540                .map(q!(|(old, add)| old + add));
541            counter = new_count.clone();
542            new_count.into_stream()
543        }
544        .sim_output();
545
546        flow.sim().exhaustive(async || {
547            input_send.send(1);
548            assert_eq!(out_recv.next().await, 1);
549
550            input_send.send(1);
551            assert_eq!(out_recv.next().await, 2);
552
553            input_send.send(1);
554            assert_eq!(out_recv.next().await, 3);
555        });
556    }
557
558    /// Test `use::state_null` with an Optional that starts as None.
559    #[cfg(feature = "sim")]
560    #[test]
561    fn sim_state_null_optional() {
562        use crate::live_collections::Optional;
563        use crate::live_collections::boundedness::Bounded;
564        use crate::location::{Location, Tick};
565
566        let mut flow = FlowBuilder::new();
567        let node = flow.process::<()>();
568
569        let (input_send, input) = node.sim_input::<i32, _, _>();
570
571        let out_recv = sliced! {
572            let batch = use::batch(input, nondet!(/** test */));
573            let mut prev = use::state_null::<Optional<i32, Tick<_>, Bounded>>();
574
575            // Output the previous value (or -1 if none)
576            let output = prev.clone().unwrap_or(prev.location().singleton(q!(-1)));
577            // Store the current batch's first value for next tick
578            prev = batch.first();
579            output.into_stream()
580        }
581        .sim_output();
582
583        flow.sim().exhaustive(async || {
584            input_send.send(10);
585            // First tick: prev is None, so output is -1
586            assert_eq!(out_recv.next().await, -1);
587
588            input_send.send(20);
589            // Second tick: prev is Some(10), so output is 10
590            assert_eq!(out_recv.next().await, 10);
591
592            input_send.send(30);
593            // Third tick: prev is Some(20), so output is 20
594            assert_eq!(out_recv.next().await, 20);
595        });
596    }
597
598    /// Test `use::state` with `source_iter` to initialize a stream state.
599    /// On the first tick, the state is the initial `[10, 20]` from `source_iter`.
600    /// On subsequent ticks, the state is the batch from the previous tick.
601    #[test]
602    fn sim_state_source_iter() {
603        let mut flow = FlowBuilder::new();
604        let node = flow.process::<()>();
605
606        let (input_send, input) = node.sim_input::<i32, _, _>();
607
608        let out_recv = sliced! {
609            let batch = use::batch(input, nondet!(/** test */));
610            let mut items = use::state(|l| l.source_iter(q!([10, 20])));
611
612            // Output the current state, then replace it with the batch
613            let output = items.clone();
614            items = batch;
615            output
616        }
617        .sim_output();
618
619        flow.sim().exhaustive(async || {
620            input_send.send(3);
621            // First tick: items = initial [10, 20], output = [10, 20]
622            let mut results = vec![];
623            results.push(out_recv.next().await);
624            results.push(out_recv.next().await);
625            results.sort();
626            assert_eq!(results, vec![10, 20]);
627
628            input_send.send(4);
629            // Second tick: items = [3] (from previous batch), output = [3]
630            assert_eq!(out_recv.next().await, 3);
631
632            input_send.send(5);
633            // Third tick: items = [4] (from previous batch), output = [4]
634            assert_eq!(out_recv.next().await, 4);
635        });
636    }
637
638    /// Test atomic slicing with keyed streams.
639    #[test]
640    fn sim_sliced_atomic_keyed_stream() {
641        let mut flow = FlowBuilder::new();
642        let node = flow.process::<()>();
643
644        let (input_send, input) = node.sim_input::<(i32, i32), _, _>();
645        let atomic_keyed_input = input.into_keyed().atomic();
646        let accumulated_inputs = atomic_keyed_input
647            .clone()
648            .assume_ordering(nondet!(/** Test */))
649            .fold(
650                q!(|| 0),
651                q!(|curr, new| {
652                    *curr += new;
653                }),
654            );
655
656        let out_recv = sliced! {
657            let atomic_keyed_input = use::atomic(atomic_keyed_input, nondet!(/** test */));
658            let accumulated_inputs = use::atomic(accumulated_inputs, nondet!(/** test */));
659            accumulated_inputs.join_keyed_stream(atomic_keyed_input)
660                .map(q!(|(sum, _input)| sum))
661                .entries()
662        }
663        .assume_ordering_trusted(nondet!(/** test */))
664        .sim_output();
665
666        flow.sim().exhaustive(async || {
667            input_send.send((1, 1));
668            assert_eq!(out_recv.next().await, (1, 1));
669
670            input_send.send((1, 2));
671            assert_eq!(out_recv.next().await, (1, 3));
672
673            input_send.send((2, 1));
674            assert_eq!(out_recv.next().await, (2, 1));
675
676            input_send.send((1, 3));
677            assert_eq!(out_recv.next().await, (1, 6));
678        });
679    }
680}
hydro_lang/live_collections/sliced/mod.rs

hydro_lang/live_collections/sliced/
mod.rs
