dope_core/platform/
signal.rs

1use std::{cell, marker, mem, os::fd};
2
3use dope_sys::thread::{
4    Thread,
5    signal::{self, AcquireFault, ReleaseFault, Signal},
6};
7use o3::{base::marker::ThreadBound, mem::slot};
8
9use crate::{Backend, Errno};
10
11pub type Receiver = <Backend as Thread>::Receiver;
12
13#[derive(Debug)]
14pub enum Error<'owner, E> {
15    Acquire(Acquire),
16    Run(E),
17    Release {
18        run: Option<E>,
19        cause: Release<'owner>,
20    },
21}
22
23thread_local! {
24    static CLAIMED: cell::Cell<bool> = const { cell::Cell::new(false) };
25}
26
27/// A native signal policy that could not yet be restored on its owning thread.
28#[derive(Debug)]
29#[must_use = "the signal policy must be restored before another scope can start"]
30pub struct Release<'owner> {
31    pub fault: ReleaseFault,
32    receiver: slot::Occupied<'owner, Receiver>,
33}
34
35impl Release<'_> {
36    pub fn retry(self) -> Result<(), Self> {
37        release(self.receiver)
38    }
39}
40
41fn release(receiver: slot::Occupied<'_, Receiver>) -> Result<(), Release<'_>> {
42    match signal::raw::Receiver::restore(&*receiver) {
43        Ok(()) => {
44            drop(receiver);
45            CLAIMED.with(|claimed| claimed.set(false));
46            Ok(())
47        }
48        Err(fault) => Err(Release { fault, receiver }),
49    }
50}
51
52type State<'process> = (
53    <Receiver as signal::raw::Receiver>::Source<'process>,
54    marker::PhantomData<(cell::Cell<&'process ()>, ThreadBound)>,
55);
56
57/// Linear process-lifetime view of the termination signal receiver.
58#[must_use]
59pub struct Source<'process> {
60    read: State<'process>,
61}
62
63impl Signal for Source<'_> {
64    type Error = <Receiver as signal::raw::Receiver>::Wait;
65
66    fn wait(&self) -> Result<(), Self::Error> {
67        self.read.0.wait()
68    }
69}
70
71impl Source<'_> {
72    /// Enters the exclusive signal scope on the process's original thread.
73    /// Call during single-threaded startup. Other code must not change SIGTERM
74    /// handling while `run` executes; masks on other existing threads are not updated.
75    /// Workers must start inside `run` and finish before the policy is restored.
76    pub fn scope<'owner, R, E>(
77        retained: slot::Vacant<'owner, Receiver>,
78        run: impl for<'process> FnOnce(Source<'process>) -> Result<R, E>,
79    ) -> Result<R, Error<'owner, E>> {
80        if !<Backend as Thread>::is_main() {
81            return Err(Error::Acquire(Acquire::Thread));
82        }
83        if CLAIMED.with(|claimed| claimed.replace(true)) {
84            return Err(Error::Acquire(Acquire::Busy));
85        }
86        // SAFETY: this claim excludes every other signal scope until successful
87        // restoration. The receiver stays on this thread through release.
88        let receiver = match unsafe { <Receiver as signal::raw::Receiver>::try_new() } {
89            Ok(receiver) => retained.insert(receiver),
90            Err(error) => {
91                CLAIMED.with(|claimed| claimed.set(false));
92                return Err(Error::Acquire(Acquire::from_native(error.into())));
93            }
94        };
95        let source = Source {
96            read: (
97                signal::raw::Receiver::source(&*receiver),
98                marker::PhantomData,
99            ),
100        };
101        let outcome = run(source);
102        match release(receiver) {
103            Ok(()) => outcome.map_err(Error::Run),
104            Err(cause) => Err(Error::Release {
105                run: outcome.err(),
106                cause,
107            }),
108        }
109    }
110}
111
112impl fd::AsFd for Source<'_> {
113    fn as_fd(&self) -> fd::BorrowedFd<'_> {
114        self.read.0.as_fd()
115    }
116}
117
118const _: () = {
119    assert!(mem::size_of::<Source<'static>>() == mem::size_of::<fd::BorrowedFd<'static>>());
120    assert!(mem::align_of::<Source<'static>>() == mem::align_of::<fd::BorrowedFd<'static>>());
121    assert!(
122        mem::size_of::<Option<Source<'static>>>()
123            == mem::size_of::<Option<fd::BorrowedFd<'static>>>()
124    );
125};
126
127#[derive(Debug)]
128pub enum Acquire {
129    Thread,
130    Busy,
131    MaskBlock(Errno),
132    SignalFd(Errno),
133    QueueOpen(Errno),
134    QueueCloexec(Errno),
135    Register(Errno),
136    Ignore(Errno),
137}
138
139impl Acquire {
140    fn from_native(error: AcquireFault) -> Self {
141        match error {
142            AcquireFault::MaskBlock(error) => Self::MaskBlock(error),
143            AcquireFault::SignalFd(error) => Self::SignalFd(error),
144            AcquireFault::QueueOpen(error) => Self::QueueOpen(error),
145            AcquireFault::QueueCloexec(error) => Self::QueueCloexec(error),
146            AcquireFault::Register(error) => Self::Register(error),
147            AcquireFault::Ignore(error) => Self::Ignore(error),
148        }
149    }
150}
dope_core/platform/signal.rs

dope_core/platform/
signal.rs
