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::{ExactlyOnce, Ordering, Retries, TotalOrder};
45
46/// A simulator hook handle (or a set of them) that can be created in one call to
47/// [`FlowBuilder::sim_hook`](crate::compile::builder::FlowBuilder::sim_hook).
48///
49/// Individual handle types implement this trait, and a struct of handles (a component's
50/// "testing interface") can implement it by creating every field. Fields are typed
51/// `Option<...>` so the struct doubles as a composite hook payload: its [`Default`]
52/// ("no hooks") is what a plain `nondet!(...)` guard carries, while `flow.sim_hook()`
53/// fills in every handle:
54///
55/// ```rust,ignore
56/// #[derive(Clone, Copy, Default)]
57/// pub struct CounterHooks {
58///     pub batch: Option<BatchHook<u32>>,
59///     pub snapshot: Option<SnapshotHook<u64>>,
60/// }
61///
62/// impl SimHook for CounterHooks {
63///     fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
64///         CounterHooks {
65///             batch: SimHook::create(next_id),
66///             snapshot: SimHook::create(next_id),
67///         }
68///     }
69/// }
70/// ```
71///
72/// Such structs nest (a field can itself be a struct of handles), and since handles are
73/// `Copy` a test can pass the struct around or destructure it freely.
74///
75/// Each handle's [`SimHook`] impl carries the trait bounds that the *scripted simulation
76/// codegen* for its operator kind requires (serde round-tripping for decisions, equality
77/// for value-naming decisions, `Hash + Eq + Clone` keys for keyed buffers). Since handles
78/// can only be created through this trait, binding a hook to an operator over unsupported
79/// types fails at the `flow.sim_hook()` call — an ordinary compile error in the test crate
80/// — instead of surfacing as a rustc failure inside the generated simulation dylib.
81pub trait SimHook {
82    /// Creates every handle in this value, allocating fresh IDs via `next_id`.
83    fn create(next_id: &mut dyn FnMut() -> usize) -> Self;
84}
85
86impl<H: SimHook> SimHook for Option<H> {
87    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
88        Some(H::create(next_id))
89    }
90}
91
92/// Hook scope marker: the hooked operator runs on a
93/// [`Process`](crate::location::Process) with tag `P` and has one instance, which the
94/// handle scripts directly.
95///
96/// This is the default scope of every handle type. See [the module docs](self#hook-scopes).
97pub struct OnProcess<P = ()> {
98    _phantom: PhantomData<fn(P)>,
99}
100
101/// Hook scope marker: the hooked operator runs on a
102/// [`Cluster`](crate::location::Cluster) with tag `C`, so every member runs its own
103/// independent instance of the operator.
104///
105/// Select the instance to script with [`.on(member_id)`](BatchHook::on), which yields an
106/// [`OnMember`]-scoped handle. Each member's instance makes its own decisions: a member
107/// with buffered input needs its own decision or pause.
108pub struct OnCluster<C = ()> {
109    _phantom: PhantomData<fn(C)>,
110}
111
112/// Hook scope marker: one selected member's instance of an operator running on a
113/// [`Cluster`](crate::location::Cluster) with tag `C`, produced by
114/// [`.on(member_id)`](BatchHook::on).
115///
116/// Member-scoped handles script decisions like process-scoped ones, but cannot be
117/// created or bound: operators are always bound through the unscoped [`OnCluster`]
118/// handle.
119pub struct OnMember<C = ()> {
120    _phantom: PhantomData<fn(C)>,
121}
122
123/// Hook scopes a handle can be created and bound with: [`OnProcess`] and [`OnCluster`].
124/// Operator signatures select the scope through
125/// [`Location::SimHookScope`](crate::location::Location::SimHookScope).
126#[diagnostic::on_unimplemented(
127    message = "`{Self}` is not a scope that sim hook handles can be created with",
128    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"
129)]
130#[sealed::sealed]
131pub trait BindableHookScope {}
132#[sealed::sealed]
133impl<P> BindableHookScope for OnProcess<P> {}
134#[sealed::sealed]
135impl<C> BindableHookScope for OnCluster<C> {}
136
137/// Hook scopes that name a single instance of the hooked operator and can therefore
138/// script decisions and pauses: [`OnProcess`] and [`OnMember`]. An [`OnCluster`]-scoped
139/// handle must first select a member with [`.on(member_id)`](BatchHook::on).
140#[diagnostic::on_unimplemented(
141    message = "a `{Self}`-scoped sim hook handle cannot script decisions",
142    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)`"
143)]
144#[sealed::sealed]
145pub trait ScriptableHookScope {}
146#[sealed::sealed]
147impl<P> ScriptableHookScope for OnProcess<P> {}
148#[sealed::sealed]
149impl<C> ScriptableHookScope for OnMember<C> {}
150
151/// Generates the `.on(member_id)` member-selection method on [`OnCluster`]-scoped
152/// handles, shared by every handle type.
153macro_rules! on_member_method {
154    ($handle:ident < $($param:ident),* >) => {
155        /// Selects one cluster member's instance of the hooked operator, returning an
156        /// [`OnMember`]-scoped handle with the full scripting API.
157        ///
158        /// Each member's instance is scripted independently: `handle.on(0).release(2)`
159        /// scripts member 0's next batch and says nothing about the other members, each
160        /// of which needs its own decision (or pause) when it holds buffered input. A
161        /// member ID outside the cluster's sizing is reported when the scripting call
162        /// is awaited.
163        pub fn on(&self, member_id: u32) -> $handle<$($param,)* OnMember<C>> {
164            $handle {
165                id: self.id,
166                member: Some(member_id),
167                _phantom: PhantomData,
168            }
169        }
170    };
171}
172
173/// A hook handle controlling a `batch` operator over a stream of `T` elements with ordering
174/// `O` and retry guarantee `R` (mirroring the type of the stream being batched). `S` is the
175/// handle's [scope](self#hook-scopes).
176///
177/// A decision for a batch hook says which buffered elements form the next batch released
178/// into the tick. See `hydro_lang::sim::hooks` for the decisions offered.
179pub struct BatchHook<T, O: Ordering = TotalOrder, R: Retries = ExactlyOnce, S = OnProcess> {
180    pub(crate) id: usize,
181    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
182    #[cfg_attr(
183        not(feature = "sim"),
184        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
185    )]
186    pub(crate) member: Option<u32>,
187    pub(crate) _phantom: PhantomData<fn(T, O, R, S)>,
188}
189
190impl<T, O: Ordering, R: Retries, C> BatchHook<T, O, R, OnCluster<C>> {
191    on_member_method!(BatchHook<T, O, R>);
192}
193
194impl<T, O: Ordering, R: Retries, S> Clone for BatchHook<T, O, R, S> {
195    fn clone(&self) -> Self {
196        *self
197    }
198}
199impl<T, O: Ordering, R: Retries, S> Copy for BatchHook<T, O, R, S> {}
200
201impl<T, O: Ordering, R: Retries, S> std::fmt::Debug for BatchHook<T, O, R, S> {
202    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
203        f.debug_struct("BatchHook").field("id", &self.id).finish()
204    }
205}
206
207impl<T, O: Ordering, R: Retries, S: BindableHookScope> SimHook for BatchHook<T, O, R, S>
208where
209    T: Serialize + DeserializeOwned + PartialEq,
210{
211    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
212        BatchHook {
213            id: next_id(),
214            member: None,
215            _phantom: PhantomData,
216        }
217    }
218}
219
220/// A hook handle controlling a `snapshot` operator over a singleton of `T`. `S` is the
221/// handle's [scope](self#hook-scopes).
222///
223/// A decision for a snapshot hook picks which buffered version of the state the next tick
224/// execution observes. See `hydro_lang::sim::hooks` for the decisions offered.
225pub struct SnapshotHook<T, S = OnProcess> {
226    pub(crate) id: usize,
227    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
228    #[cfg_attr(
229        not(feature = "sim"),
230        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
231    )]
232    pub(crate) member: Option<u32>,
233    pub(crate) _phantom: PhantomData<fn(T, S)>,
234}
235
236impl<T, C> SnapshotHook<T, OnCluster<C>> {
237    on_member_method!(SnapshotHook<T>);
238}
239
240impl<T, S> Clone for SnapshotHook<T, S> {
241    fn clone(&self) -> Self {
242        *self
243    }
244}
245impl<T, S> Copy for SnapshotHook<T, S> {}
246
247impl<T, S> std::fmt::Debug for SnapshotHook<T, S> {
248    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
249        f.debug_struct("SnapshotHook")
250            .field("id", &self.id)
251            .finish()
252    }
253}
254
255impl<T, S: BindableHookScope> SimHook for SnapshotHook<T, S>
256where
257    T: Clone + PartialEq + Serialize + DeserializeOwned,
258{
259    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
260        SnapshotHook {
261            id: next_id(),
262            member: None,
263            _phantom: PhantomData,
264        }
265    }
266}
267
268/// A hook handle controlling an `assume_ordering` operator over `T` elements. `S` is the
269/// handle's [scope](self#hook-scopes).
270///
271/// A top-level decision selects the next buffered element to release. An `assume_ordering`
272/// inside a tick instead takes one exhaustive ordering of that tick's complete input. See
273/// `hydro_lang::sim::hooks` for the decisions offered.
274pub struct OrderingHook<T, B: Boundedness = Unbounded, S = OnProcess> {
275    pub(crate) id: usize,
276    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
277    #[cfg_attr(
278        not(feature = "sim"),
279        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
280    )]
281    pub(crate) member: Option<u32>,
282    pub(crate) _phantom: PhantomData<fn(T, B, S)>,
283}
284
285impl<T, B: Boundedness, C> OrderingHook<T, B, OnCluster<C>> {
286    on_member_method!(OrderingHook<T, B>);
287}
288
289impl<T, B: Boundedness, S> Clone for OrderingHook<T, B, S> {
290    fn clone(&self) -> Self {
291        *self
292    }
293}
294impl<T, B: Boundedness, S> Copy for OrderingHook<T, B, S> {}
295
296impl<T, B: Boundedness, S> std::fmt::Debug for OrderingHook<T, B, S> {
297    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
298        f.debug_struct("OrderingHook")
299            .field("id", &self.id)
300            .finish()
301    }
302}
303
304impl<T, B: Boundedness, S: BindableHookScope> SimHook for OrderingHook<T, B, S>
305where
306    T: Serialize + DeserializeOwned + PartialEq,
307{
308    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
309        OrderingHook {
310            id: next_id(),
311            member: None,
312            _phantom: PhantomData,
313        }
314    }
315}
316
317/// A hook handle controlling a `batch` operator over a keyed stream with keys `K`, values
318/// `V`, per-key value ordering `O`, and retry guarantee `R` (mirroring the type of the
319/// keyed stream being batched). `S` is the handle's [scope](self#hook-scopes).
320///
321/// A decision for a keyed batch hook says which buffered `(key, value)` entries form the
322/// next batch released into the tick. See `hydro_lang::sim::hooks` for the decisions
323/// offered.
324pub struct KeyedBatchHook<K, V, O: Ordering = TotalOrder, R: Retries = ExactlyOnce, S = OnProcess> {
325    pub(crate) id: usize,
326    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
327    #[cfg_attr(
328        not(feature = "sim"),
329        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
330    )]
331    pub(crate) member: Option<u32>,
332    pub(crate) _phantom: PhantomData<fn(K, V, O, R, S)>,
333}
334
335impl<K, V, O: Ordering, R: Retries, C> KeyedBatchHook<K, V, O, R, OnCluster<C>> {
336    on_member_method!(KeyedBatchHook<K, V, O, R>);
337}
338
339impl<K, V, O: Ordering, R: Retries, S> Clone for KeyedBatchHook<K, V, O, R, S> {
340    fn clone(&self) -> Self {
341        *self
342    }
343}
344impl<K, V, O: Ordering, R: Retries, S> Copy for KeyedBatchHook<K, V, O, R, S> {}
345
346impl<K, V, O: Ordering, R: Retries, S> std::fmt::Debug for KeyedBatchHook<K, V, O, R, S> {
347    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
348        f.debug_struct("KeyedBatchHook")
349            .field("id", &self.id)
350            .finish()
351    }
352}
353
354impl<K, V, O: Ordering, R: Retries, S: BindableHookScope> SimHook for KeyedBatchHook<K, V, O, R, S>
355where
356    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
357    V: Serialize + DeserializeOwned + PartialEq,
358{
359    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
360        KeyedBatchHook {
361            id: next_id(),
362            member: None,
363            _phantom: PhantomData,
364        }
365    }
366}
367
368/// A hook handle controlling a `snapshot` (or `batch`) operator over a keyed singleton
369/// with keys `K` and values `V`. `S` is the handle's [scope](self#hook-scopes).
370///
371/// A decision for a keyed snapshot hook picks which buffered version of each key's state
372/// the next tick execution observes. See `hydro_lang::sim::hooks` for the decisions
373/// offered.
374pub struct KeyedSnapshotHook<K, V, S = OnProcess> {
375    pub(crate) id: usize,
376    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
377    #[cfg_attr(
378        not(feature = "sim"),
379        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
380    )]
381    pub(crate) member: Option<u32>,
382    pub(crate) _phantom: PhantomData<fn(K, V, S)>,
383}
384
385impl<K, V, C> KeyedSnapshotHook<K, V, OnCluster<C>> {
386    on_member_method!(KeyedSnapshotHook<K, V>);
387}
388
389impl<K, V, S> Clone for KeyedSnapshotHook<K, V, S> {
390    fn clone(&self) -> Self {
391        *self
392    }
393}
394impl<K, V, S> Copy for KeyedSnapshotHook<K, V, S> {}
395
396impl<K, V, S> std::fmt::Debug for KeyedSnapshotHook<K, V, S> {
397    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
398        f.debug_struct("KeyedSnapshotHook")
399            .field("id", &self.id)
400            .finish()
401    }
402}
403
404impl<K, V, S: BindableHookScope> SimHook for KeyedSnapshotHook<K, V, S>
405where
406    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
407    V: Clone + PartialEq + Serialize + DeserializeOwned,
408{
409    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
410        KeyedSnapshotHook {
411            id: next_id(),
412            member: None,
413            _phantom: PhantomData,
414        }
415    }
416}
417
418/// A hook handle controlling an `assume_ordering` operator over a keyed stream with keys
419/// `K` and values `V`. `S` is the handle's [scope](self#hook-scopes).
420///
421/// A top-level decision selects the next buffered `(key, value)` entry to release. An
422/// `assume_ordering` inside a tick instead takes one exhaustive per-key ordering of that
423/// tick's complete input. See `hydro_lang::sim::hooks` for the decisions offered.
424pub struct KeyedOrderingHook<K, V, B: Boundedness = Unbounded, S = OnProcess> {
425    pub(crate) id: usize,
426    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
427    #[cfg_attr(
428        not(feature = "sim"),
429        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
430    )]
431    pub(crate) member: Option<u32>,
432    pub(crate) _phantom: PhantomData<fn(K, V, B, S)>,
433}
434
435impl<K, V, B: Boundedness, C> KeyedOrderingHook<K, V, B, OnCluster<C>> {
436    on_member_method!(KeyedOrderingHook<K, V, B>);
437}
438
439impl<K, V, B: Boundedness, S> Clone for KeyedOrderingHook<K, V, B, S> {
440    fn clone(&self) -> Self {
441        *self
442    }
443}
444impl<K, V, B: Boundedness, S> Copy for KeyedOrderingHook<K, V, B, S> {}
445
446impl<K, V, B: Boundedness, S> std::fmt::Debug for KeyedOrderingHook<K, V, B, S> {
447    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
448        f.debug_struct("KeyedOrderingHook")
449            .field("id", &self.id)
450            .finish()
451    }
452}
453
454impl<K, V, B: Boundedness, S: BindableHookScope> SimHook for KeyedOrderingHook<K, V, B, S>
455where
456    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
457    V: Serialize + DeserializeOwned + PartialEq,
458{
459    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
460        KeyedOrderingHook {
461            id: next_id(),
462            member: None,
463            _phantom: PhantomData,
464        }
465    }
466}
467
468/// A hook handle controlling an `entries_partially_ordered` operator over a keyed stream
469/// with keys `K` and values `V`. `S` is the handle's [scope](self#hook-scopes).
470///
471/// The operator preserves the order of values within each key while interleaving across
472/// keys non-deterministically. A top-level decision releases the front entry of one key's
473/// buffer; inside a tick, a single decision supplies the complete interleaving. See
474/// `hydro_lang::sim::hooks` for the decisions offered.
475pub struct PartialOrderingHook<K, V, B: Boundedness = Unbounded, S = OnProcess> {
476    pub(crate) id: usize,
477    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
478    #[cfg_attr(
479        not(feature = "sim"),
480        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
481    )]
482    pub(crate) member: Option<u32>,
483    pub(crate) _phantom: PhantomData<fn(K, V, B, S)>,
484}
485
486impl<K, V, B: Boundedness, C> PartialOrderingHook<K, V, B, OnCluster<C>> {
487    on_member_method!(PartialOrderingHook<K, V, B>);
488}
489
490impl<K, V, B: Boundedness, S> Clone for PartialOrderingHook<K, V, B, S> {
491    fn clone(&self) -> Self {
492        *self
493    }
494}
495impl<K, V, B: Boundedness, S> Copy for PartialOrderingHook<K, V, B, S> {}
496
497impl<K, V, B: Boundedness, S> std::fmt::Debug for PartialOrderingHook<K, V, B, S> {
498    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
499        f.debug_struct("PartialOrderingHook")
500            .field("id", &self.id)
501            .finish()
502    }
503}
504
505impl<K, V, B: Boundedness, S: BindableHookScope> SimHook for PartialOrderingHook<K, V, B, S>
506where
507    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
508    V: Serialize + DeserializeOwned + PartialEq,
509{
510    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
511        PartialOrderingHook {
512            id: next_id(),
513            member: None,
514            _phantom: PhantomData,
515        }
516    }
517}
518
519/// A hook handle controlling a `merge_ordered` operator over streams of `T` elements. `S`
520/// is the handle's [scope](self#hook-scopes).
521///
522/// The operator preserves the order of each input while interleaving the two inputs
523/// non-deterministically. A top-level decision releases the front element of one input's
524/// buffer; inside a tick, a single decision supplies the complete interleaving. See
525/// `hydro_lang::sim::hooks` for the decisions offered.
526pub struct MergeOrderedHook<T, B: Boundedness = Unbounded, S = OnProcess> {
527    pub(crate) id: usize,
528    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
529    #[cfg_attr(
530        not(feature = "sim"),
531        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
532    )]
533    pub(crate) member: Option<u32>,
534    pub(crate) _phantom: PhantomData<fn(T, B, S)>,
535}
536
537impl<T, B: Boundedness, C> MergeOrderedHook<T, B, OnCluster<C>> {
538    on_member_method!(MergeOrderedHook<T, B>);
539}
540
541impl<T, B: Boundedness, S> Clone for MergeOrderedHook<T, B, S> {
542    fn clone(&self) -> Self {
543        *self
544    }
545}
546impl<T, B: Boundedness, S> Copy for MergeOrderedHook<T, B, S> {}
547
548impl<T, B: Boundedness, S> std::fmt::Debug for MergeOrderedHook<T, B, S> {
549    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
550        f.debug_struct("MergeOrderedHook")
551            .field("id", &self.id)
552            .finish()
553    }
554}
555
556impl<T, B: Boundedness, S: BindableHookScope> SimHook for MergeOrderedHook<T, B, S>
557where
558    T: Serialize + DeserializeOwned + PartialEq,
559{
560    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
561        MergeOrderedHook {
562            id: next_id(),
563            member: None,
564            _phantom: PhantomData,
565        }
566    }
567}
568
569/// A hook handle controlling a `merge_ordered` operator over keyed streams with keys `K`
570/// and values `V`. `S` is the handle's [scope](self#hook-scopes).
571///
572/// The operator preserves each input's order within every key while interleaving the two
573/// inputs non-deterministically (cross-key order is unconstrained). A top-level decision
574/// releases the front entry of one key's buffer in one input; inside a tick, a single
575/// decision supplies the complete interleaving. See `hydro_lang::sim::hooks` for the
576/// decisions offered.
577pub struct KeyedMergeOrderedHook<K, V, B: Boundedness = Unbounded, S = OnProcess> {
578    pub(crate) id: usize,
579    /// The member selected by `.on(member_id)`; `None` for process-scoped handles.
580    #[cfg_attr(
581        not(feature = "sim"),
582        expect(dead_code, reason = "only read by the `sim`-gated scripting API")
583    )]
584    pub(crate) member: Option<u32>,
585    pub(crate) _phantom: PhantomData<fn(K, V, B, S)>,
586}
587
588impl<K, V, B: Boundedness, C> KeyedMergeOrderedHook<K, V, B, OnCluster<C>> {
589    on_member_method!(KeyedMergeOrderedHook<K, V, B>);
590}
591
592impl<K, V, B: Boundedness, S> Clone for KeyedMergeOrderedHook<K, V, B, S> {
593    fn clone(&self) -> Self {
594        *self
595    }
596}
597impl<K, V, B: Boundedness, S> Copy for KeyedMergeOrderedHook<K, V, B, S> {}
598
599impl<K, V, B: Boundedness, S> std::fmt::Debug for KeyedMergeOrderedHook<K, V, B, S> {
600    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
601        f.debug_struct("KeyedMergeOrderedHook")
602            .field("id", &self.id)
603            .finish()
604    }
605}
606
607impl<K, V, B: Boundedness, S: BindableHookScope> SimHook for KeyedMergeOrderedHook<K, V, B, S>
608where
609    K: Hash + Eq + Clone + Serialize + DeserializeOwned,
610    V: Serialize + DeserializeOwned + PartialEq,
611{
612    fn create(next_id: &mut dyn FnMut() -> usize) -> Self {
613        KeyedMergeOrderedHook {
614            id: next_id(),
615            member: None,
616            _phantom: PhantomData,
617        }
618    }
619}
hydro_lang/sim_hooks.rs

hydro_lang/
sim_hooks.rs
