hydro_lang/
sim_hooks.rs

1//! Handle types for **simulator hooks**: scripting the decisions of unsafe operators.
2//!
3//! Every unsafe operator (like [`Stream::batch`](crate::live_collections::Stream::batch) or
4//! [`Singleton::snapshot`](crate::live_collections::Singleton::snapshot)) takes a
5//! [`NonDet`](crate::nondet::NonDet) guard. A guard can optionally carry a **hook handle**,
6//! which lets a simulation test take manual control of the non-deterministic decision made
7//! by that operator (which elements form the next batch, which version of a piece of state
8//! a snapshot reveals, ...). See `hydro_lang::sim::hooks` for the test-side scripting API.
9//!
10//! Handles are created from the [`FlowBuilder`](crate::compile::builder::FlowBuilder) via
11//! [`FlowBuilder::sim_hook`](crate::compile::builder::FlowBuilder::sim_hook) *before* the
12//! program under test is constructed, and attached to the operator they control with the
13//! `nondet!(... hook = handle)` syntax. Handles are small and `Copy`: the same value is
14//! passed into the program during construction and used later inside the test body to
15//! script decisions.
16//!
17//! # Hook scopes
18//!
19//! Every handle type carries a **scope** parameter naming the kind of root location the
20//! hooked operator runs on, mirroring
21//! [`Location::SimHookScope`](crate::location::Location::SimHookScope):
22//!
23//! - [`OnProcess<P>`] (the default): the operator has one instance, scripted directly.
24//! - [`OnCluster<C>`]: every cluster member runs its own instance of the operator;
25//!   select the one to script with [`.on(member_id)`](BatchHook::on), which yields an
26//!   [`OnMember<C>`]-scoped handle.
27//!
28//! Operators name the scope in their `NonDet` payload type, so scope mismatches fail to
29//! compile.
30//!
31//! This module contains only the handle types themselves (plain data), so components can
32//! expose hookable signatures (e.g. `nondet_batch: NonDet<Option<BatchHook<u32>>>`, passed
33//! directly to the `batch` operator it controls) without pulling
34//! in any simulator machinery; binding a hook in a flow that is *deployed* rather than
35//! simulated is harmless metadata that non-simulator backends ignore.
36
37use std::hash::Hash;
38use std::marker::PhantomData;
39
40use serde::Serialize;
41use serde::de::DeserializeOwned;
42
43use crate::live_collections::boundedness::{Boundedness, Unbounded};
44use crate::live_collections::stream::{
45    AtLeastOnce, ExactlyOnce, NoOrder, Ordering, Retries, TotalOrder,
46};
47
48/// A simulator hook handle (or a set of them) that can be created in one call to
49/// [`FlowBuilder::sim_hook`](crate::compile::builder::FlowBuilder::sim_hook).
50///
51/// Individual handle types implement this trait, and a struct of handles (a component's
52/// "testing interface") can implement it by creating every field. Fields are typed
53/// `Option<...>` so the struct doubles as a composite hook payload: its [`Default`]
54/// ("no hooks") is what a plain `nondet!(...)` guard carries, while `flow.sim_hook()`
55/// fills in every handle:
56///
57/// ```rust,ignore
58/// #[derive(Clone, Copy, Default)]
59/// pub struct CounterHooks {
60///     pub batch: Option<BatchHook<u32>>,
61///     pub snapshot: Option<SnapshotHook<u64>>,
62/// }
63///
64/// impl SimHook for CounterHooks {
65///     fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
66///         CounterHooks {
67///             batch: SimHook::create(next_id),
68///             snapshot: SimHook::create(next_id),
69///         }
70///     }
71/// }
72/// ```
73///
74/// Such structs nest (a field can itself be a struct of handles), and since handles are
75/// `Copy` a test can pass the struct around or destructure it freely.
76///
77/// Each handle's [`SimHook`] impl carries the trait bounds that the *scripted simulation
78/// codegen* for its operator kind requires (serde round-tripping for decisions, equality
79/// for value-naming decisions, `Hash + Eq + Clone` keys for keyed buffers). Since handles
80/// can only be created through this trait, binding a hook to an operator over unsupported
81/// types fails at the `flow.sim_hook()` call — an ordinary compile error in the test crate
82/// — instead of surfacing as a rustc failure inside the generated simulation dylib.
83pub trait SimHook {
84    /// Creates every handle in this value, allocating fresh IDs via `next_id`.
85    fn create(next_id: &mut dyn FnMut() -> usize) -> Self;
86}
87
88impl<H: SimHook> SimHook for Option<H> {
89    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
90        Some(H::create(next_id))
91    }
92}
93
94/// Hook scope marker: the hooked operator runs on a
95/// [`Process`](crate::location::Process) with tag `P` and has one instance, which the
96/// handle scripts directly.
97///
98/// This is the default scope of every handle type. See [the module docs](self#hook-scopes).
99pub struct OnProcess<P = ()> {
100    _phantom: PhantomData<fn(P)>,
101}
102
103/// Hook scope marker: the hooked operator runs on a
104/// [`Cluster`](crate::location::Cluster) with tag `C`, so every member runs its own
105/// independent instance of the operator.
106///
107/// Select the instance to script with [`.on(member_id)`](BatchHook::on), which yields an
108/// [`OnMember`]-scoped handle. Each member's instance makes its own decisions: a member
109/// with buffered input needs its own decision or pause.
110pub struct OnCluster<C = ()> {
111    _phantom: PhantomData<fn(C)>,
112}
113
114/// Hook scope marker: one selected member's instance of an operator running on a
115/// [`Cluster`](crate::location::Cluster) with tag `C`, produced by
116/// [`.on(member_id)`](BatchHook::on).
117///
118/// Member-scoped handles script decisions like process-scoped ones, but cannot be
119/// created or bound: operators are always bound through the unscoped [`OnCluster`]
120/// handle.
121pub struct OnMember<C = ()> {
122    _phantom: PhantomData<fn(C)>,
123}
124
125/// Hook scopes a handle can be created and bound with: [`OnProcess`] and [`OnCluster`].
126/// Operator signatures select the scope through
127/// [`Location::SimHookScope`](crate::location::Location::SimHookScope).
128#[diagnostic::on_unimplemented(
129    message = "`{Self}` is not a scope that sim hook handles can be created with",
130    note = "handles are created with the `OnProcess<P>` or `OnCluster<C>` scope of the operator they will be bound to; `OnMember` handles only arise from `.on(member_id)` at scripting time"
131)]
132#[sealed::sealed]
133pub trait BindableHookScope {}
134#[sealed::sealed]
135impl<P> BindableHookScope for OnProcess<P> {}
136#[sealed::sealed]
137impl<C> BindableHookScope for OnCluster<C> {}
138
139/// Hook scopes that name a single instance of the hooked operator and can therefore
140/// script decisions and pauses: [`OnProcess`] and [`OnMember`]. An [`OnCluster`]-scoped
141/// handle must first select a member with [`.on(member_id)`](BatchHook::on).
142#[diagnostic::on_unimplemented(
143    message = "a `{Self}`-scoped sim hook handle cannot script decisions",
144    note = "a hook bound to an operator running on a cluster has one independent instance per member; select the member to script with `.on(member_id)`"
145)]
146#[sealed::sealed]
147pub trait ScriptableHookScope {}
148#[sealed::sealed]
149impl<P> ScriptableHookScope for OnProcess<P> {}
150#[sealed::sealed]
151impl<C> ScriptableHookScope for OnMember<C> {}
152
153/// Generates the `.on(member_id)` member-selection method on [`OnCluster`]-scoped
154/// handles, shared by every handle type.
155macro_rules! on_member_method {
156    ($handle:ident < $($param:ident),* >) => {
157        /// Selects one cluster member's instance of the hooked operator, returning an
158        /// [`OnMember`]-scoped handle with the full scripting API.
159        ///
160        /// Each member's instance is scripted independently: `handle.on(0).release(2)`
161        /// scripts member 0's next batch and says nothing about the other members, each
162        /// of which needs its own decision (or pause) when it holds buffered input. A
163        /// member ID outside the cluster's sizing is reported when the scripting call
164        /// is awaited.
165        pub fn on(&self, member_id: u32) -> $handle<$($param,)* OnMember<C>> {
166            $handle {
167                id: self.id,
168                member: Some(member_id),
169                _phantom: PhantomData,
170            }
171        }
172    };
173}
174
175/// A hook handle controlling a `batch` operator over a stream of `T` elements with ordering
176/// `O` and retry guarantee `R` (mirroring the type of the stream being batched). `S` is the
177/// handle's [scope](self#hook-scopes).
178///
179/// A decision for a batch hook says which buffered elements form the next batch released
180/// into the tick. See `hydro_lang::sim::hooks` for the decisions offered.
181pub struct BatchHook<T, O: Ordering = TotalOrder, R: Retries = ExactlyOnce, S = OnProcess> {
182    pub(crate) id: usize,
183    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
184    #[cfg_attr(
185        not(feature = "sim"),
186        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
187    )]
188    pub(crate) member: Option<u32>,
189    pub(crate) _phantom: PhantomData<fn(T, O, R, S)>,
190}
191
192impl<T, O: Ordering, R: Retries, C> BatchHook<T, O, R, OnCluster<C>> {
193    on_member_method!(BatchHook<T, O, R>);
194}
195
196impl<T, O: Ordering, R: Retries, S> Clone for BatchHook<T, O, R, S> {
197    fn clone(&self) -> Self {
198        *self
199    }
200}
201impl<T, O: Ordering, R: Retries, S> Copy for BatchHook<T, O, R, S> {}
202
203impl<T, O: Ordering, R: Retries, S> std::fmt::Debug for BatchHook<T, O, R, S> {
204    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
205        f.debug_struct("BatchHook").field("id", &self.id).finish()
206    }
207}
208
209impl<T, O: Ordering, R: Retries, S: BindableHookScope> SimHook for BatchHook<T, O, R, S>
210where
211    T: Serialize + DeserializeOwned + PartialEq,
212{
213    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
214        BatchHook {
215            id: next_id(),
216            member: None,
217            _phantom: PhantomData,
218        }
219    }
220}
221
222/// A hook handle controlling a `snapshot` operator over a singleton of `T`. `S` is the
223/// handle's [scope](self#hook-scopes).
224///
225/// A decision for a snapshot hook picks which buffered version of the state the next tick
226/// execution observes. See `hydro_lang::sim::hooks` for the decisions offered.
227pub struct SnapshotHook<T, S = OnProcess> {
228    pub(crate) id: usize,
229    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
230    #[cfg_attr(
231        not(feature = "sim"),
232        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
233    )]
234    pub(crate) member: Option<u32>,
235    pub(crate) _phantom: PhantomData<fn(T, S)>,
236}
237
238impl<T, C> SnapshotHook<T, OnCluster<C>> {
239    on_member_method!(SnapshotHook<T>);
240}
241
242impl<T, S> Clone for SnapshotHook<T, S> {
243    fn clone(&self) -> Self {
244        *self
245    }
246}
247impl<T, S> Copy for SnapshotHook<T, S> {}
248
249impl<T, S> std::fmt::Debug for SnapshotHook<T, S> {
250    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
251        f.debug_struct("SnapshotHook")
252            .field("id", &self.id)
253            .finish()
254    }
255}
256
257impl<T, S: BindableHookScope> SimHook for SnapshotHook<T, S>
258where
259    T: Clone + PartialEq + Serialize + DeserializeOwned,
260{
261    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
262        SnapshotHook {
263            id: next_id(),
264            member: None,
265            _phantom: PhantomData,
266        }
267    }
268}
269
270/// A hook handle controlling an `assume_ordering` operator over `T` elements, with retry
271/// guarantee `R` (mirroring the type of the stream whose ordering is assumed). `S` is the
272/// handle's [scope](self#hook-scopes).
273///
274/// For an `ExactlyOnce` stream, a top-level decision selects the next buffered element to
275/// release, and an `assume_ordering` inside a tick instead takes one exhaustive ordering
276/// of that tick's complete input. For an `AtLeastOnce` stream, ordering additionally
277/// decides which *slots* each element's retries occupy, so top-level decisions split into
278/// `emit` (release a slot, keep the element for re-emission) and `emit_final` (release
279/// the element's last slot), and the in-tick ordering may emit each element into several
280/// slots. See `hydro_lang::sim::hooks` for the decisions offered.
281pub struct OrderingHook<T, B: Boundedness = Unbounded, R: Retries = ExactlyOnce, S = OnProcess> {
282    pub(crate) id: usize,
283    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
284    #[cfg_attr(
285        not(feature = "sim"),
286        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
287    )]
288    pub(crate) member: Option<u32>,
289    pub(crate) _phantom: PhantomData<fn(T, B, R, S)>,
290}
291
292impl<T, B: Boundedness, R: Retries, C> OrderingHook<T, B, R, OnCluster<C>> {
293    on_member_method!(OrderingHook<T, B, R>);
294}
295
296impl<T, B: Boundedness, R: Retries, S> Clone for OrderingHook<T, B, R, S> {
297    fn clone(&self) -> Self {
298        *self
299    }
300}
301impl<T, B: Boundedness, R: Retries, S> Copy for OrderingHook<T, B, R, S> {}
302
303impl<T, B: Boundedness, R: Retries, S> std::fmt::Debug for OrderingHook<T, B, R, S> {
304    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
305        f.debug_struct("OrderingHook")
306            .field("id", &self.id)
307            .finish()
308    }
309}
310
311impl<T, B: Boundedness, S: BindableHookScope> SimHook for OrderingHook<T, B, ExactlyOnce, S>
312where
313    T: Serialize + DeserializeOwned + PartialEq,
314{
315    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
316        OrderingHook {
317            id: next_id(),
318            member: None,
319            _phantom: PhantomData,
320        }
321    }
322}
323
324impl<T, B: Boundedness, S: BindableHookScope> SimHook for OrderingHook<T, B, AtLeastOnce, S>
325where
326    T: Clone + Serialize + DeserializeOwned + PartialEq,
327{
328    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
329        OrderingHook {
330            id: next_id(),
331            member: None,
332            _phantom: PhantomData,
333        }
334    }
335}
336
337/// A hook handle controlling an `assume_retries` operator over a stream of `T` elements
338/// with ordering `O` (mirroring the type of the stream whose retries are assumed). `S`
339/// is the handle's [scope](self#hook-scopes).
340///
341/// A decision for a retries hook says how many times a buffered element is released —
342/// the point where the simulator injects the duplicates the `AtLeastOnce` type says
343/// downstream must tolerate. Because every element admits arbitrarily many retries, the
344/// decision space is infinite and this non-determinism can never be explored
345/// autonomously: an `assume_retries` under simulation **must** be bound to a hook. See
346/// `hydro_lang::sim::hooks` for the decisions offered.
347pub struct RetriesHook<T, O: Ordering = NoOrder, B: Boundedness = Unbounded, S = OnProcess> {
348    pub(crate) id: usize,
349    /// The selected cluster member (set by `.on(member_id)` on an [`OnCluster`]-scoped
350    /// handle); `None` for process-scoped handles. Only read by the simulator's
351    /// scripting machinery (`hydro_lang::sim::hooks`), which is feature-gated.
352    #[cfg_attr(
353        not(feature = "sim"),
354        expect(
355            dead_code,
356            reason = "read only by the `sim`-gated scripting machinery; handles are plain data usable without it"
357        )
358    )]
359    pub(crate) member: Option<u32>,
360    pub(crate) _phantom: PhantomData<fn(T, O, B, S)>,
361}
362
363impl<T, O: Ordering, B: Boundedness, C> RetriesHook<T, O, B, OnCluster<C>> {
364    on_member_method!(RetriesHook<T, O, B>);
365}
366
367impl<T, O: Ordering, B: Boundedness, S> Clone for RetriesHook<T, O, B, S> {
368    fn clone(&self) -> Self {
369        *self
370    }
371}
372impl<T, O: Ordering, B: Boundedness, S> Copy for RetriesHook<T, O, B, S> {}
373
374impl<T, O: Ordering, B: Boundedness, S> std::fmt::Debug for RetriesHook<T, O, B, S> {
375    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
376        f.debug_struct("RetriesHook").field("id", &self.id).finish()
377    }
378}
379
380impl<T, O: Ordering, B: Boundedness, S: BindableHookScope> SimHook for RetriesHook<T, O, B, S>
381where
382    T: Clone + Serialize + DeserializeOwned + PartialEq,
383{
384    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
385        RetriesHook {
386            id: next_id(),
387            member: None,
388            _phantom: PhantomData,
389        }
390    }
391}
392
393/// A hook handle controlling a `batch` operator over a keyed stream with keys `K`, values
394/// `V`, per-key value ordering `O`, and retry guarantee `R` (mirroring the type of the
395/// keyed stream being batched). `S` is the handle's [scope](self#hook-scopes).
396///
397/// A `batch` over a keyed singleton with bounded per-key values (where each entry is final
398/// once it appears, so a batch is the set of *new* entries) is controlled by a
399/// `KeyedBatchHook<K, V>` with the default `TotalOrder` / `ExactlyOnce` parameters.
400///
401/// A decision for a keyed batch hook says which buffered `(key, value)` entries form the
402/// next batch released into the tick. See `hydro_lang::sim::hooks` for the decisions
403/// offered.
404pub struct KeyedBatchHook<K, V, O: Ordering = TotalOrder, R: Retries = ExactlyOnce, S = OnProcess> {
405    pub(crate) id: usize,
406    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
407    #[cfg_attr(
408        not(feature = "sim"),
409        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
410    )]
411    pub(crate) member: Option<u32>,
412    pub(crate) _phantom: PhantomData<fn(K, V, O, R, S)>,
413}
414
415impl<K, V, O: Ordering, R: Retries, C> KeyedBatchHook<K, V, O, R, OnCluster<C>> {
416    on_member_method!(KeyedBatchHook<K, V, O, R>);
417}
418
419impl<K, V, O: Ordering, R: Retries, S> Clone for KeyedBatchHook<K, V, O, R, S> {
420    fn clone(&self) -> Self {
421        *self
422    }
423}
424impl<K, V, O: Ordering, R: Retries, S> Copy for KeyedBatchHook<K, V, O, R, S> {}
425
426impl<K, V, O: Ordering, R: Retries, S> std::fmt::Debug for KeyedBatchHook<K, V, O, R, S> {
427    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
428        f.debug_struct("KeyedBatchHook")
429            .field("id", &self.id)
430            .finish()
431    }
432}
433
434impl<K, V, O: Ordering, R: Retries, S: BindableHookScope> SimHook for KeyedBatchHook<K, V, O, R, S>
435where
436    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
437    V: Serialize + DeserializeOwned + PartialEq,
438{
439    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
440        KeyedBatchHook {
441            id: next_id(),
442            member: None,
443            _phantom: PhantomData,
444        }
445    }
446}
447
448/// A hook handle controlling a `snapshot` operator over a keyed singleton with keys `K` and
449/// values `V`. `S` is the handle's [scope](self#hook-scopes).
450///
451/// A decision for a keyed snapshot hook picks which buffered version of each key's state
452/// the next tick execution observes. See `hydro_lang::sim::hooks` for the decisions
453/// offered.
454pub struct KeyedSnapshotHook<K, V, S = OnProcess> {
455    pub(crate) id: usize,
456    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
457    #[cfg_attr(
458        not(feature = "sim"),
459        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
460    )]
461    pub(crate) member: Option<u32>,
462    pub(crate) _phantom: PhantomData<fn(K, V, S)>,
463}
464
465impl<K, V, C> KeyedSnapshotHook<K, V, OnCluster<C>> {
466    on_member_method!(KeyedSnapshotHook<K, V>);
467}
468
469impl<K, V, S> Clone for KeyedSnapshotHook<K, V, S> {
470    fn clone(&self) -> Self {
471        *self
472    }
473}
474impl<K, V, S> Copy for KeyedSnapshotHook<K, V, S> {}
475
476impl<K, V, S> std::fmt::Debug for KeyedSnapshotHook<K, V, S> {
477    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
478        f.debug_struct("KeyedSnapshotHook")
479            .field("id", &self.id)
480            .finish()
481    }
482}
483
484impl<K, V, S: BindableHookScope> SimHook for KeyedSnapshotHook<K, V, S>
485where
486    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
487    V: Clone + PartialEq + Serialize + DeserializeOwned,
488{
489    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
490        KeyedSnapshotHook {
491            id: next_id(),
492            member: None,
493            _phantom: PhantomData,
494        }
495    }
496}
497
498/// A hook handle controlling an `assume_ordering` operator over a keyed stream with keys
499/// `K` and values `V`. `S` is the handle's [scope](self#hook-scopes).
500///
501/// A top-level decision selects the next buffered `(key, value)` entry to release. An
502/// `assume_ordering` inside a tick instead takes one exhaustive per-key ordering of that
503/// tick's complete input. See `hydro_lang::sim::hooks` for the decisions offered.
504pub struct KeyedOrderingHook<K, V, B: Boundedness = Unbounded, S = OnProcess> {
505    pub(crate) id: usize,
506    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
507    #[cfg_attr(
508        not(feature = "sim"),
509        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
510    )]
511    pub(crate) member: Option<u32>,
512    pub(crate) _phantom: PhantomData<fn(K, V, B, S)>,
513}
514
515impl<K, V, B: Boundedness, C> KeyedOrderingHook<K, V, B, OnCluster<C>> {
516    on_member_method!(KeyedOrderingHook<K, V, B>);
517}
518
519impl<K, V, B: Boundedness, S> Clone for KeyedOrderingHook<K, V, B, S> {
520    fn clone(&self) -> Self {
521        *self
522    }
523}
524impl<K, V, B: Boundedness, S> Copy for KeyedOrderingHook<K, V, B, S> {}
525
526impl<K, V, B: Boundedness, S> std::fmt::Debug for KeyedOrderingHook<K, V, B, S> {
527    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
528        f.debug_struct("KeyedOrderingHook")
529            .field("id", &self.id)
530            .finish()
531    }
532}
533
534impl<K, V, B: Boundedness, S: BindableHookScope> SimHook for KeyedOrderingHook<K, V, B, S>
535where
536    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
537    V: Serialize + DeserializeOwned + PartialEq,
538{
539    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
540        KeyedOrderingHook {
541            id: next_id(),
542            member: None,
543            _phantom: PhantomData,
544        }
545    }
546}
547
548/// A hook handle controlling an `entries_partially_ordered` operator over a keyed stream
549/// with keys `K` and values `V`. `S` is the handle's [scope](self#hook-scopes).
550///
551/// The operator preserves the order of values within each key while interleaving across
552/// keys non-deterministically. A top-level decision releases the front entry of one key's
553/// buffer; inside a tick, a single decision supplies the complete interleaving. See
554/// `hydro_lang::sim::hooks` for the decisions offered.
555pub struct PartialOrderingHook<K, V, B: Boundedness = Unbounded, S = OnProcess> {
556    pub(crate) id: usize,
557    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
558    #[cfg_attr(
559        not(feature = "sim"),
560        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
561    )]
562    pub(crate) member: Option<u32>,
563    pub(crate) _phantom: PhantomData<fn(K, V, B, S)>,
564}
565
566impl<K, V, B: Boundedness, C> PartialOrderingHook<K, V, B, OnCluster<C>> {
567    on_member_method!(PartialOrderingHook<K, V, B>);
568}
569
570impl<K, V, B: Boundedness, S> Clone for PartialOrderingHook<K, V, B, S> {
571    fn clone(&self) -> Self {
572        *self
573    }
574}
575impl<K, V, B: Boundedness, S> Copy for PartialOrderingHook<K, V, B, S> {}
576
577impl<K, V, B: Boundedness, S> std::fmt::Debug for PartialOrderingHook<K, V, B, S> {
578    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
579        f.debug_struct("PartialOrderingHook")
580            .field("id", &self.id)
581            .finish()
582    }
583}
584
585impl<K, V, B: Boundedness, S: BindableHookScope> SimHook for PartialOrderingHook<K, V, B, S>
586where
587    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
588    V: Serialize + DeserializeOwned + PartialEq,
589{
590    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
591        PartialOrderingHook {
592            id: next_id(),
593            member: None,
594            _phantom: PhantomData,
595        }
596    }
597}
598
599/// A hook handle controlling a `merge_ordered` operator over streams of `T` elements. `S`
600/// is the handle's [scope](self#hook-scopes).
601///
602/// The operator preserves the order of each input while interleaving the two inputs
603/// non-deterministically. A top-level decision releases the front element of one input's
604/// buffer; inside a tick, a single decision supplies the complete interleaving. See
605/// `hydro_lang::sim::hooks` for the decisions offered.
606pub struct MergeOrderedHook<T, B: Boundedness = Unbounded, S = OnProcess> {
607    pub(crate) id: usize,
608    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
609    #[cfg_attr(
610        not(feature = "sim"),
611        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
612    )]
613    pub(crate) member: Option<u32>,
614    pub(crate) _phantom: PhantomData<fn(T, B, S)>,
615}
616
617impl<T, B: Boundedness, C> MergeOrderedHook<T, B, OnCluster<C>> {
618    on_member_method!(MergeOrderedHook<T, B>);
619}
620
621impl<T, B: Boundedness, S> Clone for MergeOrderedHook<T, B, S> {
622    fn clone(&self) -> Self {
623        *self
624    }
625}
626impl<T, B: Boundedness, S> Copy for MergeOrderedHook<T, B, S> {}
627
628impl<T, B: Boundedness, S> std::fmt::Debug for MergeOrderedHook<T, B, S> {
629    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
630        f.debug_struct("MergeOrderedHook")
631            .field("id", &self.id)
632            .finish()
633    }
634}
635
636impl<T, B: Boundedness, S: BindableHookScope> SimHook for MergeOrderedHook<T, B, S>
637where
638    T: Serialize + DeserializeOwned + PartialEq,
639{
640    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
641        MergeOrderedHook {
642            id: next_id(),
643            member: None,
644            _phantom: PhantomData,
645        }
646    }
647}
648
649/// A hook handle controlling a `merge_ordered` operator over keyed streams with keys `K`
650/// and values `V`. `S` is the handle's [scope](self#hook-scopes).
651///
652/// The operator preserves each input's order within every key while interleaving the two
653/// inputs non-deterministically (cross-key order is unconstrained). A top-level decision
654/// releases the front entry of one key's buffer in one input; inside a tick, a single
655/// decision supplies the complete interleaving. See `hydro_lang::sim::hooks` for the
656/// decisions offered.
657pub struct KeyedMergeOrderedHook<K, V, B: Boundedness = Unbounded, S = OnProcess> {
658    pub(crate) id: usize,
659    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
660    #[cfg_attr(
661        not(feature = "sim"),
662        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
663    )]
664    pub(crate) member: Option<u32>,
665    pub(crate) _phantom: PhantomData<fn(K, V, B, S)>,
666}
667
668impl<K, V, B: Boundedness, C> KeyedMergeOrderedHook<K, V, B, OnCluster<C>> {
669    on_member_method!(KeyedMergeOrderedHook<K, V, B>);
670}
671
672impl<K, V, B: Boundedness, S> Clone for KeyedMergeOrderedHook<K, V, B, S> {
673    fn clone(&self) -> Self {
674        *self
675    }
676}
677impl<K, V, B: Boundedness, S> Copy for KeyedMergeOrderedHook<K, V, B, S> {}
678
679impl<K, V, B: Boundedness, S> std::fmt::Debug for KeyedMergeOrderedHook<K, V, B, S> {
680    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
681        f.debug_struct("KeyedMergeOrderedHook")
682            .field("id", &self.id)
683            .finish()
684    }
685}
686
687impl<K, V, B: Boundedness, S: BindableHookScope> SimHook for KeyedMergeOrderedHook<K, V, B, S>
688where
689    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
690    V: Serialize + DeserializeOwned + PartialEq,
691{
692    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
693        KeyedMergeOrderedHook {
694            id: next_id(),
695            member: None,
696            _phantom: PhantomData,
697        }
698    }
699}
hydro_lang/sim_hooks.rs

hydro_lang/
sim_hooks.rs
