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::unhooked(),
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::Optional<T, L, crate::live_collections::optional::InitNone>;
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 =
492        super::Optional<T, crate::location::Atomic<L>, crate::live_collections::optional::InitNone>;
493
494    fn unslice(self) -> Self::Unsliced {
495        self.collection.latest_atomic()
496    }
497}
498
499impl<'a, T, L: Location<'a>> Unslicable for style::Atomic<super::Optional<T, Tick<L>, Bounded>> {
500    type Unsliced = super::Optional<T, crate::location::Atomic<L>, Unbounded>;
501
502    fn unslice(self) -> Self::Unsliced {
503        self.collection.latest_atomic()
504    }
505}
506
507impl<'a, K, V, L: Location<'a>, O: Ordering, R: Retries> Unslicable
508    for style::Atomic<super::KeyedStream<K, V, Tick<L>, Bounded, O, R>>
509{
510    type Unsliced = super::KeyedStream<K, V, crate::location::Atomic<L>, Unbounded, O, R>;
511
512    fn unslice(self) -> Self::Unsliced {
513        self.collection.all_ticks_atomic()
514    }
515}
516
517#[cfg(feature = "sim")]
518#[cfg(test)]
519mod tests {
520    use stageleft::q;
521
522    use super::sliced;
523    use crate::location::Location;
524    use crate::nondet::nondet;
525    use crate::prelude::FlowBuilder;
526
527    /// Test a counter using `use::state` with an initial singleton value.
528    /// Each input increments the counter, and we verify the output after each tick.
529    #[test]
530    fn sim_state_counter() {
531        let mut flow = FlowBuilder::new();
532        let node = flow.process::<()>();
533
534        let (input_send, input) = node.sim_input::<i32, _, _>();
535
536        let out_recv = sliced! {
537            let batch = use::batch(input, nondet!(/** test */));
538            let mut counter = use::state(|l| l.singleton(q!(0)));
539
540            let new_count = counter.clone().zip(batch.count())
541                .map(q!(|(old, add)| old + add));
542            counter = new_count.clone();
543            new_count.into_stream()
544        }
545        .sim_output();
546
547        flow.sim().exhaustive(async || {
548            input_send.send(1);
549            assert_eq!(out_recv.next().await, 1);
550
551            input_send.send(1);
552            assert_eq!(out_recv.next().await, 2);
553
554            input_send.send(1);
555            assert_eq!(out_recv.next().await, 3);
556        });
557    }
558
559    /// Test `use::state_null` with an Optional that starts as None.
560    #[cfg(feature = "sim")]
561    #[test]
562    fn sim_state_null_optional() {
563        use crate::live_collections::Optional;
564        use crate::live_collections::boundedness::Bounded;
565        use crate::location::{Location, Tick};
566
567        let mut flow = FlowBuilder::new();
568        let node = flow.process::<()>();
569
570        let (input_send, input) = node.sim_input::<i32, _, _>();
571
572        let out_recv = sliced! {
573            let batch = use::batch(input, nondet!(/** test */));
574            let mut prev = use::state_null::<Optional<i32, Tick<_>, Bounded>>();
575
576            // Output the previous value (or -1 if none)
577            let output = prev.clone().unwrap_or(prev.location().singleton(q!(-1)));
578            // Store the current batch's first value for next tick
579            prev = batch.first();
580            output.into_stream()
581        }
582        .sim_output();
583
584        flow.sim().exhaustive(async || {
585            input_send.send(10);
586            // First tick: prev is None, so output is -1
587            assert_eq!(out_recv.next().await, -1);
588
589            input_send.send(20);
590            // Second tick: prev is Some(10), so output is 10
591            assert_eq!(out_recv.next().await, 10);
592
593            input_send.send(30);
594            // Third tick: prev is Some(20), so output is 20
595            assert_eq!(out_recv.next().await, 20);
596        });
597    }
598
599    /// Test `use::state` with `source_iter` to initialize a stream state.
600    /// On the first tick, the state is the initial `[10, 20]` from `source_iter`.
601    /// On subsequent ticks, the state is the batch from the previous tick.
602    #[test]
603    fn sim_state_source_iter() {
604        let mut flow = FlowBuilder::new();
605        let node = flow.process::<()>();
606
607        let (input_send, input) = node.sim_input::<i32, _, _>();
608
609        let out_recv = sliced! {
610            let batch = use::batch(input, nondet!(/** test */));
611            let mut items = use::state(|l| l.source_iter(q!([10, 20])));
612
613            // Output the current state, then replace it with the batch
614            let output = items.clone();
615            items = batch;
616            output
617        }
618        .sim_output();
619
620        flow.sim().exhaustive(async || {
621            input_send.send(3);
622            // First tick: items = initial [10, 20], output = [10, 20]
623            let mut results = vec![];
624            results.push(out_recv.next().await);
625            results.push(out_recv.next().await);
626            results.sort();
627            assert_eq!(results, vec![10, 20]);
628
629            input_send.send(4);
630            // Second tick: items = [3] (from previous batch), output = [3]
631            assert_eq!(out_recv.next().await, 3);
632
633            input_send.send(5);
634            // Third tick: items = [4] (from previous batch), output = [4]
635            assert_eq!(out_recv.next().await, 4);
636        });
637    }
638
639    /// Test atomic slicing with keyed streams.
640    #[test]
641    fn sim_sliced_atomic_keyed_stream() {
642        let mut flow = FlowBuilder::new();
643        let node = flow.process::<()>();
644
645        let (input_send, input) = node.sim_input::<(i32, i32), _, _>();
646        let atomic_keyed_input = input.into_keyed().atomic();
647        let accumulated_inputs = atomic_keyed_input
648            .clone()
649            .assume_ordering(nondet!(/** Test */))
650            .fold(
651                q!(|| 0),
652                q!(|curr, new| {
653                    *curr += new;
654                }),
655            );
656
657        let out_recv = sliced! {
658            let atomic_keyed_input = use::atomic(atomic_keyed_input, nondet!(/** test */));
659            let accumulated_inputs = use::atomic(accumulated_inputs, nondet!(/** test */));
660            accumulated_inputs.join_keyed_stream(atomic_keyed_input)
661                .map(q!(|(sum, _input)| sum))
662                .entries()
663        }
664        .assume_ordering_trusted(nondet!(/** test */))
665        .sim_output();
666
667        flow.sim().exhaustive(async || {
668            input_send.send((1, 1));
669            assert_eq!(out_recv.next().await, (1, 1));
670
671            input_send.send((1, 2));
672            assert_eq!(out_recv.next().await, (1, 3));
673
674            input_send.send((2, 1));
675            assert_eq!(out_recv.next().await, (2, 1));
676
677            input_send.send((1, 3));
678            assert_eq!(out_recv.next().await, (1, 6));
679        });
680    }
681}
hydro_lang/live_collections/sliced/mod.rs

hydro_lang/live_collections/sliced/
mod.rs
