harmos_runtime/operations/
scope.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::collections::BTreeMap;
18use std::sync::atomic::{AtomicBool, Ordering};
19use std::sync::{Arc, Mutex, Weak};
20// Third Party Imports
21use tokio::sync::watch;
22// Local Imports
23use super::lane::Admission;
24use crate::operations::OperationInstanceId;
25use crate::operations::instances::{Instances, Reservation};
26use crate::state::Position;
27use crate::supervision::instance::{Instance, Phase, Status};
28
29/* === Constants === */
30
31/* === Macros === */
32
33/* === Type Aliases === */
34
35/* === Structs === */
36
37#[derive(Clone, Copy)]
38pub struct Scope<'a> {
39    core: &'a ScopeCore,
40}
41
42#[derive(Clone)]
43pub(crate) struct ScopeCore {
44    instances: Arc<Instances>,
45    parent: Weak<Instance>,
46    admission: Admission,
47    cause: Option<Position>,
48    children: Arc<Mutex<BTreeMap<OperationInstanceId, Child>>>,
49    closed: Arc<AtomicBool>,
50}
51
52pub(crate) struct ClosedScope {
53    children: Vec<Child>,
54}
55
56pub(crate) struct Child {
57    settled: watch::Receiver<Status>,
58    control: Arc<Instance>,
59}
60
61/* === Enums === */
62
63/* === Traits === */
64
65/* === Implementations === */
66
67impl<'a> Scope<'a> {
68    pub(crate) fn owned(&self) -> ScopeCore {
69        self.core.clone()
70    }
71
72    pub(crate) fn over(core: &'a ScopeCore) -> Self {
73        Self { core }
74    }
75
76    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
77    pub(crate) fn original_control(&self) -> Weak<Instance> {
78        self.core.parent.clone()
79    }
80
81    pub(crate) fn reserve_tracked<O>(
82        &self, id: OperationInstanceId, lane: &'static str, submitted: tokio::time::Instant, privileged: bool,
83        retired: impl Fn() + Send + Sync + 'static,
84    ) -> crate::Result<Reservation<O>>
85    where
86        O: Send + 'static,
87    {
88        let mut tracked = self.core.children.lock().expect("scope child lock is healthy");
89        if self.core.closed.load(Ordering::Acquire) {
90            return Err(crate::Error::Closing { domain: "operation scope" });
91        }
92        let children = Arc::downgrade(&self.core.children);
93        let preparation = self.core.instances.reserve(id, lane, submitted, privileged, move || {
94            if let Some(children) = children.upgrade() {
95                let removed = children.lock().expect("scope child lock is healthy").remove(&id);
96                drop(removed);
97            }
98            retired();
99        })?;
100        let child = Child::new(preparation.control());
101        if let Some(parent) = self.core.parent.upgrade() {
102            parent.track(&child.settlement());
103        }
104        tracked.insert(id, child);
105        Ok(preparation.inherit(self.core.cause))
106    }
107
108    pub fn now(&self) -> tokio::time::Instant {
109        crate::supervision::executor::now()
110    }
111
112    pub fn resources<A: crate::Application>(&self) -> &'a A::Resources {
113        self.core.instances.resources_of::<A>()
114    }
115
116    pub fn records<A: crate::Application>(&self) -> crate::records::RecordManager<A> {
117        crate::records::RecordManager::new(self.state::<A>())
118    }
119
120    pub fn storage<A: crate::Application>(&self) -> Option<&'a crate::storage::StorageManager<crate::storage::ErasedCatalog>> {
121        self.core.instances.handles_of::<A>().storage.get()
122    }
123
124    pub fn streams<A: crate::Application>(&self) -> crate::streams::StreamManager<A> {
125        self.core.instances.handles_of::<A>().streams.clone()
126    }
127
128    pub fn evidence<A: crate::Application>(&self) -> crate::evidence::Evidence<A> {
129        self.core.instances.handles_of::<A>().evidence.clone()
130    }
131
132    pub fn state<A: crate::Application>(&self) -> crate::state::StateManager<A> {
133        let journal = self.core.instances.state_of::<A>();
134        match self.core.cause {
135            Some(cause) => journal.caused_by(cause),
136            None => journal,
137        }
138    }
139}
140
141impl std::fmt::Debug for Scope<'_> {
142    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
143        formatter.debug_struct("Scope").field("lane", &self.core.admission.name()).field("cause", &self.core.cause).finish()
144    }
145}
146
147impl ScopeCore {
148    pub(crate) fn new(instances: Arc<Instances>, admission: Admission, cause: Option<Position>, parent: Weak<Instance>) -> Self {
149        Self {
150            instances,
151            parent,
152            admission,
153            cause,
154            children: Arc::new(Mutex::new(BTreeMap::new())),
155            closed: Arc::new(AtomicBool::new(false)),
156        }
157    }
158
159    pub(crate) fn close(self) -> ClosedScope {
160        let children: Vec<_> = {
161            let mut children = self.children.lock().expect("scope child lock is healthy");
162            self.closed.store(true, Ordering::Release);
163            std::mem::take(&mut *children).into_values().collect()
164        };
165        for child in &children {
166            child.control.cancel();
167        }
168
169        ClosedScope { children }
170    }
171}
172
173impl ClosedScope {
174    pub(crate) async fn drain(&mut self) {
175        for child in &mut self.children {
176            let _ = child.settled.wait_for(|status| status.phase() == Phase::Settled).await;
177        }
178    }
179}
180
181impl Child {
182    pub(crate) fn settlement(&self) -> Arc<Instance> {
183        Arc::clone(&self.control)
184    }
185
186    pub(crate) fn new(control: Arc<Instance>) -> Self {
187        let settled = control.settlement();
188        Self { settled, control }
189    }
190}
191
192/* === Public Functions === */
193
194/* === Private Functions === */
195
196/* === Tests === */
harmos_runtime/operations/scope.rs

harmos_runtime/operations/
scope.rs
