基于线程池的并行DFS无性能提升问题排查求助

问题描述 #

我尝试用Rust的多线程实现串行与并行DFS，已经创建了线程池并将节点处理任务提交进去，但并行版本不仅比串行慢，执行时间还不随构造线程池时传入的线程数变化。想问问哪里出错了？另外我是不是意外拷贝了不该拷贝的内容？

use std::sync::{Arc, Mutex};
use num_cpus;
use threadpool::ThreadPool as BaseThreadPool;
use std::collections::HashSet; // 补全原代码缺失的导入

pub struct GraphMatrix {
    adjacency_matrix: Vec<Vec<bool>>,
}

pub fn process_node_dfs(node_index: usize, adjacency_matrix: Arc<Vec<Vec<bool>>>, pool: BaseThreadPool, result: Arc<Mutex<Vec<usize>>>, visited: Arc<Mutex<HashSet<usize>>>) {
    let edges = &adjacency_matrix[node_index];

    result.lock().unwrap().push(node_index);

    for neighbour_index in 0..edges.len() {
        let mut visitedGuard: std::sync::MutexGuard<HashSet<usize>> = visited.lock().unwrap();
        if visitedGuard.contains(&neighbour_index) {
            std::mem::drop(visitedGuard);
            continue;
        } 
        else {
            visitedGuard.insert(neighbour_index);
        }
        std::mem::drop(visitedGuard);

        let adjacency_matrix_clone: Arc<Vec<Vec<bool>>> = Arc::clone(&adjacency_matrix);
        let visited_clone = Arc::clone(&visited);
        let result_clone = Arc::clone(&result);
        pool.execute(move || process_node_dfs(neighbour_index, adjacency_matrix_clone, pool.clone(), result_clone, visited_clone));
    }
}
impl GraphMatrix {
    
    pub fn parallel_dfs(&self, start_node: usize) -> Vec<usize> {
        let mut result: Arc<Mutex<Vec<usize>>> = Arc::new(Mutex::new(Vec::new()));
        let mut visited: Arc<Mutex<HashSet<usize>>> = Arc::new(Mutex::new(HashSet::new()));
        let pool: BaseThreadPool = BaseThreadPool::new(6);
        
        visited.lock().unwrap().insert(start_node);
        
        let adjacency_matrix_clone: Arc<Vec<Vec<bool>>> = Arc::new(self.adjacency_matrix.clone()); // Terrible
        let pool_clone = Arc::clone(&pool);
        let visited_clone = Arc::clone(&visited);
        let result_clone = Arc::clone(&result);
        pool.execute({
            let pool = pool.clone(); 
            move || process_node_dfs(start_node, adjacency_matrix_clone, pool, result_clone, visited_clone)});
        pool.join();
    
        return result.lock().unwrap().clone();
    }
}

核心问题分析 #

1. 全局互斥锁导致完全串行化 #

visited和result都使用了全局Mutex，每次处理邻居节点都要加锁检查、插入或修改结果。这会导致所有线程都在争抢这两把锁，整个DFS流程退化为串行执行——线程数再多也没用，反而会因为锁竞争的上下文切换开销，比纯串行版本更慢。

2. 不必要的邻接矩阵深拷贝 #

在parallel_dfs中，Arc::new(self.adjacency_matrix.clone())是深拷贝整个邻接矩阵，这会带来巨大的内存和时间开销，尤其是当矩阵规模较大时。实际上你只需要共享矩阵的只读引用，应该用Arc包装原矩阵的指针，而不是克隆整个矩阵。

3. 线程池的错误传递与克隆开销 #

你在process_node_dfs中每次提交任务都调用pool.clone()，线程池的克隆操作会创建新的句柄，但多次克隆不仅没必要，还会带来额外开销。正确的做法是把线程池包装成Arc，让所有任务共享同一个线程池实例。

4. 锁持有时间过长且频率过高 #

在遍历邻居的循环中，你每次迭代都要获取、释放visited锁，频繁的锁操作会加剧竞争。应该一次性获取锁，批量处理所有未访问的邻居，再释放锁后提交任务。

优化方案 #

1. 减少锁竞争：改用细粒度同步或无锁结构 #

• 用DashMap（无锁哈希表）替代Mutex<HashSet>，避免全局锁的竞争；
• 对visited进行分片，比如按节点哈希值分配到不同的锁分片，降低锁的粒度。

2. 避免不必要的拷贝 #

调整GraphMatrix结构，将邻接矩阵直接存储为Arc，避免深拷贝：

pub struct GraphMatrix {
    adjacency_matrix: Arc<Vec<Vec<bool>>>,
}

// 在parallel_dfs中直接克隆Arc指针（仅拷贝地址，无矩阵内容拷贝）
let adjacency_matrix_clone = Arc::clone(&self.adjacency_matrix);

3. 用Arc共享线程池 #

将线程池包装成Arc，避免多次克隆：

// 修改parallel_dfs中的线程池创建
let pool = Arc::new(BaseThreadPool::new(6));

// 修改process_node_dfs的参数和任务提交逻辑
pub fn process_node_dfs(
    node_index: usize,
    adjacency_matrix: Arc<Vec<Vec<bool>>>,
    pool: Arc<BaseThreadPool>,
    result: Arc<Mutex<Vec<usize>>>,
    visited: Arc<Mutex<HashSet<usize>>>,
) {
    let edges = &adjacency_matrix[node_index];
    result.lock().unwrap().push(node_index);

    // 批量处理未访问邻居，减少锁持有时间
    let mut unvisited = Vec::new();
    {
        let mut visited_guard = visited.lock().unwrap();
        for neighbour_index in 0..edges.len() {
            if edges[neighbour_index] && !visited_guard.contains(&neighbour_index) {
                visited_guard.insert(neighbour_index);
                unvisited.push(neighbour_index);
            }
        }
    }

    // 提交任务
    for neighbour in unvisited {
        pool.execute(move || {
            process_node_dfs(
                neighbour,
                Arc::clone(&adjacency_matrix),
                Arc::clone(&pool),
                Arc::clone(&result),
                Arc::clone(&visited),
            )
        });
    }
}

4. 优化结果收集的锁开销 #

如果不需要严格的DFS顺序，可以考虑每个线程维护自己的结果列表，最后合并，避免全局Mutex<Vec<usize>>的竞争：

// 示例思路：每个任务返回自己的结果片段，用channel收集后合并
// 需要调整process_node_dfs的返回值，引入channel机制

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