aion-rs 0.3.0

Transport-agnostic Aion workflow engine with durability, replay, timers, and supervision.
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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

//! Engine shutdown gating: lifecycle operations are counted in, shutdown
//! closes the gate and waits for the count to drain before tearing the
//! runtime down.

use std::sync::{Arc, Condvar, Mutex};

use crate::EngineError;

#[derive(Clone, Default)]
pub(super) struct ShutdownGate {
    inner: Arc<ShutdownGateInner>,
}

#[derive(Default)]
struct ShutdownGateInner {
    state: Mutex<ShutdownState>,
    idle: Condvar,
}

#[derive(Default)]
struct ShutdownState {
    shutting_down: bool,
    active_operations: usize,
}

impl ShutdownGate {
    pub(super) fn begin_start(&self) -> Result<LifecycleOperation, EngineError> {
        let mut state = self.state()?;
        if state.shutting_down {
            return Err(EngineError::ShuttingDown);
        }
        state.active_operations += 1;
        Ok(LifecycleOperation {
            inner: Arc::clone(&self.inner),
        })
    }

    pub(super) fn begin_operation(&self) -> Result<LifecycleOperation, EngineError> {
        let mut state = self.state()?;
        // Refuse like `begin_start`: an operation admitted while
        // `close_and_wait` is draining would prolong the drain arbitrarily,
        // and the engine it would act on is already tearing down.
        if state.shutting_down {
            return Err(EngineError::ShuttingDown);
        }
        state.active_operations += 1;
        Ok(LifecycleOperation {
            inner: Arc::clone(&self.inner),
        })
    }

    pub(super) fn close_and_wait(&self) -> Result<(), EngineError> {
        let mut state = self.state()?;
        state.shutting_down = true;
        while state.active_operations > 0 {
            state = self
                .inner
                .idle
                .wait(state)
                .map_err(|_| EngineError::RegistryPoisoned)?;
        }
        Ok(())
    }

    fn state(&self) -> Result<std::sync::MutexGuard<'_, ShutdownState>, EngineError> {
        self.inner
            .state
            .lock()
            .map_err(|_| EngineError::RegistryPoisoned)
    }
}

pub(super) struct LifecycleOperation {
    inner: Arc<ShutdownGateInner>,
}

impl Drop for LifecycleOperation {
    fn drop(&mut self) {
        if let Ok(mut state) = self.inner.state.lock() {
            state.active_operations = state.active_operations.saturating_sub(1);
            if state.active_operations == 0 {
                self.inner.idle.notify_all();
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::ShutdownGate;
    use crate::EngineError;

    type TestResult = Result<(), Box<dyn std::error::Error>>;

    #[test]
    fn operations_are_refused_once_shutdown_begins() -> TestResult {
        let gate = ShutdownGate::default();
        // Admitted while open; the gate drains to idle on drop.
        drop(gate.begin_operation()?);
        gate.close_and_wait()?;

        assert!(matches!(
            gate.begin_operation(),
            Err(EngineError::ShuttingDown)
        ));
        assert!(matches!(gate.begin_start(), Err(EngineError::ShuttingDown)));
        Ok(())
    }

    #[test]
    fn close_waits_for_admitted_operations_to_drain() -> TestResult {
        let gate = ShutdownGate::default();
        let operation = gate.begin_operation()?;
        let closer = {
            let gate = gate.clone();
            std::thread::spawn(move || gate.close_and_wait())
        };
        // The closer parks until the admitted operation drops.
        std::thread::sleep(std::time::Duration::from_millis(20));
        assert!(!closer.is_finished());
        drop(operation);
        closer
            .join()
            .map_err(|_| "close_and_wait thread panicked")??;
        Ok(())
    }
}