Rust 异步信号处理与 async 运行时交互机制：从 Unix 信号到生产级 graceful shutdown

在 Rust async 程序中，SIGTERM 和 SIGINT 的处理远不只是 "注册一个 ctrl+c handler" 那么简单。当 Unix 信号的异步投递遇上 Tokio 的 work-stealing 调度器，你可能会遇到 mutex 在中持锁期间被信号中断、Tokio runtime 在 drop 时任务未正常清理、信号风暴导致死锁等一系列生产级问题。本文从 Unix 信号语义出发，深入剖析 Rust async 生态中信号处理的陷阱与最佳实践。

---

一、Unix 信号的基础语义

Unix 信号（Signal）是一种异步通知机制，用于通知进程发生了某种事件。每个信号都有一个默认动作：终止、终止并转储核心、忽略或暂停进程。

信号类型       编号    默认动作      典型触发来源
─────────────────────────────────────────────────────
SIGHUP         1      终止          终端断开 / 配置重载
SIGINT         2      终止          Ctrl+C
SIGQUIT        3      终止+core     Ctrl+\
SIGKILL        9      终止(不可捕获) kill -9
SIGTERM        15     终止          kill(默认)
SIGUSR1        10     终止          用户自定义
SIGUSR2        12     终止          用户自定义
SIGPIPE        13     终止          写读端关闭的管道/套接字

信号的关键特性是"异步投递"——内核会在进程从内核态返回用户态时，或者在特定调度点检查待处理的信号。这意味着信号可能打断正在执行的任何用户态代码。

信号的同步 vs 异步

并非所有信号都是异步的。有些信号是同步产生的，可以精确定位到具体指令：

• 异步信号：SIGINT、SIGTERM、SIGHUP 等，由外部事件触发
• 同步信号：SIGSEGV、SIGBUS、SIGFPE、SIGILL 等，由具体指令错误触发

对于本文讨论的"信号处理"，我们主要关注异步信号，特别是 graceful shutdown 场景中的 SIGTERM 和 SIGINT。

二、为什么 Rust async 的信号处理是困难的

2.1 传统同步程序的处理模式

在传统同步程序中，信号处理相对直接：

// C 语言中的经典信号处理
volatile sig_atomic_t shutdown_flag = 0;

void handle_signal(int sig) {
    shutdown_flag = 1;
}

int main() {
    signal(SIGTERM, handle_signal);
    signal(SIGINT, handle_signal);

    while (!shutdown_flag) {
        // 主循环
        do_work();
    }
    cleanup();
}

这种模式的核心是：signal handler 只设置一个 sig_atomic_t 标志，主循环定期检查。这是安全的，因为 sig_atomic_t 的读写在 POSIX 上是原子的。

2.2 async 世界的问题

在 Rust async 编程中，以下因素使信号处理变得复杂：

1. 没有明确的"主循环"：async 任务没有传统意义上的 while true 循环等待信号
2. 任务可能在 .await 点被挂起：信号到来时，任务可能长时间不执行
3. Mutex 在中持锁时被信号中断：如果 async mutex 在持锁期间进程被 SIGKILL/SIGSTOP 以外的信号打断，可能导致锁状态不一致
4. Tokio runtime 的 drop 语义：runtime drop 时会取消所有运行中的任务，但取消也需要时间

2.3 信号投递的不确定性

在 Linux 上，信号投递遵循以下规则：

• 信号从内核态返回用户态前被检查和投递
• 如果进程所有线程都阻塞了某个信号，信号保持 pending
• 只有一个线程会接收到信号（随机选择或指定）
• 如果嵌套信号处理发生，可能使用独立信号栈（SA_ONSTACK）

这带来了一个关键问题：在多线程 Tokio runtime 中，SIGTERM 可能被投递到任意 worker 线程而非 driver 线程。

三、Tokio 的信号处理机制

3.1 tokio::signal 模块

Tokio 提供了 tokio::signal 模块来处理 Unix 和 Windows 信号：

use tokio::signal::unix::{signal, SignalKind};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 创建 SIGTERM 信号流
    let mut sigterm = signal(SignalKind::terminate())?;
    // 创建 SIGINT 信号流  
    let mut sigint = signal(SignalKind::interrupt())?;

    tokio::select! {
        _ = sigterm.recv() => {
            println!("收到 SIGTERM，开始优雅关闭");
        }
        _ = sigint.recv() => {
            println!("收到 SIGINT，开始优雅关闭");
        }
        _ = run_server() => {
            println!("服务器正常退出");
        }
    }

    Ok(())
}

3.2 Tokio signal 的内部实现

Tokio 的 signal 实现依赖以下机制：

1. Self-Pipe Trick：Tokio 使用一个管道（Linux 上可用 eventfd 替代），在信号处理函数的上下文中向其写入数据
2. Global Signal Handler：Tokio 在启动时注册一个全局的 sigaction 处理器，该处理器只做一件事——向 pipe/eventfd 写入一个字节
3. Async Reader：在 async 任务中通过 poll 读取 pipe/eventfd 来等待信号

这种设计的关键优势是：信号 handler 只做异步安全的写操作，真正的处理发生在 async 任务中，可以使用正常的 Rust 异步设施。

3.3 常见陷阱：双重信号导致强制退出

很多生产程序实现了"收到两次 Ctrl+C 强制退出"的逻辑，但这里有个细节值得注意：

// 危险模式
async fn wait_for_shutdown() -> ShutdownMode {
    let mut sigterm = signal(SignalKind::terminate()).unwrap();
    let mut sigint = signal(SignalKind::interrupt()).unwrap();

    let first_signal = tokio::select! {
        _ = sigterm.recv() => Signal::Term,
        _ = sigint.recv() => Signal::Int,
    };

    // 问题：等待第二个信号时，第一个信号的 handler 可能已经被 reset
    let second_timeout = tokio::time::timeout(
        Duration::from_secs(5),
        tokio::select! {
            _ = sigterm.recv() => Signal::Term,
            _ = sigint.recv() => Signal::Int,
        }
    );

    match second_timeout {
        Ok(_) => ShutdownMode::Forceful, // 用户再次按了 Ctrl+C
        Err(_) => ShutdownMode::Graceful,
    }
}

四、Channel 机制与信号的交互

4.1 为什么信号相关通信不用普通 Mutex

在异步系统中，一个经典的错误尝试是：

// 反模式：在 async 上下文中使用 std::sync::Mutex
static SHUTDOWN: AtomicBool = AtomicBool::new(false);

fn signal_handler() {
    SHUTDOWN.store(true, Ordering::SeqCst);  // 这是安全的
}

async fn worker_loop() {
    loop {
        if SHUTDOWN.load(Ordering::SeqCst) {
            break;
        }
        // 执行工作
    }
}

这在标记判断上没有问题，但 channel 更适合信号的"一次性通知"语义，因为：

1. 广播：信号通常是一个广播事件，channel 天然支持多消费者
2. 阻塞等待：channel 允许任务挂起等待信号，而非忙检查
3. Once语义：信号通常只需处理一次，tokio::sync::watch 和 tokio::sync::broadcast 天然支持一次性通知

4.2 使用 broadcast channel 实现 graceful shutdown

use tokio::sync::broadcast;

struct ShutdownCoordinator {
    notify: broadcast::Sender<()>,
}

impl ShutdownCoordinator {
    fn new(capacity: usize) -> Self {
        let (notify, _) = broadcast::channel(capacity);
        Self { notify }
    }

    fn subscribe(&self) -> broadcast::Receiver<()> {
        self.notify.subscribe()
    }

    fn shutdown(&self) {
        // 发送给所有订阅者，忽略无接收者的情况
        let _ = self.notify.send(());
    }
}

async fn worker(id: usize, mut shutdown_rx: broadcast::Receiver<()>) {
    loop {
        tokio::select! {
            _ = shutdown_rx.recv() => {
                log::info!("Worker {} received shutdown signal", id);
                // 执行清理逻辑
                break
            }
            _ = do_async_work() => {
                // 工作完成
            }
        }
    }
}

4.3 使用 watch channel 实现Cancellation Token

对于更复杂的场景，特别是需要传递 shutdown 原因时：

use tokio::sync::watch;

#[derive(Clone, Debug)]
pub enum ShutdownReason {
    Signal(SignalKind),
    AdminCommand(String),
    HealthCheckFailed(String),
}

pub struct CancellationToken {
    tx: watch::Sender<Option<ShutdownReason>>,
    rx: watch::Receiver<Option<ShutdownReason>>,
}

impl CancellationToken {
    pub fn new() -> Self {
        let (tx, rx) = watch::channel(None);
        Self { tx, rx }
    }

    pub fn token(&self) -> ChildToken {
        ChildToken {
            rx: self.rx.clone(),
        }
    }

    pub fn cancel(&self, reason: ShutdownReason) {
        let _ = self.tx.send(Some(reason));
    }
}

#[derive(Clone)]
pub struct ChildToken {
    rx: watch::Receiver<Option<ShutdownReason>>,
}

impl ChildToken {
    pub async fn cancelled(&mut self) -> ShutdownReason {
        // 如果已经取消，立即返回
        if let Some(reason) = self.rx.borrow().clone() {
            return reason;
        }
        // 等待取消信号
        self.rx.changed().await.unwrap();
        self.rx.borrow().clone().unwrap()
    }
}

五、Graceful Shutdown 的生产级实现

5.1 完整的 graceful shutdown 流程

一个生产级系统的 graceful shutdown 通常包含以下步骤：

1. 接收到 SIGTERM/SIGINT 信号
2. 停止接受新请求（关闭 listener）
3. 通知所有运行中的任务（发送 shutdown 信号）
4. 等待现有请求处理完成（设置超时）
5. 关闭数据库连接、Redis 连接等有状态资源
6. 刷新日志、metrics
7. 退出进程

5.2 生产级实现：完整的 shutdown orchestrator

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{broadcast, Mutex, Semaphore};
use tokio::time::timeout;

pub struct GracefulShutdown {
    // 通知所有组件开始 shutdown
    notify: broadcast::Sender<()>,
    // 跟踪正在处理的任务数
    in_flight: Arc<Semaphore>,
    // 最大并发限制
    max_in_flight: usize,
    // shutdown 超时
    timeout: Duration,
}

impl GracefulShutdown {
    pub fn new(max_in_flight: usize, shutdown_timeout: Duration) -> Self {
        let (notify, _) = broadcast::channel(1024);
        let in_flight = Arc::new(Semaphore::new(max_in_flight));

        Self {
            notify,
            in_flight,
            max_in_flight,
            timeout: shutdown_timeout,
        }
    }

    /// 创建一个新的工作许可（表示开始一个任务）
    pub async fn acquire_permit(&self) -> Result<WorkGuard, ShutdownError> {
        // try_acquire 在已经开始 shutdown 时快速失败
        match self.in_flight.try_acquire() {
            Ok(permit) => {
                let mut rx = self.notify.subscribe();
                Ok(WorkGuard {
                    _permit: permit,
                    shutdown_rx: rx,
                    shutdown_notify: self.notify.clone(),
                })
            }
            Err(_) => Err(ShutdownError::ShuttingDown),
        }
    }

    /// 等待所有工作中的任务完成后返回
    pub async fn wait_for_completion(&self) {
        // 尝试获取所有 permit，这意味着所有当前任务都已完成
        let _ = timeout(self.timeout, async {
            // 获取全部 permit
            let permits = self.in_flight
                .acquire_many(self.max_in_flight as u32)
                .await
                .expect("Failed to acquire permits");
            // 释放它们
            drop(permits);
        }).await;
    }

    /// 发送 shutdown 信号
    pub fn initiate_shutdown(&self) {
        log::info!("Graceful shutdown initiated");
        let _ = self.notify.send(());
    }

    pub fn subscribe(&self) -> broadcast::Receiver<()> {
        self.notify.subscribe()
    }
}

pub struct WorkGuard {
    _permit: tokio::sync::OwnedSemaphorePermit,
    shutdown_rx: broadcast::Receiver<()>,
    shutdown_notify: broadcast::Sender<()>,
}

impl WorkGuard {
    pub async fn cancelled(&mut self) {
        let _ = self.shutdown_rx.recv().await;
    }
}

#[derive(Debug)]
pub enum ShutdownError {
    ShuttingDown,
}

5.3 与 axum 集成的实践

use axum::{routing::get, Router, Extension, response::IntoResponse};
use std::sync::Arc;

async fn health_check() -> impl IntoResponse {
    "ok"
}

async fn handle_request(
    Extension(shutdown): Extension<Arc<GracefulShutdown>>,
) -> Result<String, &'static str> {
    let _guard = shutdown.acquire_permit().await.map_err(|_| "Server is shutting down")?;

    // 模拟长时间处理工作
    tokio::time::sleep(Duration::from_secs(2)).await;

    Ok("Request processed".to_string())
}

pub async fn run_server() {
    let shutdown = Arc::new(GracefulShutdown::new(100, Duration::from_secs(30)));

    // 克隆用于 signal handler
    let shutdown_for_signal = shutdown.clone();

    // 启动信号监听任务
    tokio::spawn(async move {
        let mut sigterm = signal(SignalKind::terminate()).unwrap();
        let mut sigint = signal(SignalKind::interrupt()).unwrap();

        tokio::select! {
            _ = sigterm.recv() => log::info!("Received SIGTERM"),
            _ = sigint.recv() => log::info!("Received SIGINT"),
        }

        shutdown_for_signal.initiate_shutdown();
    });

    let app = Router::new()
        .route("/health", get(health_check))
        .route("/api/process", get(handle_request))
        .layer(Extension(shutdown.clone()));

    let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await.unwrap();

    log::info!("Server listening on 0.0.0.0:8080");

    axum::serve(listener, app)
        .with_graceful_shutdown(async move {
            // 等待 shutdown 信号
            let mut rx = shutdown.subscribe();
            let _ = rx.recv().await;

            log::info!("Waiting for in-flight requests to complete...");
            shutdown.wait_for_completion().await;
            log::info!("All requests completed, shutting down");
        })
        .await
        .unwrap();
}

六、常见陷阱与维修案例

6.1 陷阱一：在 signal handler 中执行非 async-signal-safe 操作

// 绝对禁止：在 signal handler 中执行内存分配或锁操作
extern "C" fn unsafe_handler(sig: libc::c_int) {
    // 危险！malloc 不是 async-signal-safe 的
    let msg = format!("Received signal {}", sig);  // 触发 malloc
    println!("{}", msg);  // 获取 stdout 锁
    std::process::exit(0);  // 非异步安全
}

Tokio 的 signal() 实现将 handler 限制在安全操作范围内（向 eventfd/pipe 写入），避免了这个问题。但如果你手动注册 sigaction，必须严格遵守 POSIX 定义的 async-signal-safe 函数列表。

6.2 陷阱二：Nested signal handler 导致死锁

如下场景可能导致死锁：

场景：
1. Main thread: 持有某个锁，接收 SIGINT，开始 handler
2. Signal handler: 试图获取同一锁
3. 结果: 死锁

解决方案：signal handler 只做极少且确定安全的操作。

6.3 陷阱三：Tokio runtime drop 时任务中的 .await 被中断

// 注意：runtime drop 会取消所有任务
async fn critical_cleanup() {
    // 这段代码可能在 runtime drop 时被取消
    if let Err(e) = database.close().await {
        // 永远不会执行到这里，因为 drop 取消了任务
    }
}

上面的代码是危险的：database.close() 是一个异步清理操作，而 Tokio runtime drop 会取消所有正在运行的任务。如果 close 操作在执行到一半时被中断，可能导致连接泄漏或数据不一致。

正确的做法：在 runtime drop 之前提供显式的 shutdown 阶段：

async fn main_app() -> Result<(), Box<dyn std::error::Error>> {
    let db = Database::connect("postgres://...").await?;

    // 启动应用
    let app_handle = tokio::spawn(run_server(db.clone()));

    // 等待 shutdown 信号
    wait_for_signal().await;

    // 应用退出
    app_handle.await??;

    // 在 runtime drop 之前显式清理
    db.close().await?;

    Ok(())
}

#[tokio::main]
async fn main() {
    if let Err(e) = main_app().await {
        eprintln!("Error: {}", e);
    }
    // runtime 在这里 drop，但所有资源已清理完毕
}

6.4 陷阱四：忘记处理 SIGPIPE

在 Unix 上，向一个读者已关闭的管道写入数据会触发 SIGPIPE。默认动作是终止进程。在 Rust 网络编程中，特别是写 socket 时，SIGPIPE 可能导致程序莫名退出：

// 解决方案：在程序启动时忽略 SIGPIPE
unsafe {
    libc::signal(libc::SIGPIPE, libc::SIG_IGN);
}

// 或者在 Cargo.toml 中设置（如果是 Unix 环境）
// 这确保写操作返回 EPIPE 错误而非触发信号

Tokio 和标准库的 TCP stream 通常已经处理了这个问题，但如果你直接操作 raw fd，务必注意。

七、Linux 特有机制：signalfd 与 pidfd

7.1 signalfd 替代传统 signal handler

Linux 特有的 signalfd 允许将信号转换为文件描述符上的读取事件，这天然适合与 epoll/io_uring 集成的异步运行时：

use libc::{sigset_t, sigemptyset, sigaddset, sigprocmask, SIG_BLOCK, signalfd, signalfd_siginfo};
use std::os::unix::io::RawFd;

struct SignalFd {
    fd: RawFd,
    signals: Vec<c_int>,
}

impl SignalFd {
    fn new(signals: &[c_int]) -> nix::Result<Self> {
        unsafe {
            let mut sigset: sigset_t = std::mem::zeroed();
            sigemptyset(&mut sigset);
            for &sig in signals {
                sigaddset(&mut sigset, sig);
            }
            // 阻塞这些信号使其不通过传统方式投递
            if sigprocmask(SIG_BLOCK, &sigset, std::ptr::null_mut()) != 0 {
                return Err(nix::errno::Errno::last());
            }

            let fd = signalfd(-1, &sigset, 0);
            if fd < 0 {
                return Err(nix::errno::Errno::last());
            }

            Ok(Self { fd, signals: signals.to_vec() })
        }
    }

    fn as_raw_fd(&self) -> RawFd {
        self.fd
    }

    fn read_signal(&self) -> Option<signalfd_siginfo> {
        unsafe {
            let mut info: signalfd_siginfo = std::mem::zeroed();
            let ret = libc::read(
                self.fd, 
                &mut info as *mut _ as *mut libc::c_void,
                std::mem::size_of::<signalfd_siginfo>()
            );
            if ret == std::mem::size_of::<signalfd_siginfo>() as isize {
                Some(info)
            } else {
                None
            }
        }
    }
}

signalfd 的优势是信号变成了 I/O 事件，无需 self-pipe trick，与 io_uring 配合使用时更加高效。Tokio 从 1.28+ 开始在 Linux 上使用 signalfd 优化信号监听的实现。

7.2 pidfd 用于监控进程退出

Linux 5.3+ 引入了 pidfd 机制，允许通过文件描述符监控进程退出，无需传统的 waitpid 系统调用：

use libc::{pidfd_open, P_PIDFD};

// 可用于监控子进程退出，与 io_uring/epoll 集成
fn monitor_child(pid: i32) -> nix::Result<RawFd> {
    unsafe {
        let fd = pidfd_open(pid, 0);
        if fd < 0 {
            return Err(nix::errno::Errno::last());
        }
        Ok(fd)
    }
}

这对于实现进程管理器/监督者（类似 tini、s6、supervisord）非常有用，可以在同一事件循环中处理信号、超时和子进程退出。

八、跨运行时兼容性实践

不同 async 运行时的信号处理策略有所不同：

运行时	信号处理方式	推荐做法
Tokio	tokio::signal 模块	使用官方 signal API
async-std	底层用 signalfd	直接读取 signalfd
smol	内部用 epoll+signalfd	通过 async-io 集成
glommio	io_uring + signalfd	通过 nrfd 事件

统一的抽象：

/// 运行时不感知的 Shutdown Signal
pub struct ShutdownSignal {
    rx: tokio::sync::watch::Receiver<bool>,
}

impl ShutdownSignal {
    /// 启动平台特定的信号监听
    pub fn new() -> Result<Self, SignalError> {
        let (tx, rx) = tokio::sync::watch::channel(false);

        #[cfg(unix)]
        {
            let mut sigterm = signal(SignalKind::terminate())?;
            let mut sigint = signal(SignalKind::interrupt())?;

            tokio::spawn(async move {
                tokio::select! {
                    _ = sigterm.recv() => {},
                    _ = sigint.recv() => {},
                }
                let _ = tx.send(true);
            });
        }

        #[cfg(windows)]
        {
            // Windows 使用 ctrl_c / ctrl_break
            tokio::spawn(async move {
                let _ = tokio::signal::ctrl_c().await;
                let _ = tx.send(true);
            });
        }

        Ok(Self { rx })
    }

    pub async fn waited(&mut self) {
        let _ = self.rx.changed().await;
    }
}

九、总结与实践建议

Rust async 系统的信号处理是一个多维度的问题，涉及 Unix 语义理解、运行时调度、channel 通信和跨线程协调。以下是从工程实践中总结的关键原则：

1. 不要自己写 sigaction：使用 Tokio 的 signal() 或 signalfd-based 方案
2. 用 channel 传递 shutdown 通知：broadcast 用于多消费者，watch 用于一次性事件
3. Graceful shutdown 需要超时保护：永远不要假设所有任务都会及时退出
4. runtime drop 前完成关键清理：连接关闭、文件同步、状态持久化等
5. 关注 SIGPIPE：网络编程中不要忽略它
6. 使用 signalfd 优化性能：当与 io_uring 配合时显著减少 syscall 开销

最后，一个常被忽视的点：在容器化环境（Kubernetes）中，SIGTERM 从 Controller 发来到进程实际收到之间可能有数秒的延迟。结合 terminationGracePeriodSeconds 配置和应用的 shutdown 超时设定，确保在 K8s pod 终止时不会因处理不当而丢失请求。

---

技术栈版本：Rust 1.78+, Tokio 1.37+, axum 0.7+, Linux 5.15+

本文首发于 ybb.press