Rust应用死锁求助：并发线程终止时触发死锁问题

问题分析 #

你的代码出现死锁的核心原因是：

1. 子线程启动时会立即获取NetworkReceiverThread的Mutex锁，并在run方法的整个循环过程中一直持有该锁。
2. 主线程调用stop()方法时，需要获取同一个Mutex锁来设置终止标志，但子线程始终没有释放锁。
3. 子线程循环等待终止标志被设置，而主线程无法获取锁来设置标志，形成循环等待，导致死锁。

修复方案 #

我们需要重构代码，让终止标志可以不通过Mutex访问，同时确保Mutex锁不会被长时间持有：

1. 将终止标志从NetworkReceiverThread中移出，使用Arc<AtomicBool>在主线程和子线程间共享，原子类型无需锁即可安全访问。
2. 修改线程逻辑，仅在需要执行实际工作时才获取Mutex锁，执行完成后立即释放，避免长时间持有锁。
3. 调整StartThread trait的run方法，让它返回一个布尔值表示是否继续运行，增加灵活性。

修正后的代码 #

thread.rs

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

pub struct RunningThreadInterface<T> {
    instance: Arc<Mutex<T>>,
    thread_join_handle: thread::JoinHandle<()>,
    terminate_flag: Arc<AtomicBool>,
}

pub trait StartThread<T> {
    /// 执行一次线程工作，返回true表示继续运行
    fn run(&mut self) -> bool;
    
    /// 启动线程并返回控制句柄
    fn start(self, thread_ID: String) -> RunningThreadInterface<T>;
}

pub trait TerminateThread {
    fn stop(&mut self);
    fn wait(self);
}

NetworkReceiver.rs

use std::{thread, time};
use std::sync::{Arc, Mutex};

use crate::thread::*;

#[derive(Default)]
pub struct NetworkReceiverThread {
    thread_ID: String,
}

impl NetworkReceiverThread {
    pub fn new() -> NetworkReceiverThread {
        NetworkReceiverThread {
            thread_ID: String::from(""),
        }
    }
}

impl StartThread<NetworkReceiverThread> for NetworkReceiverThread {
    fn run(&mut self) -> bool {
        // 模拟网络接收工作
        println!("receiver thread");
        true // 继续运行
    }

    fn start(mut self, thread_ID: String) -> RunningThreadInterface<NetworkReceiverThread> {
        self.thread_ID = thread_ID.clone();

        let terminate_flag = Arc::new(AtomicBool::new(false));
        let terminate_flag_clone = Arc::clone(&terminate_flag);

        let instance = Arc::new(Mutex::new(self));
        let instance_clone = Arc::clone(&instance);

        let thread_handle = thread::Builder::new()
            .name(thread_ID)
            .spawn(move || {
                while !terminate_flag_clone.load(std::sync::atomic::Ordering::SeqCst) {
                    // 仅在执行工作时获取锁
                    let mut inner = instance_clone.lock().unwrap();
                    let should_continue = inner.run();
                    drop(inner); // 显式释放锁

                    if !should_continue {
                        terminate_flag_clone.store(true, std::sync::atomic::Ordering::SeqCst);
                        break;
                    }

                    thread::sleep(time::Duration::from_secs(1));
                }
            })
            .unwrap();

        RunningThreadInterface {
            instance,
            thread_join_handle: thread_handle,
            terminate_flag,
        }
    }
}

impl TerminateThread for RunningThreadInterface<NetworkReceiverThread> {
    fn stop(&mut self) {
        self.terminate_flag.store(true, std::sync::atomic::Ordering::SeqCst);
    }

    fn wait(self) {
        self.thread_join_handle.join().unwrap();
    }
}

main.rs

mod thread;
mod NetworkReceiver;

use std::time;
use thread::*;
use NetworkReceiver::NetworkReceiverThread;

fn main() {
    let network_receiver = NetworkReceiverThread::new();

    let mut network_receiver: RunningThreadInterface<NetworkReceiverThread> = network_receiver.start(String::from("NT"));

    std::thread::sleep(time::Duration::from_secs(5));

    network_receiver.stop();
    network_receiver.wait();

    println!("线程已成功终止");
}

关键改进点 #

• 原子终止标志：使用Arc<AtomicBool>共享终止状态，无需锁即可安全修改和读取。
• 锁的细粒度控制：仅在执行实际工作时获取Mutex锁，避免长时间持有导致其他线程阻塞。
• 清晰的线程控制逻辑：stop()直接设置终止标志，子线程在每次循环开始时检查标志，确保及时响应终止指令。

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