hydro_lang/location/
process.rs

1//! Definition of the [`Process`] location type, representing a single-node
2//! compute location in a distributed Hydro program.
3//!
4//! A [`Process`] is the simplest kind of location: it corresponds to exactly one
5//! machine (or OS process) and all live collections placed on it are materialized
6//! on that single node. Use a process when the computation does not need to be
7//! replicated or partitioned across multiple nodes.
8//!
9//! Processes are created via [`FlowBuilder::process`](crate::compile::builder::FlowBuilder::process)
10//! and are parameterized by a **tag type** (`ProcessTag`) that lets the type
11//! system distinguish different processes at compile time.
12
13use std::fmt::{Debug, Formatter};
14use std::marker::PhantomData;
15
16use super::{Location, LocationId};
17use crate::compile::builder::FlowState;
18use crate::location::{LocationKey, TopLevel};
19use crate::staging_util::Invariant;
20
21/// A single-node location in a distributed Hydro program.
22///
23/// `Process` represents exactly one machine (or OS process) and is one of the
24/// core location types that implements the [`Location`] trait. Live collections
25/// placed on a `Process` are materialized entirely on that single node.
26///
27/// The type parameter `ProcessTag` is a compile-time marker that differentiates
28/// distinct processes in the same dataflow graph (e.g. `Process<'a, Leader>` vs
29/// `Process<'a, Follower>`). It defaults to `()` when only one process is
30/// needed.
31///
32/// # Creating a Process
33/// ```rust
34/// # #[cfg(feature = "build")]
35/// # {
36/// # use hydro_lang::prelude::*;
37/// struct MyTag;
38/// let mut flow = FlowBuilder::new();
39/// let node = flow.process::<MyTag>();
40/// # let _ = &node;
41/// # let _ = flow.finalize();
42/// # }
43/// ```
44pub struct Process<'a, ProcessTag = ()> {
45    pub(crate) key: LocationKey,
46    pub(crate) flow_state: FlowState,
47    pub(crate) _phantom: Invariant<'a, ProcessTag>,
48}
49
50impl<P> Debug for Process<'_, P> {
51    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
52        write!(f, "Process({})", self.key)
53    }
54}
55
56impl<P> Eq for Process<'_, P> {}
57impl<P> PartialEq for Process<'_, P> {
58    fn eq(&self, other: &Self) -> bool {
59        self.key == other.key && FlowState::ptr_eq(&self.flow_state, &other.flow_state)
60    }
61}
62
63impl<P> Clone for Process<'_, P> {
64    fn clone(&self) -> Self {
65        Process {
66            key: self.key,
67            flow_state: self.flow_state.clone(),
68            _phantom: PhantomData,
69        }
70    }
71}
72
73impl<'a, P> super::dynamic::DynLocation for Process<'a, P> {
74    fn dyn_id(&self) -> LocationId {
75        LocationId::Process(self.key)
76    }
77
78    fn flow_state(&self) -> &FlowState {
79        &self.flow_state
80    }
81
82    fn is_top_level() -> bool {
83        true
84    }
85
86    fn multiversioned(&self) -> bool {
87        false // processes are always single-versioned
88    }
89
90    fn cluster_consistency() -> Option<super::dynamic::ClusterConsistency> {
91        None
92    }
93}
94
95impl<'a, P> Location<'a> for Process<'a, P> {
96    type Root = Self;
97
98    type SimHookScope = crate::sim_hooks::OnProcess<P>;
99
100    type DropConsistency = Self;
101
102    fn consistency() -> Option<super::dynamic::ClusterConsistency> {
103        None
104    }
105
106    fn root(&self) -> Self::Root {
107        self.clone()
108    }
109
110    fn drop_consistency(&self) -> Self::DropConsistency {
111        self.clone()
112    }
113
114    fn from_drop_consistency(l2: Self::DropConsistency) -> Self {
115        l2
116    }
117}
118
119impl<'a, P> TopLevel<'a> for Process<'a, P> {}
hydro_lang/location/process.rs

hydro_lang/location/
process.rs
