使用Rayon多线程访问共享数据及终止开关的Rust代码报错解决

问题 #

我用Rayon的par_iter()并行运行多个耗时任务，这些任务需要访问同一组checked usizes（要不时添加和检查），而且要在第一个线程完成时终止所有线程，所以设置了kill_switch，当它设为true时强制迭代退出。

当前代码：

let mut checked: HashSet<usize> = HashSet::new();
let mut kill_switch: bool = false;

permutations.par_iter().for_each(|order| {
    let board = Board::new(board_map.clone(), order.clone());
    let mut bubbles: Vec<(i8, i8)> = Vec::new();
    if let Some(bubbles) = board.solve(&mut bubbles, &mut checked, &kill_switch) {
        kill_switch = true;
        bubbles.into_iter().for_each(|bubble| {
            dbg!(bubble);
        });
    }
})

出现以下错误：

• cannot borrow checked as mutable, as it is a captured variable in a Fn closure cannot borrow as mutable [E0596]
• cannot assign to kill_switch, as it is a captured variable in a Fn closure cannot assign [E0594]

解决方案 #

问题核心是Rayon的for_each要求闭包为Fn（不可变捕获），且普通变量在多线程下直接操作会存在数据竞争，必须用线程安全的同步原语替换。

1. 替换线程不安全变量 #

• 用Arc<Mutex<HashSet<usize>>>包装checked：Arc实现多线程共享所有权，Mutex保证同一时间只有一个线程能修改集合。
• 用Arc<AtomicBool>实现kill_switch：AtomicBool支持原子操作，可线程安全地修改和读取开关状态。

2. 调整迭代逻辑 #

在每个任务开始前检查AtomicBool状态，触发则直接退出；找到结果后原子性设置开关为true，让其他线程快速终止。

修改后的代码 #

use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicBool, Ordering};
use rayon::prelude::*;

// 创建线程安全的共享集合与终止开关
let checked = Arc::new(Mutex::new(HashSet::<usize>::new()));
let kill_switch = Arc::new(AtomicBool::new(false));

permutations.par_iter().for_each(|order| {
    // 提前检查终止开关，避免无效计算
    if kill_switch.load(Ordering::Relaxed) {
        return;
    }

    let board = Board::new(board_map.clone(), order.clone());
    let mut bubbles: Vec<(i8, i8)> = Vec::new();

    // 锁定集合，保证线程安全的读写
    let mut checked_lock = checked.lock().unwrap();
    if let Some(bubbles) = board.solve(&mut bubbles, &mut checked_lock, &kill_switch) {
        // 原子性设置终止开关，通知所有线程退出
        kill_switch.store(true, Ordering::Relaxed);
        bubbles.into_iter().for_each(|bubble| {
            dbg!(bubble);
        });
    }
})

补充说明 #

• Ordering::Relaxed是最轻量的内存顺序，仅用于终止开关这类无严格内存依赖的场景；若有复杂同步需求，可替换为Ordering::SeqCst。
• Mutex::lock()返回Result，示例用unwrap()简化处理，实际项目可根据情况添加锁中毒等错误处理逻辑。

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