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

hydro_lang/location/
tick.rs
