如何选择性关闭其他线程持有的Unix域套接字（监听/客户端）

我使用uds crate实现Unix域套接字本地通信，当前代码中监听端和客户端线程阻塞在recv操作上，无法从外部触发套接字关闭和线程终止，需要解决选择性关闭特定套接字的问题。

解决方案思路 #

要实现外部触发线程终止，核心是让阻塞的IO操作能被中断，同时提供线程间安全的终止信号。这里采用原子布尔标志+非阻塞套接字的方案：

• 用Arc<AtomicBool>作为跨线程的终止信号，主线程可通过设置该标志通知子线程退出
• 将套接字设置为非阻塞模式，避免accept/recv无限阻塞，线程可定期检查终止标志
• 监听端为每个新连接启动独立线程处理，保证主监听线程能及时响应终止信号

修改后的代码 #

uds_listener.rs #

use std::thread;
use std::sync::{Arc, atomic::{AtomicBool, Ordering}};
use std::io;
use uds::UnixSeqpacketListener;

pub struct UdsListener {
    listener: UnixSeqpacketListener,
    stop_flag: Arc<AtomicBool>,
}

impl UdsListener {
    pub fn new(sock_name: &str) -> (Self, Arc<AtomicBool>) {
        let stop_flag = Arc::new(AtomicBool::new(false));
        let listener = Self::bind(sock_name);
        // 设置监听套接字为非阻塞模式
        listener.set_nonblocking(true).expect("Failed to set listener nonblocking");
        (Self { listener, stop_flag: stop_flag.clone() }, stop_flag)
    }

    fn bind(sock_name: &str) -> UnixSeqpacketListener {
        match UnixSeqpacketListener::bind(sock_name) {
            Ok(listener) => listener,
            Err(e) => panic!("uds listener bind failed for {sock_name}: {e:?}"),
        }
    }

    pub fn spawn_thread(mut self) -> thread::JoinHandle<()> {
        thread::spawn(move || {
            self.run();
        })
    }

    fn run(&mut self) {
        while !self.stop_flag.load(Ordering::Relaxed) {
            match self.listener.accept_unix_addr() {
                Ok((mut socket, _)) => {
                    // 设置连接套接字为非阻塞
                    socket.set_nonblocking(true).expect("Failed to set connection nonblocking");
                    let conn_stop_flag = self.stop_flag.clone();
                    // 启动独立线程处理连接，避免阻塞监听
                    thread::spawn(move || {
                        let mut buff = [0u8; 8192];
                        while !conn_stop_flag.load(Ordering::Relaxed) {
                            match socket.recv(&mut buff) {
                                Ok(length) => {
                                    if length == 0 {
                                        break;
                                    }
                                    // 这里添加数据处理逻辑
                                },
                                Err(e) => match e.kind() {
                                    io::ErrorKind::WouldBlock => {
                                        // 无数据时短暂休眠，避免忙等
                                        thread::sleep(std::time::Duration::from_millis(10));
                                    },
                                    _ => {
                                        eprintln!("uds listener read failed: {e:?}");
                                        break;
                                    }
                                }
                            }
                        }
                    });
                },
                Err(e) => match e.kind() {
                    io::ErrorKind::WouldBlock => {
                        thread::sleep(std::time::Duration::from_millis(10));
                    },
                    _ => {
                        eprintln!("uds listener accept failed: {e:?}");
                        break;
                    }
                }
            }
        }
        // 显式释放监听套接字，触发关闭
        drop(self.listener);
    }
}

uds_client.rs #

use std::{time, thread};
use std::sync::{Arc, atomic::{AtomicBool, Ordering}};
use std::io;
use uds::UnixSeqpacketConn;

pub struct UdsClient {
    socket: UnixSeqpacketConn,
    stop_flag: Arc<AtomicBool>,
}

impl UdsClient {
    pub fn new(sock_name: &str) -> (Self, Arc<AtomicBool>) {
        let stop_flag = Arc::new(AtomicBool::new(false));
        let socket = Self::connect(sock_name);
        // 设置客户端套接字为非阻塞
        socket.set_nonblocking(true).expect("Failed to set client nonblocking");
        (Self { socket, stop_flag: stop_flag.clone() }, stop_flag)
    }

    fn connect(sock_name: &str) -> UnixSeqpacketConn {
        loop {
            match UnixSeqpacketConn::connect(sock_name) {
                Ok(socket) => return socket,
                Err(_) => thread::sleep(time::Duration::from_secs(1)),
            }
        }
    }

    pub fn spawn_thread(mut self) -> thread::JoinHandle<()> {
        thread::spawn(move || {
            self.run();
        })
    }

    fn run(&mut self) {
        let mut buff = [0u8; 8192];
        while !self.stop_flag.load(Ordering::Relaxed) {
            match self.socket.recv(&mut buff) {
                Ok(length) => {
                    if length == 0 {
                        break;
                    }
                    // 这里添加数据处理逻辑
                },
                Err(e) => match e.kind() {
                    io::ErrorKind::WouldBlock => {
                        thread::sleep(time::Duration::from_millis(10));
                    },
                    _ => {
                        eprintln!("uds client read failed: {e:?}");
                        break;
                    }
                }
            }
        }
        // 显式释放客户端套接字
        drop(self.socket);
    }
}

main.rs #

mod uds_listener;
mod uds_client;

use std::thread::JoinHandle;
use std::sync::{Arc, atomic::AtomicBool};

use uds_listener::UdsListener;
use uds_client::UdsClient;
    
fn main() {
    let mut uds_listeners_thread_join_handle_list: Vec<JoinHandle<()>> = Vec::new();
    let mut listener_stop_flags: Vec<Arc<AtomicBool>> = Vec::new();
    
    for i in 0..256 {
        let sock_path = format!("/tmp/sock_{i}.sock");
        let (uds_listener, stop_flag) = UdsListener::new(&sock_path);
        let uds_listener_jh = uds_listener.spawn_thread();
        uds_listeners_thread_join_handle_list.push(uds_listener_jh);
        listener_stop_flags.push(stop_flag);
    }

    let mut uds_clients_thread_join_handle_list: Vec<JoinHandle<()>> = Vec::new();
    let mut client_stop_flags: Vec<Arc<AtomicBool>> = Vec::new();
    
    for i in 0..256 {
        let sock_path = format!("/tmp/sock_{i}.sock");
        let (uds_client, stop_flag) = UdsClient::new(&sock_path);
        let uds_client_jh = uds_client.spawn_thread();
        uds_clients_thread_join_handle_list.push(uds_client_jh);
        client_stop_flags.push(stop_flag);
    }

    // 示例：运行5秒后关闭前10个监听端和客户端
    println!("Running for 5 seconds...");
    std::thread::sleep(std::time::Duration::from_secs(5));

    println!("Stopping first 10 listeners and clients...");
    for flag in listener_stop_flags.iter().take(10) {
        flag.store(true, std::sync::atomic::Ordering::Relaxed);
    }
    for flag in client_stop_flags.iter().take(10) {
        flag.store(true, std::sync::atomic::Ordering::Relaxed);
    }

    // 等待线程退出
    println!("Waiting for listener threads to join");
    while let Some(ljh) = uds_listeners_thread_join_handle_list.pop() {
        let _ = ljh.join();
    }

    println!("Waiting for client threads to join");
    while let Some(cjh) = uds_clients_thread_join_handle_list.pop() {
        let _ = cjh.join();
    }

    // 清理套接字文件
    for i in 0..256 {
        let _ = std::fs::remove_file(format!("/tmp/sock_{i}.sock"));
    }
}

方案说明 #

1. 线程安全的终止信号：Arc<AtomicBool>提供了轻量且线程安全的终止标志，主线程通过store(true)通知子线程退出循环。
2. 非阻塞套接字：将套接字设置为非阻塞模式后，accept和recv不会无限阻塞，当无数据/连接时返回WouldBlock，线程可借此机会检查终止标志。
3. 独立连接线程：监听端为每个新连接启动独立线程，确保主监听线程不会被单个连接阻塞，能及时响应终止信号。
4. 选择性关闭：通过保存对应套接字的终止标志，可实现精准关闭任意监听端或客户端。

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