hydro_lang/compile/
deploy_provider.rs

1use std::io::Error;
2use std::pin::Pin;
3
4use bytes::{Bytes, BytesMut};
5use dfir_lang::graph::{AsCodeOptions, DfirGraph};
6use futures::{Sink, Stream};
7use serde::Serialize;
8use serde::de::DeserializeOwned;
9
10use crate::compile::builder::ExternalPortId;
11use crate::location::dynamic::LocationId;
12use crate::location::{LocationKey, NetworkHint};
13
14pub trait Deploy<'a> {
15    type Meta: Default;
16    type InstantiateEnv;
17
18    type Process: Node<Meta = Self::Meta, InstantiateEnv = Self::InstantiateEnv> + Clone;
19    type Cluster: Node<Meta = Self::Meta, InstantiateEnv = Self::InstantiateEnv> + Clone;
20    type External: Node<Meta = Self::Meta, InstantiateEnv = Self::InstantiateEnv>
21        + RegisterPort<'a, Self>;
22
23    /// Whether this deployment backend supports network channels that leave serialization to code
24    /// outside of Hydro (see [`crate::networking::Embedded`]). When `false`, attempting to use such
25    /// a channel will panic during instantiation.
26    const SUPPORTS_EXTERNAL_SERIALIZATION: bool = false;
27
28    /// Generates the source and sink expressions when connecting a [`Self::Process`] to another
29    /// [`Self::Process`].
30    ///
31    /// The [`Self::InstantiateEnv`] can be used to record metadata about the created channel. The
32    /// provided `name` is the user-configured channel name from the network IR node.
33    ///
34    /// When `external_types` is [`Some`], the channel leaves serialization to code outside
35    /// of Hydro and the provided type is the raw element type that flows across the channel
36    /// unserialized.
37    #[expect(clippy::too_many_arguments, reason = "networking codegen")]
38    fn o2o_sink_source(
39        env: &mut Self::InstantiateEnv,
40        p1: &Self::Process,
41        p1_port: &<Self::Process as Node>::Port,
42        p2: &Self::Process,
43        p2_port: &<Self::Process as Node>::Port,
44        name: Option<&str>,
45        networking_info: &crate::networking::NetworkingInfo,
46        external_types: Option<(&syn::Type, &syn::Type)>,
47    ) -> (syn::Expr, syn::Expr);
48
49    /// Performs any runtime wiring needed after code generation for a
50    /// [`Self::Process`]-to-[`Self::Process`] channel.
51    ///
52    /// The returned closure is executed once all locations have been instantiated.
53    fn o2o_connect(
54        p1: &Self::Process,
55        p1_port: &<Self::Process as Node>::Port,
56        p2: &Self::Process,
57        p2_port: &<Self::Process as Node>::Port,
58    ) -> Box<dyn FnOnce()>;
59
60    /// Generates the source and sink expressions when connecting a [`Self::Process`] to a
61    /// [`Self::Cluster`] (one-to-many).
62    ///
63    /// The sink expression is used on the sending process and the source expression on each
64    /// receiving cluster member. The [`Self::InstantiateEnv`] can be used to record metadata
65    /// about the created channel. The provided `name` is the user-configured channel name
66    /// from the network IR node.
67    ///
68    /// When `external_types` is [`Some`], the channel leaves serialization to code outside
69    /// of Hydro and the provided type is the raw element type that flows across the channel
70    /// unserialized.
71    #[expect(clippy::too_many_arguments, reason = "networking codegen")]
72    fn o2m_sink_source(
73        env: &mut Self::InstantiateEnv,
74        p1: &Self::Process,
75        p1_port: &<Self::Process as Node>::Port,
76        c2: &Self::Cluster,
77        c2_port: &<Self::Cluster as Node>::Port,
78        name: Option<&str>,
79        networking_info: &crate::networking::NetworkingInfo,
80        external_types: Option<(&syn::Type, &syn::Type)>,
81    ) -> (syn::Expr, syn::Expr);
82
83    /// Performs any runtime wiring needed after code generation for a
84    /// [`Self::Process`]-to-[`Self::Cluster`] channel.
85    ///
86    /// The returned closure is executed once all locations have been instantiated.
87    fn o2m_connect(
88        p1: &Self::Process,
89        p1_port: &<Self::Process as Node>::Port,
90        c2: &Self::Cluster,
91        c2_port: &<Self::Cluster as Node>::Port,
92    ) -> Box<dyn FnOnce()>;
93
94    /// Generates the source and sink expressions when connecting a [`Self::Cluster`] to a
95    /// [`Self::Process`] (many-to-one).
96    ///
97    /// The sink expression is used on each sending cluster member and the source expression
98    /// on the receiving process. The [`Self::InstantiateEnv`] can be used to record metadata
99    /// about the created channel. The provided `name` is the user-configured channel name
100    /// from the network IR node.
101    ///
102    /// When `external_types` is [`Some`], the channel leaves serialization to code outside
103    /// of Hydro and the provided type is the raw element type that flows across the channel
104    /// unserialized.
105    #[expect(clippy::too_many_arguments, reason = "networking codegen")]
106    fn m2o_sink_source(
107        env: &mut Self::InstantiateEnv,
108        c1: &Self::Cluster,
109        c1_port: &<Self::Cluster as Node>::Port,
110        p2: &Self::Process,
111        p2_port: &<Self::Process as Node>::Port,
112        name: Option<&str>,
113        networking_info: &crate::networking::NetworkingInfo,
114        external_types: Option<(&syn::Type, &syn::Type)>,
115    ) -> (syn::Expr, syn::Expr);
116
117    /// Performs any runtime wiring needed after code generation for a
118    /// [`Self::Cluster`]-to-[`Self::Process`] channel.
119    ///
120    /// The returned closure is executed once all locations have been instantiated.
121    fn m2o_connect(
122        c1: &Self::Cluster,
123        c1_port: &<Self::Cluster as Node>::Port,
124        p2: &Self::Process,
125        p2_port: &<Self::Process as Node>::Port,
126    ) -> Box<dyn FnOnce()>;
127
128    /// Generates the source and sink expressions when connecting a [`Self::Cluster`] to another
129    /// [`Self::Cluster`] (many-to-many).
130    ///
131    /// The sink expression is used on each sending cluster member and the source expression
132    /// on each receiving cluster member. The [`Self::InstantiateEnv`] can be used to record
133    /// metadata about the created channel. The provided `name` is the user-configured channel
134    /// name from the network IR node.
135    ///
136    /// When `external_types` is [`Some`], the channel leaves serialization to code outside
137    /// of Hydro and the provided type is the raw element type that flows across the channel
138    /// unserialized.
139    #[expect(clippy::too_many_arguments, reason = "networking codegen")]
140    fn m2m_sink_source(
141        env: &mut Self::InstantiateEnv,
142        c1: &Self::Cluster,
143        c1_port: &<Self::Cluster as Node>::Port,
144        c2: &Self::Cluster,
145        c2_port: &<Self::Cluster as Node>::Port,
146        name: Option<&str>,
147        networking_info: &crate::networking::NetworkingInfo,
148        external_types: Option<(&syn::Type, &syn::Type)>,
149    ) -> (syn::Expr, syn::Expr);
150
151    /// Performs any runtime wiring needed after code generation for a
152    /// [`Self::Cluster`]-to-[`Self::Cluster`] channel.
153    ///
154    /// The returned closure is executed once all locations have been instantiated.
155    fn m2m_connect(
156        c1: &Self::Cluster,
157        c1_port: &<Self::Cluster as Node>::Port,
158        c2: &Self::Cluster,
159        c2_port: &<Self::Cluster as Node>::Port,
160    ) -> Box<dyn FnOnce()>;
161
162    fn e2o_many_source(
163        extra_stmts: &mut Vec<syn::Stmt>,
164        p2: &Self::Process,
165        p2_port: &<Self::Process as Node>::Port,
166        codec_type: &syn::Type,
167        shared_handle: String,
168    ) -> syn::Expr;
169    fn e2o_many_sink(shared_handle: String) -> syn::Expr;
170
171    fn e2o_source(
172        extra_stmts: &mut Vec<syn::Stmt>,
173        p1: &Self::External,
174        p1_port: &<Self::External as Node>::Port,
175        p2: &Self::Process,
176        p2_port: &<Self::Process as Node>::Port,
177        codec_type: &syn::Type,
178        shared_handle: String,
179    ) -> syn::Expr;
180    fn e2o_connect(
181        p1: &Self::External,
182        p1_port: &<Self::External as Node>::Port,
183        p2: &Self::Process,
184        p2_port: &<Self::Process as Node>::Port,
185        many: bool,
186        server_hint: NetworkHint,
187    ) -> Box<dyn FnOnce()>;
188
189    fn o2e_sink(
190        p1: &Self::Process,
191        p1_port: &<Self::Process as Node>::Port,
192        p2: &Self::External,
193        p2_port: &<Self::External as Node>::Port,
194        shared_handle: String,
195    ) -> syn::Expr;
196
197    fn e2m_source(
198        extra_stmts: &mut Vec<syn::Stmt>,
199        p1: &Self::External,
200        p1_port: &<Self::External as Node>::Port,
201        c2: &Self::Cluster,
202        c2_port: &<Self::Cluster as Node>::Port,
203        codec_type: &syn::Type,
204        shared_handle: String,
205    ) -> syn::Expr {
206        let _ = (
207            extra_stmts,
208            p1,
209            p1_port,
210            c2,
211            c2_port,
212            codec_type,
213            shared_handle,
214        );
215        todo!("e2m_source is not yet supported for this deploy backend")
216    }
217
218    fn e2m_connect(
219        p1: &Self::External,
220        p1_port: &<Self::External as Node>::Port,
221        c2: &Self::Cluster,
222        c2_port: &<Self::Cluster as Node>::Port,
223        server_hint: NetworkHint,
224    ) -> Box<dyn FnOnce()> {
225        let _ = (p1, p1_port, c2, c2_port, server_hint);
226        todo!("e2m_connect is not yet supported for this deploy backend")
227    }
228
229    fn m2e_sink(
230        c1: &Self::Cluster,
231        c1_port: &<Self::Cluster as Node>::Port,
232        p2: &Self::External,
233        p2_port: &<Self::External as Node>::Port,
234        shared_handle: String,
235    ) -> syn::Expr {
236        let _ = (c1, c1_port, p2, p2_port, shared_handle);
237        todo!("m2e_sink is not yet supported for this deploy backend")
238    }
239
240    /// The member IDs of `of_cluster`'s members, as a `&[TaglessMemberId]` expression.
241    fn cluster_ids(of_cluster: LocationKey) -> syn::Expr;
242
243    /// This member's own ID, as a `TaglessMemberId` expression.
244    fn cluster_self_id() -> syn::Expr;
245
246    /// A `Box<dyn Stream<Item = (TaglessMemberId, MembershipEvent)> + Unpin>` expression: the
247    /// membership events of `location_id`, as seen from `at_location`.
248    fn cluster_membership_stream(
249        env: &mut Self::InstantiateEnv,
250        at_location: &LocationId,
251        location_id: &LocationId,
252    ) -> syn::Expr;
253
254    /// Registers an embedded stream input for the given ident and element type.
255    ///
256    /// Only meaningful for the embedded deployment backend. The default
257    /// implementation panics.
258    fn register_embedded_stream_input(
259        _env: &mut Self::InstantiateEnv,
260        _location_key: LocationKey,
261        _ident: &syn::Ident,
262        _element_type: &syn::Type,
263    ) {
264        panic!("register_embedded_stream_input is only supported by EmbeddedDeploy");
265    }
266
267    /// Registers an embedded singleton input for the given ident and element type.
268    ///
269    /// Only meaningful for the embedded deployment backend. The default
270    /// implementation panics.
271    fn register_embedded_singleton_input(
272        _env: &mut Self::InstantiateEnv,
273        _location_key: LocationKey,
274        _ident: &syn::Ident,
275        _element_type: &syn::Type,
276    ) {
277        panic!("register_embedded_singleton_input is only supported by EmbeddedDeploy");
278    }
279
280    /// Registers an embedded output for the given ident and element type.
281    ///
282    /// Only meaningful for the embedded deployment backend. The default
283    /// implementation panics.
284    fn register_embedded_output(
285        _env: &mut Self::InstantiateEnv,
286        _location_key: LocationKey,
287        _ident: &syn::Ident,
288        _element_type: &syn::Type,
289    ) {
290        panic!("register_embedded_output is only supported by EmbeddedDeploy");
291    }
292}
293
294pub trait ProcessSpec<'a, D>
295where
296    D: Deploy<'a> + ?Sized,
297{
298    fn build(self, location_key: LocationKey, name_hint: &str) -> D::Process;
299}
300
301pub trait IntoProcessSpec<'a, D>
302where
303    D: Deploy<'a> + ?Sized,
304{
305    type ProcessSpec: ProcessSpec<'a, D>;
306    fn into_process_spec(self) -> Self::ProcessSpec;
307}
308
309impl<'a, D, T> IntoProcessSpec<'a, D> for T
310where
311    D: Deploy<'a> + ?Sized,
312    T: ProcessSpec<'a, D>,
313{
314    type ProcessSpec = T;
315    fn into_process_spec(self) -> Self::ProcessSpec {
316        self
317    }
318}
319
320pub trait ClusterSpec<'a, D>
321where
322    D: Deploy<'a> + ?Sized,
323{
324    fn build(self, location_key: LocationKey, name_hint: &str) -> D::Cluster;
325}
326
327pub trait ExternalSpec<'a, D>
328where
329    D: Deploy<'a> + ?Sized,
330{
331    fn build(self, location_key: LocationKey, name_hint: &str) -> D::External;
332}
333
334pub trait Node {
335    /// A logical communication endpoint for this node.
336    ///
337    /// Implementors are free to choose the concrete representation (for example,
338    /// a handle or identifier), but it must be `Clone` so that a single logical
339    /// port can be duplicated and passed to multiple consumers. New ports are
340    /// allocated via [`Self::next_port`].
341    type Port: Clone;
342    type Meta: Default;
343    type InstantiateEnv;
344
345    /// Allocates and returns a new port.
346    fn next_port(&self) -> Self::Port;
347
348    fn update_meta(&self, meta: &Self::Meta);
349
350    fn instantiate(
351        &self,
352        env: &mut Self::InstantiateEnv,
353        meta: &mut Self::Meta,
354        graph: DfirGraph,
355        extra_stmts: &[syn::Stmt],
356        sidecars: &[syn::Expr],
357        as_code_options: &AsCodeOptions,
358    );
359}
360
361pub type DynSourceSink<Out, In, InErr> = (
362    Pin<Box<dyn Stream<Item = Out>>>,
363    Pin<Box<dyn Sink<In, Error = InErr>>>,
364);
365
366pub trait RegisterPort<'a, D>: Node + Clone
367where
368    D: Deploy<'a> + ?Sized,
369{
370    fn register(&self, external_port_id: ExternalPortId, port: Self::Port);
371
372    fn as_bytes_bidi(
373        &self,
374        external_port_id: ExternalPortId,
375    ) -> impl Future<Output = DynSourceSink<Result<BytesMut, Error>, Bytes, Error>> + 'a;
376
377    fn as_bincode_bidi<InT, OutT>(
378        &self,
379        external_port_id: ExternalPortId,
380    ) -> impl Future<Output = DynSourceSink<OutT, InT, Error>> + 'a
381    where
382        InT: Serialize + 'static,
383        OutT: DeserializeOwned + 'static;
384
385    fn as_bincode_sink<T>(
386        &self,
387        external_port_id: ExternalPortId,
388    ) -> impl Future<Output = Pin<Box<dyn Sink<T, Error = Error>>>> + 'a
389    where
390        T: Serialize + 'static;
391
392    fn as_bincode_source<T>(
393        &self,
394        external_port_id: ExternalPortId,
395    ) -> impl Future<Output = Pin<Box<dyn Stream<Item = T>>>> + 'a
396    where
397        T: DeserializeOwned + 'static;
398}
hydro_lang/compile/deploy_provider.rs

hydro_lang/compile/
deploy_provider.rs
