Rust多线程文件搜索出现死锁问题求助

核心问题本质 #

当线程数达上限且待检查目录为空时，线程不终止、主循环卡死，问题大概率出在线程计数同步逻辑漏洞或任务状态感知缺失，而非单纯的Mutex释放问题。

具体排查方向 #

1. 线程计数的增减时机错误 #

• 确认线程启动时是否立即递增计数，退出前最后一步递减计数，且两个操作都在Mutex保护下。如果线程在添加子目录后就提前减计数，会导致主程序误判空闲线程数，启动过多线程，最终线程数超限且无法正常回收。
• 绝对不能在处理文件对比这类耗时操作时持有计数锁，锁仅应在修改/读取计数的瞬间持有。

2. Mutex持有时间过长 #

• 检查线程内是否在执行文件哈希计算、相似度对比等耗时任务时，还持有directories_to_check或current_amount_threads的锁。这种情况会导致主程序无法获取锁检查队列或计数，陷入空循环。
• 正确做法：仅在读写待检查目录队列或线程计数时短时间持有锁，操作完成后立即释放，耗时任务完全在锁外执行。

3. 线程退出条件未覆盖所有场景 #

• 检查线程是否在当前目录文件处理完成且无后续子目录可添加时才退出。如果线程在添加完子目录后就挂起等待，会导致线程存活但无任务，占用计数名额，主程序误以为还有活跃线程。
• 排查线程内是否存在无限等待逻辑，比如错误依赖永远不会触发的信号。

4. 缺少条件变量配合 #

• 如果主程序是轮询式检查队列和计数，会导致效率低下且可能错过线程退出的信号。必须配合条件变量：
  • 线程完成任务递减计数时，发送信号通知主程序。
  • 有新目录加入队列时，同样发送信号唤醒主程序。
• 没有条件变量的话，主程序可能在目录队列为空但线程仍在运行时持续空转，线程完成后也无法及时感知计数变化。

5. 未处理线程panic #

• 线程如果发生未捕获的panic，会直接退出但不执行递减计数的逻辑，导致计数一直停留在最大值，主程序误以为还有线程在运行，无法退出循环。
• 解决：用std::panic::catch_unwind包裹线程核心逻辑，确保即使panic也能正确更新线程计数。

关键代码修正示例 #

线程工作函数 #

fn thread_worker(
    dir: PathBuf,
    dirs_mutex: Arc<Mutex<Vec<PathBuf>>>,
    count_mutex: Arc<Mutex<usize>>,
    cond_var: Arc<Condvar>,
) {
    // 启动时立即递增计数，完成后释放锁
    let mut count = count_mutex.lock().unwrap();
    *count += 1;
    drop(count);

    let mut sub_dirs = Vec::new();
    // 处理当前目录文件对比（耗时操作，无锁）
    process_file_similarity(&dir);
    // 收集子目录
    if let Ok(entries) = fs::read_dir(dir) {
        for entry in entries.flatten() {
            if entry.file_type().unwrap().is_dir() {
                sub_dirs.push(entry.path());
            }
        }
    }

    // 短时间持有锁添加子目录，然后通知主程序
    let mut dirs = dirs_mutex.lock().unwrap();
    dirs.extend(sub_dirs);
    drop(dirs);
    cond_var.notify_one();

    // 退出前递减计数，通知主程序
    let mut count = count_mutex.lock().unwrap();
    *count -= 1;
    drop(count);
    cond_var.notify_one();
}

主程序逻辑 #

fn search_compare() {
    let start_dir = PathBuf::from("/your/start/path");
    let dirs_mutex = Arc::new(Mutex::new(vec![start_dir]));
    let count_mutex = Arc::new(Mutex::new(0));
    let cond_var = Arc::new(Condvar::new());
    const MAX_THREADS: usize = 4;

    loop {
        let mut count = count_mutex.lock().unwrap();
        let mut dirs = dirs_mutex.lock().unwrap();

        // 退出条件：无活跃线程且无待处理目录
        if *count == 0 && dirs.is_empty() {
            break;
        }

        // 启动新线程：有目录且线程数未达上限
        while !dirs.is_empty() && *count < MAX_THREADS {
            let dir = dirs.pop().unwrap();
            let dirs_clone = Arc::clone(&dirs_mutex);
            let count_clone = Arc::clone(&count_mutex);
            let cond_clone = Arc::clone(&cond_var);

            std::thread::spawn(move || {
                // 捕获panic，确保计数更新
                let _ = std::panic::catch_unwind(|| {
                    thread_worker(dir, dirs_clone, count_clone, cond_clone);
                });
            });
        }

        // 释放锁后等待信号
        drop(dirs);
        count = cond_var.wait(count).unwrap();
    }
}

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