harmos-runtime 0.1.4

Harmos runtime: managed state, undo, recovery, background work and typed streams.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

/* ***
 *
 * Harmos
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * ***
 */

/* === Modules === */

/* === Re-Exports === */

/* === Imports === */

// Standard Library Imports

// Third Party Imports

// Local Imports
use super::*;

/* === Constants === */

/* === Macros === */

/* === Type Aliases === */

/* === Structs === */

/* === Enums === */

/* === Traits === */

/* === Implementations === */

impl OperationManager {
    pub(crate) fn close_admission(&self) {
        self.close_entries(true);
    }

    pub(crate) fn close(&self) {
        self.close_entries(false);
    }

    pub(super) fn close_entries(&self, _retain_shutdown: bool) {
        let mut state = self.registry.state.lock().expect("operation registry lock is healthy");
        if _retain_shutdown && state.closing {
            return;
        }
        state.closing = true;
        let retired = std::mem::take(&mut state.cache);
        #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
        let staging: Vec<_> = state.staging.iter().filter_map(std::sync::Weak::upgrade).collect();
        for entry in state.entries.values().flat_map(|providers| providers.values()) {
            let _pending = entry.pending.lock().expect("operation settlement lock is healthy");
            if _retain_shutdown && entry.retirement_control() {
                continue;
            }
            if _retain_shutdown && entry.active.load(Ordering::Acquire) {
                entry.retirement.store(Retirement::Runtime as u8, Ordering::Release);
            }
            entry.active.store(false, Ordering::Release);
            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
            let ordinary = entry.sdk.is_none();
            #[cfg(not(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar")))]
            let ordinary = true;
            if _retain_shutdown && ordinary {
                for control in _pending.values() {
                    control.cancel();
                }
            }
        }
        drop(state);
        drop(retired);
        #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
        for scope in staging {
            scope.close_staging();
        }
    }

    pub async fn shutdown(&self) -> Result<()> {
        self.close();
        self.instances.close();
        let entries: Vec<_> = self
            .registry
            .state
            .lock()
            .expect("operation registry lock is healthy")
            .entries
            .values()
            .flat_map(|providers| providers.values())
            .cloned()
            .collect();
        for entry in entries {
            settle(&entry).await?;
        }
        self.instances.settle().await?;
        let retired = std::mem::take(&mut self.registry.state.lock().expect("operation registry lock is healthy").entries);
        drop(retired);
        Ok(())
    }
}

/* === Public Functions === */

/* === Private Functions === */

/* === Tests === */