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

harmos_runtime/operations/
scope.rs
