如何原子性访问分属同一/不同锁区域的网格多点？

安全阿岳

2026-8-9

多线程网格的原子性多点操作实现问题

我有一个支持多线程并发访问的大型网格，已将其划分为多个独立加锁的区域。现在需要实现原子性操作指定点集的功能——这些点可能属于同一区域，也可能分散在不同区域。

现有代码

pub struct RwGrid<T>{
    width : usize,
    height : usize,
    region_size : usize,
    regions : [RwLock<Vec<T>>; TOTAL_REGIONS]
}

impl<T: Copy + Colored> Grid<T> for RwGrid<T>{
    ...
    fn set_if<F>(&self, p : Point, f : F, value : T) -> bool where
        F : Fn(T) -> bool{
            let (region_index, index_in_region) = self.map_coordinates(p);
            let mut region = self.regions[region_index].write().unwrap();
            let pre_existing = region[index_in_region];
            if f(pre_existing){
                region[index_in_region] = value;
                true
            } else {false}
        }
    ...
}

其中map_coordinates是将笛卡尔坐标映射到对应区域索引及区域内点索引的辅助函数。

需求与问题

我的核心目标是实现set_if的变体，使其能原子性操作一组点（具体为某点的9个邻域点）。这类点集可能跨多个区域，且必须按特定顺序获取锁以避免死锁。

原子性要求至关重要：例如要保证「无相邻红点」的不变性，若线程非原子性检查邻域，可能导致两个相邻点被同时设为红色。

当前实现双点操作的set_if_2需根据点是否同区域分支处理，扩展到9点会产生大量冗余代码。我提出两种思路但存疑虑：

思路1：函数返回Vec<RwLockWriteGuard<T>>及点对应向量索引的结构；
思路2：使用无锁的unsafe Vec存储数据，搭配区域「伪锁」，分离锁获取与数据访问逻辑。

另外，网格的环绕特性（由fix函数实现）增加了死锁规避难度：按方位顺序锁可能形成死锁循环。

思路分析与最优方案

思路可行性判断

思路1：返回锁守卫集合

可行，但需注意关键细节：

必须严格按区域索引从小到大的固定顺序获取锁，完全不考虑点的方位或网格环绕情况——这是从根源避免死锁的核心，所有线程遵循同一锁顺序，不会出现循环等待。
建议定义绑定点、锁守卫、区域内索引的结构体，比如：
```
struct LockedPoint<'a, T> {
    guard: RwLockWriteGuard<'a, Vec<T>>,
    idx_in_region: usize,
}
```
再通过HashMap<Point, LockedPoint<'a, T>>或有序列表返回，方便后续操作。
缺点是锁持有时间会随操作延长，但针对9个邻域这种小规模点集，影响可忽略。

思路2：无锁Vec+伪锁

不推荐，风险极高：

Rust的Vec本身不具备线程安全性，手动用unsafe操作极易触发数据竞争，违背Rust安全模型，后续维护成本极高。
伪锁若实现不当（如缺少正确内存屏障），根本无法保证原子性，反而会引入隐蔽bug。

更优方案：封装批量操作逻辑

推荐把锁获取、批量检查、批量更新的逻辑完全封装在函数内部，避免调用方处理锁细节，同时消除冗余代码：

统一锁获取顺序：提取所有点对应的区域索引，去重后按升序获取写锁——与网格环绕无关，仅靠固定的区域索引排序就能避免死锁。

实现批量原子操作：

use std::collections::BTreeMap;
use std::sync::RwLockWriteGuard;

fn batch_set_if<F>(&self, points: &[Point], f: F, value: T) -> bool
where
    F: Fn(T) -> bool,
{
    // 1. 按区域分组存储点信息，BTreeMap自动按区域索引排序
    let mut region_points: BTreeMap<usize, Vec<(Point, usize)>> = BTreeMap::new();
    for &p in points {
        let (region_idx, idx_in_region) = self.map_coordinates(p);
        region_points.entry(region_idx).or_default().push((p, idx_in_region));
    }

    // 2. 按顺序获取所有需要的写锁
    let mut guards: Vec<(usize, RwLockWriteGuard<Vec<T>>)> = Vec::new();
    for &region_idx in region_points.keys() {
        guards.push((region_idx, self.regions[region_idx].write().unwrap()));
    }

    // 3. 先批量检查所有点是否满足条件（只读不写）
    let mut can_update = true;
    let mut update_targets: Vec<(usize, usize)> = Vec::new();
    for (region_idx, guard) in &guards {
        if let Some(points_in_region) = region_points.get(region_idx) {
            for &(_, idx) in points_in_region {
                let val = guard[idx];
                if !f(val) {
                    can_update = false;
                    break;
                }
                update_targets.push((*region_idx, idx));
            }
            if !can_update {
                break;
            }
        }
    }

    // 4. 所有条件满足时批量更新
    if can_update {
        for (region_idx, idx) in update_targets {
            let (_, guard) = guards.iter_mut().find(|(rid, _)| rid == &region_idx).unwrap();
            guard[idx] = value;
        }
        true
    } else {
        false
    }
}

适配邻域场景：只需把9个邻域点收集成Vec<Point>传入该函数即可，无需区分是否跨区域，逻辑完全统一。

关键注意事项

锁顺序优先：永远按区域索引升序获取锁，这是避免死锁的唯一可靠方式，与网格环绕无关。
先检查后更新：将检查与更新拆分两步，确保要么全更新、要么全不更新，严格保证原子性。
缩短锁持有时间：避免在持有锁时执行无关计算，函数内仅保留必要的锁操作逻辑。

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

如何原子性访问分属同一/不同锁区域的网格多点？

现有代码 #

需求与问题 #

思路分析与最优方案 #

思路可行性判断 #