hydro_lang/location/
dynamic.rs

1//! Definitions for interacting with locations using an untyped interface.
2//!
3//! Under the hood, locations are associated with a [`LocationId`] value that
4//! uniquely identifies the location. Manipulating these values is useful for
5//! observability and transforming the Hydro IR.
6
7use serde::{Deserialize, Serialize};
8
9use super::LocationKey;
10use crate::compile::builder::ClockId;
11#[cfg(stageleft_runtime)]
12use crate::compile::{
13    builder::FlowState,
14    ir::{CollectionKind, HydroIrMetadata},
15};
16use crate::location::LocationType;
17
18/// An enumeration representing the consistency guarantee of a live collection on a cluster.
19#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Hash, Serialize, Deserialize)]
20pub enum ClusterConsistency {
21    /// No consistency is guaranteed, see [`super::cluster::NoConsistency`].
22    NoConsistency,
23    /// Eventual consistency is guaranteed, see [`super::cluster::EventualConsistency`].
24    EventualConsistency,
25}
26
27/// An enumeration representing a location heirarchy, including "virtual" locations (atomic/tick).
28#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Hash, Serialize, Deserialize)]
29pub enum LocationId {
30    /// A process root location (i.e. a single node).
31    Process(LocationKey),
32    /// A cluster root location (i.e. multiple nodes).
33    Cluster(LocationKey),
34    /// An atomic region, within a tick.
35    Atomic(
36        /// The tick that the atomic region is associated with.
37        Box<LocationId>,
38    ),
39    /// A tick within a location.
40    Tick {
41        /// The `ClockId` of this tick, or `None` if `parent_location` is an `Atomic`
42        tick: Option<ClockId>,
43        /// What location this tick is within.
44        parent_location: Box<LocationId>,
45    },
46}
47
48/// Implement Debug to Display-print the key, reduces snapshot verbosity.
49impl std::fmt::Debug for LocationId {
50    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
51        match self {
52            LocationId::Process(key) => write!(f, "Process({key})"),
53            LocationId::Cluster(key) => write!(f, "Cluster({key})"),
54            LocationId::Atomic(tick) => write!(f, "Atomic({tick:?})"),
55            LocationId::Tick {
56                tick,
57                parent_location,
58            } => write!(f, "Tick({tick:?}, {parent_location:?})"),
59        }
60    }
61}
62
63impl LocationId {
64    /// The [`LocationType`] of this location ID. `None` if this is not a root location.
65    pub fn location_type(&self) -> Option<LocationType> {
66        match self {
67            LocationId::Process(_) => Some(LocationType::Process),
68            LocationId::Cluster(_) => Some(LocationType::Cluster),
69            _ => None,
70        }
71    }
72}
73
74#[expect(missing_docs, reason = "TODO")]
75impl LocationId {
76    pub fn root(&self) -> &LocationId {
77        match self {
78            LocationId::Process(_) => self,
79            LocationId::Cluster(_) => self,
80            LocationId::Atomic(tick) => tick.root(),
81            LocationId::Tick {
82                tick: _,
83                parent_location,
84            } => parent_location.root(),
85        }
86    }
87
88    pub fn is_root(&self) -> bool {
89        match self {
90            LocationId::Process(_) | LocationId::Cluster(_) => true,
91            LocationId::Atomic(_) => false,
92            LocationId::Tick { .. } => false,
93        }
94    }
95
96    pub fn is_top_level(&self) -> bool {
97        match self {
98            LocationId::Process(_) | LocationId::Cluster(_) => true,
99            LocationId::Atomic(_) => true,
100            LocationId::Tick { .. } => false,
101        }
102    }
103
104    pub fn key(&self) -> LocationKey {
105        match self {
106            LocationId::Process(id) => *id,
107            LocationId::Cluster(id) => *id,
108            LocationId::Atomic(_) => panic!("cannot get raw id for atomic"),
109            LocationId::Tick { .. } => panic!("cannot get raw id for tick"),
110        }
111    }
112
113    pub fn swap_root(&mut self, new_root: LocationId) {
114        match self {
115            LocationId::Tick {
116                tick: _,
117                parent_location,
118            } => {
119                parent_location.swap_root(new_root);
120            }
121            LocationId::Atomic(tick) => {
122                tick.swap_root(new_root);
123            }
124            _ => {
125                assert!(new_root.is_root());
126                *self = new_root;
127            }
128        }
129    }
130
131    pub fn new_node_metadata(
132        self,
133        collection_kind: CollectionKind,
134        consistency: Option<ClusterConsistency>,
135    ) -> HydroIrMetadata {
136        use crate::compile::ir::HydroIrOpMetadata;
137        use crate::compile::ir::backtrace::Backtrace;
138
139        HydroIrMetadata {
140            location_id: self,
141            collection_kind,
142            cardinality: None,
143            tag: None,
144            consistency,
145            op: HydroIrOpMetadata {
146                backtrace: Backtrace::get_backtrace(3),
147                cpu_usage: None,
148                network_recv_cpu_usage: None,
149                id: None,
150                sim_hook_id: None,
151            },
152        }
153    }
154}
155
156#[cfg(stageleft_runtime)]
157pub(crate) trait DynLocation: Clone {
158    fn dyn_id(&self) -> LocationId;
159
160    fn flow_state(&self) -> &FlowState;
161    fn is_top_level() -> bool;
162    fn multiversioned(&self) -> bool;
163    fn cluster_consistency() -> Option<ClusterConsistency>;
164
165    fn new_node_metadata(&self, collection_kind: CollectionKind) -> HydroIrMetadata {
166        self.dyn_id()
167            .new_node_metadata(collection_kind, Self::cluster_consistency())
168    }
169}
hydro_lang/location/dynamic.rs

hydro_lang/location/
dynamic.rs
