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 records<A: crate::Application>(&self) -> crate::records::RecordManager<A> {
113        crate::records::RecordManager::new(self.state::<A>())
114    }
115
116    pub fn storage<A: crate::Application>(&self) -> Option<&'a crate::storage::StorageManager<crate::storage::ErasedCatalog>> {
117        self.core.instances.handles_of::<A>().storage.get()
118    }
119
120    pub fn streams<A: crate::Application>(&self) -> crate::streams::StreamManager<A> {
121        self.core.instances.handles_of::<A>().streams.clone()
122    }
123
124    pub fn state<A: crate::Application>(&self) -> crate::state::StateManager<A> {
125        let journal = self.core.instances.state_of::<A>();
126        match self.core.cause {
127            Some(cause) => journal.caused_by(cause),
128            None => journal,
129        }
130    }
131}
132
133impl std::fmt::Debug for Scope<'_> {
134    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
135        formatter.debug_struct("Scope").field("lane", &self.core.admission.name()).field("cause", &self.core.cause).finish()
136    }
137}
138
139impl ScopeCore {
140    pub(crate) fn new(instances: Arc<Instances>, admission: Admission, cause: Option<Position>, parent: Weak<Instance>) -> Self {
141        Self {
142            instances,
143            parent,
144            admission,
145            cause,
146            children: Arc::new(Mutex::new(BTreeMap::new())),
147            closed: Arc::new(AtomicBool::new(false)),
148        }
149    }
150
151    pub(crate) fn close(self) -> ClosedScope {
152        let children: Vec<_> = {
153            let mut children = self.children.lock().expect("scope child lock is healthy");
154            self.closed.store(true, Ordering::Release);
155            std::mem::take(&mut *children).into_values().collect()
156        };
157        for child in &children {
158            child.control.cancel();
159        }
160
161        ClosedScope { children }
162    }
163}
164
165impl ClosedScope {
166    pub(crate) async fn drain(&mut self) {
167        for child in &mut self.children {
168            let _ = child.settled.wait_for(|status| status.phase() == Phase::Settled).await;
169        }
170    }
171}
172
173impl Child {
174    pub(crate) fn settlement(&self) -> Arc<Instance> {
175        Arc::clone(&self.control)
176    }
177
178    pub(crate) fn new(control: Arc<Instance>) -> Self {
179        let settled = control.settlement();
180        Self { settled, control }
181    }
182}
183
184/* === Public Functions === */
185
186/* === Private Functions === */
187
188/* === Tests === */
harmos_runtime/operations/scope.rs

harmos_runtime/operations/
scope.rs
