hydro_lang/compile/
deploy_provider.rs

1use std::io::Error;
2use std::pin::Pin;
3
4use bytes::{Bytes, BytesMut};
5use dfir_lang::graph::DfirGraph;
6use futures::{Sink, Stream};
7use serde::Serialize;
8use serde::de::DeserializeOwned;
9use stageleft::QuotedWithContext;
10
11use crate::compile::builder::ExternalPortId;
12use crate::location::dynamic::LocationId;
13use crate::location::member_id::TaglessMemberId;
14use crate::location::{LocationKey, MembershipEvent, NetworkHint};
15
16pub trait Deploy<'a> {
17    type Meta: Default;
18    type InstantiateEnv;
19
20    type Process: Node<Meta = Self::Meta, InstantiateEnv = Self::InstantiateEnv> + Clone;
21    type Cluster: Node<Meta = Self::Meta, InstantiateEnv = Self::InstantiateEnv> + Clone;
22    type External: Node<Meta = Self::Meta, InstantiateEnv = Self::InstantiateEnv>
23        + RegisterPort<'a, Self>;
24
25    /// Whether this deployment backend supports network channels that leave serialization to code
26    /// outside of Hydro (see [`crate::networking::Embedded`]). When `false`, attempting to use such
27    /// a channel will panic during instantiation.
28    const SUPPORTS_EXTERNAL_SERIALIZATION: bool = false;
29
30    /// Generates the source and sink expressions when connecting a [`Self::Process`] to another
31    /// [`Self::Process`].
32    ///
33    /// The [`Self::InstantiateEnv`] can be used to record metadata about the created channel. The
34    /// provided `name` is the user-configured channel name from the network IR node.
35    ///
36    /// When `external_types` is [`Some`], the channel leaves serialization to code outside
37    /// of Hydro and the provided type is the raw element type that flows across the channel
38    /// unserialized.
39    #[expect(clippy::too_many_arguments, reason = "networking codegen")]
40    fn o2o_sink_source(
41        env: &mut Self::InstantiateEnv,
42        p1: &Self::Process,
43        p1_port: &<Self::Process as Node>::Port,
44        p2: &Self::Process,
45        p2_port: &<Self::Process as Node>::Port,
46        name: Option<&str>,
47        networking_info: &crate::networking::NetworkingInfo,
48        external_types: Option<(&syn::Type, &syn::Type)>,
49    ) -> (syn::Expr, syn::Expr);
50
51    /// Performs any runtime wiring needed after code generation for a
52    /// [`Self::Process`]-to-[`Self::Process`] channel.
53    ///
54    /// The returned closure is executed once all locations have been instantiated.
55    fn o2o_connect(
56        p1: &Self::Process,
57        p1_port: &<Self::Process as Node>::Port,
58        p2: &Self::Process,
59        p2_port: &<Self::Process as Node>::Port,
60    ) -> Box<dyn FnOnce()>;
61
62    /// Generates the source and sink expressions when connecting a [`Self::Process`] to a
63    /// [`Self::Cluster`] (one-to-many).
64    ///
65    /// The sink expression is used on the sending process and the source expression on each
66    /// receiving cluster member. The [`Self::InstantiateEnv`] can be used to record metadata
67    /// about the created channel. The provided `name` is the user-configured channel name
68    /// from the network IR node.
69    ///
70    /// When `external_types` is [`Some`], the channel leaves serialization to code outside
71    /// of Hydro and the provided type is the raw element type that flows across the channel
72    /// unserialized.
73    #[expect(clippy::too_many_arguments, reason = "networking codegen")]
74    fn o2m_sink_source(
75        env: &mut Self::InstantiateEnv,
76        p1: &Self::Process,
77        p1_port: &<Self::Process as Node>::Port,
78        c2: &Self::Cluster,
79        c2_port: &<Self::Cluster as Node>::Port,
80        name: Option<&str>,
81        networking_info: &crate::networking::NetworkingInfo,
82        external_types: Option<(&syn::Type, &syn::Type)>,
83    ) -> (syn::Expr, syn::Expr);
84
85    /// Performs any runtime wiring needed after code generation for a
86    /// [`Self::Process`]-to-[`Self::Cluster`] channel.
87    ///
88    /// The returned closure is executed once all locations have been instantiated.
89    fn o2m_connect(
90        p1: &Self::Process,
91        p1_port: &<Self::Process as Node>::Port,
92        c2: &Self::Cluster,
93        c2_port: &<Self::Cluster as Node>::Port,
94    ) -> Box<dyn FnOnce()>;
95
96    /// Generates the source and sink expressions when connecting a [`Self::Cluster`] to a
97    /// [`Self::Process`] (many-to-one).
98    ///
99    /// The sink expression is used on each sending cluster member and the source expression
100    /// on the receiving process. The [`Self::InstantiateEnv`] can be used to record metadata
101    /// about the created channel. The provided `name` is the user-configured channel name
102    /// from the network IR node.
103    ///
104    /// When `external_types` is [`Some`], the channel leaves serialization to code outside
105    /// of Hydro and the provided type is the raw element type that flows across the channel
106    /// unserialized.
107    #[expect(clippy::too_many_arguments, reason = "networking codegen")]
108    fn m2o_sink_source(
109        env: &mut Self::InstantiateEnv,
110        c1: &Self::Cluster,
111        c1_port: &<Self::Cluster as Node>::Port,
112        p2: &Self::Process,
113        p2_port: &<Self::Process as Node>::Port,
114        name: Option<&str>,
115        networking_info: &crate::networking::NetworkingInfo,
116        external_types: Option<(&syn::Type, &syn::Type)>,
117    ) -> (syn::Expr, syn::Expr);
118
119    /// Performs any runtime wiring needed after code generation for a
120    /// [`Self::Cluster`]-to-[`Self::Process`] channel.
121    ///
122    /// The returned closure is executed once all locations have been instantiated.
123    fn m2o_connect(
124        c1: &Self::Cluster,
125        c1_port: &<Self::Cluster as Node>::Port,
126        p2: &Self::Process,
127        p2_port: &<Self::Process as Node>::Port,
128    ) -> Box<dyn FnOnce()>;
129
130    /// Generates the source and sink expressions when connecting a [`Self::Cluster`] to another
131    /// [`Self::Cluster`] (many-to-many).
132    ///
133    /// The sink expression is used on each sending cluster member and the source expression
134    /// on each receiving cluster member. The [`Self::InstantiateEnv`] can be used to record
135    /// metadata about the created channel. The provided `name` is the user-configured channel
136    /// name from the network IR node.
137    ///
138    /// When `external_types` is [`Some`], the channel leaves serialization to code outside
139    /// of Hydro and the provided type is the raw element type that flows across the channel
140    /// unserialized.
141    #[expect(clippy::too_many_arguments, reason = "networking codegen")]
142    fn m2m_sink_source(
143        env: &mut Self::InstantiateEnv,
144        c1: &Self::Cluster,
145        c1_port: &<Self::Cluster as Node>::Port,
146        c2: &Self::Cluster,
147        c2_port: &<Self::Cluster as Node>::Port,
148        name: Option<&str>,
149        networking_info: &crate::networking::NetworkingInfo,
150        external_types: Option<(&syn::Type, &syn::Type)>,
151    ) -> (syn::Expr, syn::Expr);
152
153    /// Performs any runtime wiring needed after code generation for a
154    /// [`Self::Cluster`]-to-[`Self::Cluster`] channel.
155    ///
156    /// The returned closure is executed once all locations have been instantiated.
157    fn m2m_connect(
158        c1: &Self::Cluster,
159        c1_port: &<Self::Cluster as Node>::Port,
160        c2: &Self::Cluster,
161        c2_port: &<Self::Cluster as Node>::Port,
162    ) -> Box<dyn FnOnce()>;
163
164    fn e2o_many_source(
165        extra_stmts: &mut Vec<syn::Stmt>,
166        p2: &Self::Process,
167        p2_port: &<Self::Process as Node>::Port,
168        codec_type: &syn::Type,
169        shared_handle: String,
170    ) -> syn::Expr;
171    fn e2o_many_sink(shared_handle: String) -> syn::Expr;
172
173    fn e2o_source(
174        extra_stmts: &mut Vec<syn::Stmt>,
175        p1: &Self::External,
176        p1_port: &<Self::External as Node>::Port,
177        p2: &Self::Process,
178        p2_port: &<Self::Process as Node>::Port,
179        codec_type: &syn::Type,
180        shared_handle: String,
181    ) -> syn::Expr;
182    fn e2o_connect(
183        p1: &Self::External,
184        p1_port: &<Self::External as Node>::Port,
185        p2: &Self::Process,
186        p2_port: &<Self::Process as Node>::Port,
187        many: bool,
188        server_hint: NetworkHint,
189    ) -> Box<dyn FnOnce()>;
190
191    fn o2e_sink(
192        p1: &Self::Process,
193        p1_port: &<Self::Process as Node>::Port,
194        p2: &Self::External,
195        p2_port: &<Self::External as Node>::Port,
196        shared_handle: String,
197    ) -> syn::Expr;
198
199    fn e2m_source(
200        extra_stmts: &mut Vec<syn::Stmt>,
201        p1: &Self::External,
202        p1_port: &<Self::External as Node>::Port,
203        c2: &Self::Cluster,
204        c2_port: &<Self::Cluster as Node>::Port,
205        codec_type: &syn::Type,
206        shared_handle: String,
207    ) -> syn::Expr {
208        let _ = (
209            extra_stmts,
210            p1,
211            p1_port,
212            c2,
213            c2_port,
214            codec_type,
215            shared_handle,
216        );
217        todo!("e2m_source is not yet supported for this deploy backend")
218    }
219
220    fn e2m_connect(
221        p1: &Self::External,
222        p1_port: &<Self::External as Node>::Port,
223        c2: &Self::Cluster,
224        c2_port: &<Self::Cluster as Node>::Port,
225        server_hint: NetworkHint,
226    ) -> Box<dyn FnOnce()> {
227        let _ = (p1, p1_port, c2, c2_port, server_hint);
228        todo!("e2m_connect is not yet supported for this deploy backend")
229    }
230
231    fn m2e_sink(
232        c1: &Self::Cluster,
233        c1_port: &<Self::Cluster as Node>::Port,
234        p2: &Self::External,
235        p2_port: &<Self::External as Node>::Port,
236        shared_handle: String,
237    ) -> syn::Expr {
238        let _ = (c1, c1_port, p2, p2_port, shared_handle);
239        todo!("m2e_sink is not yet supported for this deploy backend")
240    }
241
242    fn cluster_ids(
243        of_cluster: LocationKey,
244    ) -> impl QuotedWithContext<'a, &'a [TaglessMemberId], ()> + Clone + 'a;
245
246    fn cluster_self_id() -> impl QuotedWithContext<'a, TaglessMemberId, ()> + Clone + 'a;
247
248    fn cluster_membership_stream(
249        env: &mut Self::InstantiateEnv,
250        at_location: &LocationId,
251        location_id: &LocationId,
252    ) -> impl QuotedWithContext<'a, Box<dyn Stream<Item = (TaglessMemberId, MembershipEvent)> + Unpin>, ()>;
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    );
358}
359
360pub type DynSourceSink<Out, In, InErr> = (
361    Pin<Box<dyn Stream<Item = Out>>>,
362    Pin<Box<dyn Sink<In, Error = InErr>>>,
363);
364
365pub trait RegisterPort<'a, D>: Node + Clone
366where
367    D: Deploy<'a> + ?Sized,
368{
369    fn register(&self, external_port_id: ExternalPortId, port: Self::Port);
370
371    fn as_bytes_bidi(
372        &self,
373        external_port_id: ExternalPortId,
374    ) -> impl Future<Output = DynSourceSink<Result<BytesMut, Error>, Bytes, Error>> + 'a;
375
376    fn as_bincode_bidi<InT, OutT>(
377        &self,
378        external_port_id: ExternalPortId,
379    ) -> impl Future<Output = DynSourceSink<OutT, InT, Error>> + 'a
380    where
381        InT: Serialize + 'static,
382        OutT: DeserializeOwned + 'static;
383
384    fn as_bincode_sink<T>(
385        &self,
386        external_port_id: ExternalPortId,
387    ) -> impl Future<Output = Pin<Box<dyn Sink<T, Error = Error>>>> + 'a
388    where
389        T: Serialize + 'static;
390
391    fn as_bincode_source<T>(
392        &self,
393        external_port_id: ExternalPortId,
394    ) -> impl Future<Output = Pin<Box<dyn Stream<Item = T>>>> + 'a
395    where
396        T: DeserializeOwned + 'static;
397}
hydro_lang/compile/deploy_provider.rs

hydro_lang/compile/
deploy_provider.rs
