hydro_lang/
forward_handle.rs

1//! Mechanisms for introducing forward references and cycles in Hydro.
2
3use sealed::sealed;
4
5use crate::compile::builder::CycleId;
6use crate::location::Location;
7use crate::location::dynamic::LocationId;
8use crate::staging_util::Invariant;
9
10#[sealed]
11pub(crate) trait ReceiverKind {}
12
13/// Marks that the [`ForwardHandle`] is for a "forward reference" to a later-defined collection.
14///
15/// When the handle is completed, the provided collection must not depend _synchronously_
16/// (in the same tick) on the forward reference that was created earlier.
17pub enum ForwardRef {}
18
19#[sealed]
20impl ReceiverKind for ForwardRef {}
21
22/// Marks that the [`ForwardHandle`] will send a live collection to the next tick.
23///
24/// Dependency cycles are permitted for this handle type, because the collection used
25/// to complete this handle will appear on the source-side on the _next_ tick.
26pub enum TickCycle {}
27
28#[sealed]
29impl ReceiverKind for TickCycle {}
30
31pub(crate) trait ReceiverComplete<'a, Marker>
32where
33    Marker: ReceiverKind,
34{
35    fn complete(self, cycle_id: CycleId, expected_location: LocationId);
36}
37
38pub(crate) trait CycleCollection<'a, Kind>: ReceiverComplete<'a, Kind>
39where
40    Kind: ReceiverKind,
41{
42    type Location: Location<'a>;
43
44    fn create_source(id: CycleId, location: Self::Location) -> Self;
45}
46
47pub(crate) trait CycleCollectionWithInitial<'a, Kind>: ReceiverComplete<'a, Kind>
48where
49    Kind: ReceiverKind,
50{
51    type Location: Location<'a>;
52
53    fn location(&self) -> &Self::Location;
54
55    fn create_source_with_initial(
56        cycle_id: CycleId,
57        initial: Self,
58        location: Self::Location,
59    ) -> Self;
60}
61
62/// A handle that can be used to fulfill a forward reference.
63///
64/// The `C` type parameter specifies the collection type that can be used to complete the handle.
65#[expect(
66    private_bounds,
67    reason = "only Hydro collections can implement ReceiverComplete"
68)]
69pub struct ForwardHandle<'a, C: ReceiverComplete<'a, ForwardRef>> {
70    completed: bool,
71    cycle_id: CycleId,
72    expected_location: LocationId,
73    _phantom: Invariant<'a, C>,
74}
75
76#[expect(
77    private_bounds,
78    reason = "only Hydro collections can implement ReceiverComplete"
79)]
80impl<'a, C: ReceiverComplete<'a, ForwardRef>> ForwardHandle<'a, C> {
81    pub(crate) fn new(cycle_id: CycleId, expected_location: LocationId) -> Self {
82        Self {
83            completed: false,
84            cycle_id,
85            expected_location,
86            _phantom: std::marker::PhantomData,
87        }
88    }
89}
90
91impl<'a, C: ReceiverComplete<'a, ForwardRef>> Drop for ForwardHandle<'a, C> {
92    fn drop(&mut self) {
93        assert!(
94            self.completed || std::thread::panicking(),
95            "ForwardHandle dropped without being completed"
96        );
97    }
98}
99
100#[expect(
101    private_bounds,
102    reason = "only Hydro collections can implement ReceiverComplete"
103)]
104impl<'a, C: ReceiverComplete<'a, ForwardRef>> ForwardHandle<'a, C> {
105    /// Completes the forward reference with the given live collection. The initial forward reference
106    /// collection created in [`Location::forward_ref`] will resolve to this value.
107    ///
108    /// The provided value **must not** depend _synchronously_ (in the same tick) on the forward reference
109    /// collection, as doing so would create a dependency cycle. Asynchronous cycles (outside a tick) are
110    /// allowed, since the program can continue running while the cycle is processed.
111    pub fn complete(mut self, stream: impl Into<C>) {
112        self.completed = true;
113        C::complete(stream.into(), self.cycle_id, self.expected_location.clone())
114    }
115}
116
117/// A handle that can be used to complete a tick cycle by sending a collection to the next tick.
118///
119/// The `C` type parameter specifies the collection type that can be used to complete the handle.
120#[expect(
121    private_bounds,
122    reason = "only Hydro collections can implement ReceiverComplete"
123)]
124pub struct TickCycleHandle<'a, C: ReceiverComplete<'a, TickCycle>> {
125    completed: bool,
126    cycle_id: CycleId,
127    expected_location: LocationId,
128    _phantom: Invariant<'a, C>,
129}
130
131#[expect(
132    private_bounds,
133    reason = "only Hydro collections can implement ReceiverComplete"
134)]
135impl<'a, C: ReceiverComplete<'a, TickCycle>> TickCycleHandle<'a, C> {
136    pub(crate) fn new(cycle_id: CycleId, expected_location: LocationId) -> Self {
137        Self {
138            completed: false,
139            cycle_id,
140            expected_location,
141            _phantom: std::marker::PhantomData,
142        }
143    }
144}
145
146impl<'a, C: ReceiverComplete<'a, TickCycle>> Drop for TickCycleHandle<'a, C> {
147    fn drop(&mut self) {
148        assert!(
149            self.completed || std::thread::panicking(),
150            "TickCycleHandle dropped without being completed"
151        );
152    }
153}
154
155#[expect(
156    private_bounds,
157    reason = "only Hydro collections can implement ReceiverComplete"
158)]
159impl<'a, C: ReceiverComplete<'a, TickCycle>> TickCycleHandle<'a, C> {
160    /// Sends the provided collection to the next tick, where it will be materialized
161    /// in the collection returned by [`crate::location::Tick::cycle`] or
162    /// [`crate::location::Tick::cycle_with_initial`].
163    pub fn complete_next_tick(mut self, stream: impl Into<C>) {
164        self.completed = true;
165        C::complete(stream.into(), self.cycle_id, self.expected_location.clone())
166    }
167}
168
169/// A trait for completing a cycle handle with a state value.
170/// Used internally by the `sliced!` macro for state management.
171#[doc(hidden)]
172pub trait CompleteCycle<S> {
173    /// Completes the cycle with the given state value.
174    fn complete_next_tick(self, state: S);
175}
176
177impl<'a, C: ReceiverComplete<'a, TickCycle>> CompleteCycle<C> for TickCycleHandle<'a, C> {
178    fn complete_next_tick(self, state: C) {
179        TickCycleHandle::complete_next_tick(self, state)
180    }
181}
hydro_lang/forward_handle.rs

hydro_lang/
forward_handle.rs
