面试题：Rust多线程获取修改操作中的死锁问题及解决

Question

在使用Rust进行多线程编程实现获取修改操作时，可能会遇到死锁问题。请描述死锁产生的原因，在这种多线程获取修改数据场景下死锁可能是怎样出现的，并阐述如何检测和避免死锁。给出可能导致死锁的代码示例以及修正后的代码。

Answer 1

死锁产生的原因

死锁是指两个或多个线程互相等待对方释放资源，从而导致所有线程都无法继续执行的情况。在多线程编程中，死锁通常由以下四个必要条件同时满足而产生：

1. 互斥条件：资源一次只能被一个线程使用。
2. 占有并等待条件：一个线程在持有至少一个资源的同时，又请求其他资源，且在等待获取其他资源的过程中，不会释放已持有的资源。
3. 不可剥夺条件：线程已获得的资源，在未使用完之前，不能被其他线程强行剥夺，只能由该线程自己释放。
4. 循环等待条件：存在一个线程集合，其中每个线程都在等待下一个线程持有的资源，形成一个循环等待链。

在多线程获取修改数据场景下死锁可能出现的方式

在多线程获取修改数据场景中，如果多个线程需要获取多个锁来访问和修改共享数据，并且获取锁的顺序不一致，就容易出现死锁。例如，线程A先获取锁1，然后尝试获取锁2；而线程B先获取锁2，然后尝试获取锁1。如果此时线程A持有锁1，线程B持有锁2，那么两个线程就会互相等待对方释放锁，从而导致死锁。

检测死锁

1. 手动分析：仔细审查代码逻辑，检查线程获取锁的顺序是否可能形成循环等待。这需要对代码有深入的理解，尤其是在复杂的多线程场景下，可能比较困难。
2. 工具检测：
   • Rust 分析工具：例如 thread - sanitizer（TSAN），它是一个用于检测 C/C++ 和 Rust 程序中数据竞争和死锁的工具。可以通过在编译时添加相应的标志启用它，如在 cargo 构建时使用 RUSTFLAGS='-Zsanitizer=thread' cargo build --release，运行程序时，如果有死锁，TSAN 会输出详细的死锁信息，包括涉及的线程和锁的获取顺序等。

避免死锁

1. 按顺序获取锁：确保所有线程都按照相同的顺序获取锁。例如，如果有锁1和锁2，所有线程都先获取锁1，再获取锁2。
2. 使用try_lock：使用 try_lock 方法尝试获取锁，如果获取失败，线程可以选择释放已持有的锁并稍后重试，而不是一直等待，从而避免死锁。
3. 减少锁的持有时间：尽量缩短线程持有锁的时间，尽快完成对共享资源的操作并释放锁，减少其他线程等待的时间，降低死锁的可能性。

可能导致死锁的代码示例

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

fn main() {
    let data1 = Arc::new(Mutex::new(0));
    let data2 = Arc::new(Mutex::new(0));

    let data1_clone = data1.clone();
    let data2_clone = data2.clone();

    let handle1 = thread::spawn(move || {
        let mut guard1 = data1_clone.lock().unwrap();
        thread::sleep(std::time::Duration::from_millis(100));
        let mut guard2 = data2_clone.lock().unwrap();
        *guard1 += 1;
        *guard2 += 1;
    });

    let handle2 = thread::spawn(move || {
        let mut guard2 = data2.lock().unwrap();
        thread::sleep(std::time::Duration::from_millis(100));
        let mut guard1 = data1.lock().unwrap();
        *guard2 += 1;
        *guard1 += 1;
    });

    handle1.join().unwrap();
    handle2.join().unwrap();
}

在这个示例中，handle1 线程先获取 data1 的锁，再获取 data2 的锁；handle2 线程先获取 data2 的锁，再获取 data1 的锁，这就可能导致死锁。

修正后的代码（按顺序获取锁）

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

fn main() {
    let data1 = Arc::new(Mutex::new(0));
    let data2 = Arc::new(Mutex::new(0));

    let data1_clone = data1.clone();
    let data2_clone = data2.clone();

    let handle1 = thread::spawn(move || {
        let mut guard1 = data1_clone.lock().unwrap();
        let mut guard2 = data2_clone.lock().unwrap();
        *guard1 += 1;
        *guard2 += 1;
    });

    let handle2 = thread::spawn(move || {
        let mut guard1 = data1.lock().unwrap();
        let mut guard2 = data2.lock().unwrap();
        *guard1 += 1;
        *guard2 += 1;
    });

    handle1.join().unwrap();
    handle2.join().unwrap();
}

在修正后的代码中，两个线程都先获取 data1 的锁，再获取 data2 的锁，避免了死锁的发生。

修正后的代码（使用try_lock）

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

fn main() {
    let data1 = Arc::new(Mutex::new(0));
    let data2 = Arc::new(Mutex::new(0));

    let data1_clone = data1.clone();
    let data2_clone = data2.clone();

    let handle1 = thread::spawn(move || {
        loop {
            match (data1_clone.try_lock(), data2_clone.try_lock()) {
                (Ok(mut guard1), Ok(mut guard2)) => {
                    *guard1 += 1;
                    *guard2 += 1;
                    break;
                }
                _ => {
                    // 释放已获取的锁（如果有）
                    // 这里简化处理，实际可能需要更复杂逻辑
                    thread::sleep(std::time::Duration::from_millis(100));
                }
            }
        }
    });

    let handle2 = thread::spawn(move || {
        loop {
            match (data1.try_lock(), data2.try_lock()) {
                (Ok(mut guard1), Ok(mut guard2)) => {
                    *guard1 += 1;
                    *guard2 += 1;
                    break;
                }
                _ => {
                    thread::sleep(std::time::Duration::from_millis(100));
                }
            }
        }
    });

    handle1.join().unwrap();
    handle2.join().unwrap();
}

在这段代码中，使用 try_lock 尝试获取锁，如果获取失败，线程等待一段时间后重试，避免了死锁。