如何在Rust测试中仅向当前线程发送SIGTERM信号？

问题代码 #

use nix::sys::signal::{self, SigSet};

#[tokio::test]
async fn test_shutdown() {
let server = TestServer::new().await;
let proxy = ProxyUnderTest::new(server.listen_port).await;

sleep(std::time::Duration::from_millis(50));

let current_thread_id = unsafe { libc::pthread_self() };
println!("current_thread_id :{:?}", current_thread_id);

// 向当前线程发送SIGTERM信号
nix::sys::signal::pthread_kill(current_thread_id, signal::Signal::SIGTERM).unwrap();

// 在主线程中阻塞SIGTERM信号
let mut sigset = SigSet::empty();
sigset.add(signal::SIGTERM);
signal::pthread_sigmask(signal::SigmaskHow::SIG_BLOCK, Some(&sigset), None).unwrap();

// 等待proxy关闭
let reason_opt = proxy.handle.await.unwrap();
assert_eq!(reason_opt, Some(ShutdownReason::GracefulShutdown));}

问题描述 #

在Rust测试中希望仅向proxy所在线程发送SIGTERM信号，但当前该信号会被cargo test创建的所有其他线程接收。已尝试使用pthread_kill发送信号，还尝试通过SigSet和pthread_sigmask在主线程中阻塞SIGTERM信号，但均未解决问题。

核心问题分析 #

1. 目标线程错误：当前代码是向测试主线程发送SIGTERM，而非proxy运行的线程，完全偏离测试目标。
2. 信号掩码时序错误：先发送信号再阻塞主线程信号，此时信号可能已经被其他线程接收。
3. 信号掩码继承问题：Tokio工作线程会继承测试主线程的信号掩码，若不提前设置，所有线程都能接收SIGTERM。

解决方案 #

1. 提前阻塞全局线程的SIGTERM #

在测试启动时，先让测试主线程阻塞SIGTERM，后续创建的Tokio工作线程会继承这个掩码，默认无法接收SIGTERM。

2. 让Proxy在单独线程运行并解除SIGTERM阻塞 #

将Proxy放在单独线程中运行，在该线程内解除SIGTERM的阻塞，确保只有这个线程能接收信号。

3. 正确获取Proxy线程ID并发送信号 #

通过同步机制（比如channel）获取Proxy所在线程的ID，再向该线程发送SIGTERM。

修正后的代码示例 #

use nix::sys::signal::{self, SigSet, Signal};
use std::sync::mpsc;
use tokio::task;

#[derive(Debug, PartialEq)]
enum ShutdownReason {
    GracefulShutdown,
}

struct Proxy;
impl Proxy {
    fn new(_listen_port: u16) -> Self {
        Self
    }

    fn run_until_shutdown(&self) -> Option<ShutdownReason> {
        // 模拟Proxy监听SIGTERM并优雅关闭的逻辑
        let mut sigset = SigSet::empty();
        sigset.add(Signal::SIGTERM);
        match signal::sigwait(&sigset) {
            Ok(Signal::SIGTERM) => Some(ShutdownReason::GracefulShutdown),
            _ => None,
        }
    }
}

struct TestServer;
impl TestServer {
    async fn new() -> Self {
        Self
    }

    fn listen_port(&self) -> u16 {
        8080
    }
}

struct ProxyUnderTest {
    handle: task::JoinHandle<Option<ShutdownReason>>,
    thread_id: libc::pthread_t,
}

impl ProxyUnderTest {
    async fn new(listen_port: u16) -> Self {
        let (sender, receiver) = mpsc::channel();

        let handle = task::spawn_blocking(move || {
            // 在Proxy线程中解除SIGTERM阻塞
            let mut sigset = SigSet::empty();
            sigset.add(Signal::SIGTERM);
            signal::pthread_sigmask(signal::SigmaskHow::SIG_UNBLOCK, Some(&sigset), None).unwrap();

            // 发送当前线程ID给测试主线程
            let thread_id = unsafe { libc::pthread_self() };
            sender.send(thread_id).unwrap();

            // 启动Proxy运行逻辑
            let proxy = Proxy::new(listen_port);
            proxy.run_until_shutdown()
        });

        // 等待获取Proxy线程ID
        let thread_id = receiver.recv().unwrap();

        Self { handle, thread_id }
    }
}

#[tokio::test]
async fn test_shutdown() {
    // 第一步：测试主线程先阻塞SIGTERM，让所有Tokio工作线程继承该掩码
    let mut sigset = SigSet::empty();
    sigset.add(Signal::SIGTERM);
    signal::pthread_sigmask(signal::SigmaskHow::SIG_BLOCK, Some(&sigset), None).unwrap();

    let server = TestServer::new().await;
    let proxy = ProxyUnderTest::new(server.listen_port()).await;

    tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    // 向Proxy所在线程发送SIGTERM
    nix::sys::signal::pthread_kill(proxy.thread_id, Signal::SIGTERM).unwrap();

    // 等待Proxy关闭并验证结果
    let reason_opt = proxy.handle.await.unwrap();
    assert_eq!(reason_opt, Some(ShutdownReason::GracefulShutdown));
}

关键说明 #

• 信号掩码继承：Tokio的spawn_blocking创建的线程会继承测试主线程的信号掩码，主线程提前阻塞SIGTERM后，该线程默认也会阻塞，我们在其中手动解除阻塞，确保只有它能接收信号。
• 线程ID传递：用mpsc::channel在Proxy线程和测试主线程之间传递线程ID，保证能准确获取目标线程的ID。
• 时序修正：先阻塞全局信号，再启动Proxy，最后发送信号，确保只有目标线程能接收SIGTERM。

内容的提问来源于stack exchange，提问作者Vinay Sawant