rstango 0.3.0

Implementation of the Tango distributed control system
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

//! Stopping the servers of this process on the signals it is asked to terminate with, see
//! [`super::ServerBuilder::stop_on_signal`].
//!
//! There is one handler for the whole process, installed by the first server that asks for
//! it, and it stops every server that asked, the same way the `Kill` command of an admin
//! device does: the thread waiting on each server then deletes its devices. A signal only
//! ever stops them, a second one stopping them again, which does nothing more.

use std::sync::{Arc, Mutex, OnceLock, Weak};

use futures::executor::block_on;
use futures::future::join_all;
use tracing::{info, warn};

use crate::error::{Error, ErrorKind, Result};

use super::server::Server;


/// The servers to stop on a signal, held weakly: a server that is gone has nothing left to
/// stop, and is pruned on the next registration.
static SERVERS: Mutex<Vec<Weak<Server>>> = Mutex::new(Vec::new());

/// Whether the handler was installed, which is only ever tried once: what it failed with is
/// kept, so that every server asking for it fails alike.
static INSTALLED: OnceLock<std::result::Result<(), String>> = OnceLock::new();

/// Install the handler of this process, if it is not already, failing if it cannot be, an
/// application having installed one of its own above all.
pub fn install() -> Result<()> {
    INSTALLED.get_or_init(|| ctrlc::set_handler(stop_all).map_err(|err| err.to_string()))
        .clone()
        .map_err(|err| Error::new(ErrorKind::Internal)
            .with_description(format!("Cannot stop the server on signals, which can be disabled on its builder: {err}")))
}

/// Stop that server on the next signal, see [`install`], which must have been done.
pub fn register(server: &Arc<Server>) {
    let mut servers = SERVERS.lock().unwrap();
    servers.retain(|server| server.strong_count() > 0);
    servers.push(Arc::downgrade(server));
}

/// The handler, run on a thread of its own, which may therefore block.
fn stop_all() {

    let servers = SERVERS.lock().unwrap().iter()
        .filter_map(Weak::upgrade)
        .collect::<Vec<_>>();

    info!(servers = servers.len(), "Stopping on a termination signal");

    let stopped = block_on(join_all(servers.iter().map(|server| server.stop())));
    for (server, stopped) in servers.iter().zip(stopped) {
        if let Err(err) = stopped {
            warn!(server = %server.name(), %err, "Failed to unexport the server stopped on a signal");
        }
    }

}