harmos_runtime/operations/manager/
shutdown.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
17
18// Third Party Imports
19
20// Local Imports
21use super::*;
22
23/* === Constants === */
24
25/* === Macros === */
26
27/* === Type Aliases === */
28
29/* === Structs === */
30
31/* === Enums === */
32
33/* === Traits === */
34
35/* === Implementations === */
36
37impl OperationManager {
38    pub(crate) fn close_admission(&self) {
39        self.close_entries(true);
40    }
41
42    pub(crate) fn close(&self) {
43        self.close_entries(false);
44    }
45
46    pub(super) fn close_entries(&self, _retain_shutdown: bool) {
47        let mut state = self.registry.state.lock().expect("operation registry lock is healthy");
48        if _retain_shutdown && state.closing {
49            return;
50        }
51        state.closing = true;
52        let retired = std::mem::take(&mut state.cache);
53        #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
54        let staging: Vec<_> = state.staging.iter().filter_map(std::sync::Weak::upgrade).collect();
55        for entry in state.entries.values().flat_map(|providers| providers.values()) {
56            let _pending = entry.pending.lock().expect("operation settlement lock is healthy");
57            if _retain_shutdown && entry.retirement_control() {
58                continue;
59            }
60            if _retain_shutdown && entry.active.load(Ordering::Acquire) {
61                entry.retirement.store(Retirement::Runtime as u8, Ordering::Release);
62            }
63            entry.active.store(false, Ordering::Release);
64            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
65            let ordinary = entry.sdk.is_none();
66            #[cfg(not(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar")))]
67            let ordinary = true;
68            if _retain_shutdown && ordinary {
69                for control in _pending.values() {
70                    control.cancel();
71                }
72            }
73        }
74        drop(state);
75        drop(retired);
76        #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
77        for scope in staging {
78            scope.close_staging();
79        }
80    }
81
82    pub async fn shutdown(&self) -> Result<()> {
83        self.close();
84        self.instances.close();
85        let entries: Vec<_> = self
86            .registry
87            .state
88            .lock()
89            .expect("operation registry lock is healthy")
90            .entries
91            .values()
92            .flat_map(|providers| providers.values())
93            .cloned()
94            .collect();
95        for entry in entries {
96            settle(&entry).await?;
97        }
98        self.instances.settle().await?;
99        let retired = std::mem::take(&mut self.registry.state.lock().expect("operation registry lock is healthy").entries);
100        drop(retired);
101        Ok(())
102    }
103}
104
105/* === Public Functions === */
106
107/* === Private Functions === */
108
109/* === Tests === */
harmos_runtime/operations/manager/shutdown.rs

harmos_runtime/operations/manager/
shutdown.rs
