hydro_lang/location/
tick.rs

1//! Clock domains for batching streaming data into discrete time steps.
2//!
3//! In Hydro, a [`Tick`] represents a logical clock that can be used to batch
4//! unbounded streaming data into discrete, bounded time steps. This is essential
5//! for implementing iterative algorithms, synchronizing data across multiple
6//! streams, and performing aggregations over windows of data.
7//!
8//! A tick is created from a top-level location (such as [`super::Process`] or [`super::Cluster`])
9//! using [`Location::tick`]. Once inside a tick, bounded live collections can be
10//! manipulated with operations like fold, reduce, and cross-product, and the
11//! results can be emitted back to the unbounded stream using methods like
12//! `all_ticks()`.
13//!
14//! The [`Atomic`] wrapper provides atomicity guarantees within a tick, ensuring
15//! that reads and writes within a tick are serialized.
16
17use stageleft::{QuotedWithContext, q};
18
19#[cfg(stageleft_runtime)]
20use super::dynamic::DynLocation;
21use super::{Location, LocationId};
22use crate::compile::builder::{ClockId, FlowState};
23use crate::compile::ir::{HydroNode, HydroSource};
24#[cfg(stageleft_runtime)]
25use crate::forward_handle::{CycleCollection, CycleCollectionWithInitial};
26use crate::forward_handle::{TickCycle, TickCycleHandle};
27#[cfg(feature = "tokio")]
28use crate::live_collections::Singleton;
29use crate::live_collections::boundedness::Bounded;
30use crate::live_collections::optional::Optional;
31use crate::live_collections::stream::{ExactlyOnce, Stream, TotalOrder};
32use crate::location::TopLevel;
33#[cfg(feature = "tokio")]
34use crate::nondet::NonDet;
35use crate::nondet::nondet;
36
37/// A location wrapper that provides atomicity guarantees within a [`Tick`].
38///
39/// An `Atomic` context establishes a happens-before relationship between operations:
40/// - Downstream computations from `atomic()` are associated with an internal tick
41/// - Outputs from `end_atomic()` are held until all computations in the tick complete
42/// - Snapshots via `use::atomic` are guaranteed to reflect all updates from associated `end_atomic()`
43///
44/// This ensures read-after-write consistency: if a client receives an acknowledgement
45/// from `end_atomic()`, any subsequent `use::atomic` snapshot will include the effects
46/// of that acknowledged operation.
47#[derive(Clone)]
48pub struct Atomic<Loc> {
49    pub(crate) tick: Tick<Loc>,
50}
51
52impl<L: DynLocation> DynLocation for Atomic<L> {
53    fn dyn_id(&self) -> LocationId {
54        LocationId::Atomic(Box::new(self.tick.dyn_id()))
55    }
56
57    fn flow_state(&self) -> &FlowState {
58        self.tick.flow_state()
59    }
60
61    fn is_top_level() -> bool {
62        L::is_top_level()
63    }
64
65    fn multiversioned(&self) -> bool {
66        self.tick.multiversioned()
67    }
68
69    fn cluster_consistency() -> Option<super::dynamic::ClusterConsistency> {
70        L::cluster_consistency()
71    }
72}
73
74impl<'a, L> Location<'a> for Atomic<L>
75where
76    L: Location<'a>,
77{
78    type Root = L::Root;
79
80    type DropConsistency = Atomic<L::DropConsistency>;
81
82    fn consistency() -> Option<super::dynamic::ClusterConsistency> {
83        L::consistency()
84    }
85
86    fn root(&self) -> Self::Root {
87        self.tick.root()
88    }
89
90    fn drop_consistency(&self) -> Self::DropConsistency {
91        Atomic {
92            tick: self.tick.drop_consistency(),
93        }
94    }
95
96    fn from_drop_consistency(l2: Self::DropConsistency) -> Self {
97        Atomic {
98            tick: Tick::from_drop_consistency(l2.tick),
99        }
100    }
101}
102
103/// Trait for live collections that can be deferred by one tick.
104///
105/// When a collection implements `DeferTick`, calling `defer_tick` delays its
106/// values by one clock cycle. This is primarily used internally to implement
107/// tick-based cycles ([`Tick::cycle`]), ensuring that feedback loops advance
108/// by one tick to avoid infinite recursion within a single tick.
109pub trait DeferTick {
110    /// Returns a new collection whose values are delayed by one tick.
111    fn defer_tick(self) -> Self;
112}
113
114/// Marks the stream as being inside the single global clock domain.
115#[derive(Clone)]
116pub struct Tick<L> {
117    /// `None` if `l` is `Atomic`.
118    pub(crate) id: Option<ClockId>,
119    /// Location.
120    pub(crate) l: L,
121}
122
123impl<L: DynLocation> DynLocation for Tick<L> {
124    fn dyn_id(&self) -> LocationId {
125        LocationId::Tick {
126            tick: self.id,
127            parent_location: Box::new(self.l.dyn_id()),
128        }
129    }
130
131    fn flow_state(&self) -> &FlowState {
132        self.l.flow_state()
133    }
134
135    fn is_top_level() -> bool {
136        false
137    }
138
139    fn multiversioned(&self) -> bool {
140        self.l.multiversioned()
141    }
142
143    fn cluster_consistency() -> Option<super::dynamic::ClusterConsistency> {
144        L::cluster_consistency()
145    }
146}
147
148impl<'a, L> Location<'a> for Tick<L>
149where
150    L: Location<'a>,
151{
152    type Root = L::Root;
153
154    type DropConsistency = Tick<L::DropConsistency>;
155
156    fn consistency() -> Option<super::dynamic::ClusterConsistency> {
157        L::consistency()
158    }
159
160    fn root(&self) -> Self::Root {
161        self.l.root()
162    }
163
164    fn drop_consistency(&self) -> Self::DropConsistency {
165        Tick {
166            id: self.id,
167            l: self.l.drop_consistency(),
168        }
169    }
170
171    fn from_drop_consistency(l2: Self::DropConsistency) -> Self {
172        Tick {
173            id: l2.id,
174            l: L::from_drop_consistency(l2.l),
175        }
176    }
177}
178
179impl<'a, L> Tick<L>
180where
181    L: Location<'a>,
182{
183    /// Returns a reference to the parent location that this tick is located at.
184    ///
185    /// For example, if a `Tick` was created from a `Process`, this returns a reference
186    /// to that `Process`.
187    pub fn parent_location(&self) -> &L {
188        &self.l
189    }
190
191    /// Use [`Self::parent_location`] instead.
192    #[deprecated(note = "use `.parent_location()` instead")]
193    pub fn outer(&self) -> &L {
194        self.parent_location()
195    }
196
197    /// Creates a bounded stream of `()` values inside this tick, with a fixed batch size.
198    ///
199    /// This is useful for driving computations inside a tick that need to process
200    /// a specific number of elements per tick. Each tick will produce exactly
201    /// `batch_size` unit values.
202    pub fn spin_batch(
203        &self,
204        batch_size: impl QuotedWithContext<
205            'a,
206            usize,
207            crate::live_collections::OperatorContext<
208                L,
209                crate::live_collections::boundedness::Unbounded,
210            >,
211        > + Copy
212        + 'a,
213    ) -> Stream<(), Self, Bounded, TotalOrder, ExactlyOnce>
214    where
215        L: TopLevel<'a>,
216    {
217        let out = self
218            .l
219            .spin()
220            .flat_map_ordered(q!(move |_| 0..batch_size))
221            .map(q!(|_| ()));
222
223        let inner = out.batch(self, nondet!(/** at runtime, `spin` produces a single value per tick, so each batch is guaranteed to be the same size. */));
224        Stream::new(self.clone(), inner.ir_node.replace(HydroNode::Placeholder))
225    }
226
227    /// Creates an [`Optional`] which has a null value on every tick.
228    ///
229    /// # Example
230    /// ```rust
231    /// # #[cfg(feature = "deploy")] {
232    /// # use hydro_lang::prelude::*;
233    /// # use futures::StreamExt;
234    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
235    /// let tick = process.tick();
236    /// let optional = tick.none::<i32>();
237    /// optional.unwrap_or(tick.singleton(q!(123)))
238    /// # .all_ticks()
239    /// # }, |mut stream| async move {
240    /// // 123
241    /// # assert_eq!(stream.next().await.unwrap(), 123);
242    /// # }));
243    /// # }
244    /// ```
245    pub fn none<T>(&self) -> Optional<T, Self, Bounded> {
246        let e = q!([]);
247        let e = QuotedWithContext::<'a, [(); 0], Self>::splice_typed_ctx(e, self);
248
249        let unit_optional: Optional<(), Self, Bounded> = Optional::new(
250            self.clone(),
251            HydroNode::Source {
252                source: HydroSource::Iter(e.into()),
253                metadata: self.new_node_metadata(Optional::<(), Self, Bounded>::collection_kind()),
254            },
255        );
256
257        unit_optional.map(q!(|_| unreachable!())) // always empty
258    }
259
260    /// Creates an [`Optional`] which will have the provided static value on the first tick, and be
261    /// null on all subsequent ticks.
262    ///
263    /// This is useful for bootstrapping stateful computations which need an initial value.
264    ///
265    /// # Example
266    /// ```rust
267    /// # #[cfg(feature = "deploy")] {
268    /// # use hydro_lang::prelude::*;
269    /// # use futures::StreamExt;
270    /// # tokio_test::block_on(hydro_lang::test_util::stream_transform_test(|process| {
271    /// let tick = process.tick();
272    /// // ticks are lazy by default, forces the second tick to run
273    /// tick.spin_batch(q!(1)).all_ticks().for_each(q!(|_| {}));
274    /// let optional = tick.optional_first_tick(q!(5));
275    /// optional.unwrap_or(tick.singleton(q!(123))).all_ticks()
276    /// # }, |mut stream| async move {
277    /// // 5, 123, 123, 123, ...
278    /// # assert_eq!(stream.next().await.unwrap(), 5);
279    /// # assert_eq!(stream.next().await.unwrap(), 123);
280    /// # assert_eq!(stream.next().await.unwrap(), 123);
281    /// # assert_eq!(stream.next().await.unwrap(), 123);
282    /// # }));
283    /// # }
284    /// ```
285    pub fn optional_first_tick<T: Clone>(
286        &self,
287        e: impl QuotedWithContext<'a, T, Tick<L>>,
288    ) -> Optional<T, Self, Bounded> {
289        let e = e.splice_untyped_ctx(self);
290
291        Optional::new(
292            self.clone(),
293            HydroNode::SingletonSource {
294                value: e.into(),
295                first_tick_only: true,
296                metadata: self.new_node_metadata(Optional::<T, Self, Bounded>::collection_kind()),
297            },
298        )
299    }
300
301    /// Returns the current wall-clock time as a [`Singleton`] containing a
302    /// [`tokio::time::Instant`].
303    ///
304    /// # Non-Determinism
305    /// Reading wall-clock time is inherently non-deterministic because the
306    /// value depends on when the tick executes. A [`NonDet`] guard is required
307    /// to acknowledge this.
308    #[cfg(feature = "tokio")]
309    pub fn current_tick_instant(
310        &self,
311        _nondet: NonDet,
312    ) -> Singleton<tokio::time::Instant, Tick<L::DropConsistency>, Bounded>
313    where
314        Self: Sized,
315    {
316        // TODO(shadaj): this is a simulator hole, should be reported as unsupported until it is
317        self.singleton(q!(tokio::time::Instant::now()))
318    }
319
320    /// Creates a feedback cycle within this tick for implementing iterative computations.
321    ///
322    /// Returns a handle that must be completed with the actual collection, and a placeholder
323    /// collection that represents the output of the previous tick (deferred by one tick).
324    /// This is useful for implementing fixed-point computations where the output of one
325    /// tick feeds into the input of the next.
326    ///
327    /// The cycle automatically defers values by one tick to prevent infinite recursion.
328    #[expect(
329        private_bounds,
330        reason = "only Hydro collections can implement ReceiverComplete"
331    )]
332    pub fn cycle<S, L2: Location<'a, DropConsistency = Tick<L::DropConsistency>>>(
333        &self,
334    ) -> (TickCycleHandle<'a, S>, S)
335    where
336        S: CycleCollection<'a, TickCycle, Location = L2> + DeferTick,
337    {
338        let cycle_id = self.flow_state().borrow_mut().next_cycle_id();
339        (
340            TickCycleHandle::new(cycle_id, Location::id(self)),
341            S::create_source(cycle_id, self.clone().with_consistency_of()).defer_tick(),
342        )
343    }
344
345    /// Creates a feedback cycle with an initial value for the first tick.
346    ///
347    /// Similar to [`Tick::cycle`], but allows providing an initial collection
348    /// that will be used as the value on the first tick before any feedback
349    /// is available. This is useful for bootstrapping iterative computations
350    /// that need a starting state.
351    #[expect(
352        private_bounds,
353        reason = "only Hydro collections can implement ReceiverComplete"
354    )]
355    pub fn cycle_with_initial<S, L2: Location<'a, DropConsistency = Tick<L::DropConsistency>>>(
356        &self,
357        initial: S,
358    ) -> (TickCycleHandle<'a, S>, S)
359    where
360        S: CycleCollectionWithInitial<'a, TickCycle, Location = L2>,
361    {
362        let cycle_id = self.flow_state().borrow_mut().next_cycle_id();
363        (
364            TickCycleHandle::new(cycle_id, Location::id(self)),
365            // no need to defer_tick, create_source_with_initial does it for us
366            S::create_source_with_initial(cycle_id, initial, self.clone().with_consistency_of()),
367        )
368    }
369}
370
371#[cfg(test)]
372mod tests {
373    #[cfg(feature = "sim")]
374    use stageleft::q;
375
376    #[cfg(feature = "sim")]
377    use crate::live_collections::sliced::sliced;
378    #[cfg(feature = "sim")]
379    use crate::location::Location;
380    #[cfg(feature = "sim")]
381    use crate::nondet::nondet;
382    #[cfg(feature = "sim")]
383    use crate::prelude::FlowBuilder;
384
385    #[cfg(feature = "sim")]
386    #[test]
387    fn sim_atomic_stream() {
388        let mut flow = FlowBuilder::new();
389        let node = flow.process::<()>();
390
391        let (write_send, write_req) = node.sim_input();
392        let (read_send, read_req) = node.sim_input::<(), _, _>();
393
394        let atomic_write = write_req.atomic();
395        let current_state = atomic_write.clone().fold(
396            q!(|| 0),
397            q!(|state: &mut i32, v: i32| {
398                *state += v;
399            }),
400        );
401
402        let write_ack_recv = atomic_write.end_atomic().sim_output();
403        let read_response_recv = sliced! {
404            let batch_of_req = use::batch(read_req, nondet!(/** test */));
405            let latest_singleton = use::atomic(current_state, nondet!(/** test */));
406            batch_of_req.cross_singleton(latest_singleton)
407        }
408        .sim_output();
409
410        let sim_compiled = flow.sim().compiled();
411        let instances = sim_compiled.exhaustive(async || {
412            write_send.send(1);
413            write_ack_recv.assert_yields([1]).await;
414            read_send.send(());
415            assert!(read_response_recv.next().await.1 >= 1);
416        });
417
418        assert_eq!(instances, 1);
419
420        let instances_read_before_write = sim_compiled.exhaustive(async || {
421            write_send.send(1);
422            read_send.send(());
423            write_ack_recv.assert_yields([1]).await;
424            let _ = read_response_recv.next().await;
425        });
426
427        assert_eq!(instances_read_before_write, 3); // read before write, write before read, both in same tick
428    }
429
430    #[cfg(feature = "sim")]
431    #[test]
432    #[should_panic]
433    fn sim_non_atomic_stream() {
434        // shows that atomic is necessary
435        let mut flow = FlowBuilder::new();
436        let node = flow.process::<()>();
437
438        let (write_send, write_req) = node.sim_input();
439        let (read_send, read_req) = node.sim_input::<(), _, _>();
440
441        let current_state = write_req.clone().fold(
442            q!(|| 0),
443            q!(|state: &mut i32, v: i32| {
444                *state += v;
445            }),
446        );
447
448        let write_ack_recv = write_req.sim_output();
449
450        let read_response_recv = sliced! {
451            let batch_of_req = use::batch(read_req, nondet!(/** test */));
452            let latest_singleton = use::snapshot(current_state, nondet!(/** test */));
453            batch_of_req.cross_singleton(latest_singleton)
454        }
455        .sim_output();
456
457        flow.sim().exhaustive(async || {
458            write_send.send(1);
459            write_ack_recv.assert_yields([1]).await;
460            read_send.send(());
461
462            let (_, v) = read_response_recv.next().await;
463            assert_eq!(v, 1);
464        });
465    }
466}
hydro_lang/location/tick.rs

hydro_lang/location/
tick.rs
