DPOR

主线位置：主线 1 的决策点。每个 loom 操作之前的 branch 都会调 Execution::schedule()。它做两件事：给「之前某个位置、换别的线程先跑可能得到不同结果」的地方登记回溯；选出现在该跑的线程。前者决定 Path 上会出现哪些 Pending，也就决定了 step 之后还有哪些轮要跑。

flowchart TB
  s["schedule()"] --> a["① 对每个有 operation 的线程 T：<br/>取该对象上与之相关的上一次访问 A"]
  a --> b{"A.dpor_vv ≤ T.dpor_vv ?"}
  b -->|是：A 在 T 之前，不必换序| c
  b -->|否| bt["path.backtrack(A.path_id, T)：<br/>在 A 发生的那个 Schedule 上把 T 标成 Pending"]
  bt --> c["② 选默认线程：当前线程能跑就是它；<br/>否则选 yield_count 最小的可运行线程"]
  c --> d["③ path.branch_thread：重放就按记录，<br/>走到头就以默认线程新建 Schedule"]
  d --> e{"④ 有线程可跑？"}
  e -->|否且未全部结束| dl["panic：deadlock; threads = …"]
  e -->|是| f["⑤ 被选线程的 dpor_vv 并入 A 的时钟、自增；<br/>记为该对象的 last access（位置 = 本分支）"]
  f --> g["⑥ 让出过的线程恢复可运行；返回是否切换"]

DPOR（dynamic partial order reduction，动态偏序归约）的出发点是：两个线程的操作如果互不相关（比如读同一个变量，或者访问不同的变量），先后交换不会改变结果，就不必两种顺序都跑。loom 的 crate 文档把它叫作「equivalent execution elimination」，举的例子正是「两个线程读同一个原子变量，loom 不会为它们的先后再跑一轮」（lib.rs:322-330）。下面按 execution.rs:135-256 的顺序读它的实现。

两套向量时钟

每个线程身上有三套 VersionVec（rt/thread.rs:20-27）：

causality 和 released 在 03 内存模型 讲。本页只关心 dpor_vv，它和内存序无关：一次 Relaxed 的 load 读了另一个线程的 store，在 causality 上不建立任何关系，在 dpor_vv 上却同样会把那次 store 并进来。DPOR 要回答的是「这两个操作的执行顺序能不能交换」，不是「这两个操作之间有没有同步」。

每个对象记着自己的「上一次访问」，类型是 Access：

pub(crate) struct Access {
    path_id: usize,
    dpor_vv: VersionVec,
}

impl Access {
    pub(crate) fn happens_before(&self, version: &VersionVec) -> bool {
        self.dpor_vv <= *version
    }
}

path_id 是那次访问被调度时所在的 Schedule 在路径里的下标，dpor_vv 是那个线程当时的 DPOR 时钟。<= 是向量时钟的偏序（VersionVec 的 PartialOrd，不可比时 <= 为假）。

什么算「相关」

last_dependent_access(operation) 按对象和动作取「与这次操作相关的上一次访问」（object.rs:211-225 分派到各类型）：

记录一侧对应地分开：原子每次访问都更新 last_access，只有 store / RMW 再更新 last_non_load_access（atomic.rs:836-848）。所以两个 load 之间永远不会互相触发回溯，load 和它之前最近的那次 store 才会。Arc 的注释写得直白：「Clones are only dependent with inspections. Drops are dependent between each other.」两个线程各 clone 一次，先后对结果没有影响。

UnsafeCell 不在表里：单元格访问不调 branch，不产生调度分支，也不参与 DPOR。它的竞争靠 causality 比较当场判定（UnsafeCell 与原子的非同步访问）。

① 登记回溯点

execution.rs:143-162：

for (th_id, th) in self.threads.iter() {
    let operation = match th.operation {
        Some(operation) => operation,
        None => continue,
    };

    if let Some(access) = self.objects.last_dependent_access(operation) {
        if access.happens_before(&th.dpor_vv) {
            // The previous access happened before this access, thus
            // there is no race.
            continue;
        }

        // Get the point to backtrack to
        let point = access.path_id();

        // Track backtracking point
        self.path.backtrack(point, th_id);
    }
}

对每一个线程（包括当前线程和暂时没被选中、操作还挂在身上的线程），看它下一步要做的操作：同一对象上相关的上一次访问 A，如果 A 不在它的 DPOR 历史里（A.dpor_vv 不 ≤ 它的 dpor_vv），说明 A 是别的线程做的、两者之间没有先后约束。那么在 A 被调度的那个位置，本可以先让这个线程跑，于是对那个位置调 Path::backtrack(point, th_id)：

pub(super) fn backtrack(&mut self, mut point: usize, thread_id: thread::Id) {
    let schedule = loop {
        if let Some(schedule_ref) =
            object::Ref::from_usize(point).downcast::<Schedule>(&self.branches)
        {
            let schedule = schedule_ref.get_mut(&mut self.branches);

            if schedule.exploring {
                schedule.backtrack(thread_id, self.preemption_bound);
                break schedule;
            }
        }

        if point == 0 {
            return;
        }

        point -= 1;
    };

    // …有抢占上限时额外补一个回溯点，见下一章…
}

从 point 往前找第一个 exploring 的 Schedule（point 本身一般就是 Schedule；被 stop_exploring 关掉的会被跳过），调 Schedule::backtrack：

fn backtrack(&mut self, thread_id: thread::Id, preemption_bound: Option<u8>) {
    assert!(self.exploring);

    if let Some(bound) = preemption_bound {
        // …已达上限就 return，见下一章…
    }

    let thread_id = thread_id.as_usize();

    if thread_id >= self.threads.len() {
        return;
    }

    if self.threads[thread_id].is_enabled() {
        self.threads[thread_id].explore();
    } else {
        for th in &mut self.threads {
            th.explore();
        }
    }
}

explore() 只把 Skip 改成 Pending，对 Active、Visited、Yield、Pending 不起作用（path.rs:525-530）。分两种情况：

• 目标线程在那个位置能跑（不是 Disabled）：只把它标成 Pending。
• 目标线程在那个位置不能跑（例如那时还没被 spawn，或在等锁）：把那个位置上所有 Skip 的线程都标成 Pending，也就是「那里换谁先跑都试一下」。

Pending 不会立刻生效，要等这一轮结束、step 从末尾往前回溯到这个位置时，才把它变成下一轮的 Active（Path）。

② 选默认线程：尽量不抢占

execution.rs:164-185：

// It's important to avoid pre-emption as much as possible
let mut initial = Some(self.threads.active_id());

// If the thread is not runnable, then we can pick any arbitrary other
// runnable thread.
if !self.threads.active().is_runnable() {
    initial = None;

    for (i, th) in self.threads.iter() {
        if !th.is_runnable() {
            continue;
        }

        if let Some(ref mut init) = initial {
            if th.yield_count < self.threads[*init].yield_count {
                *init = i;
            }
        } else {
            initial = Some(i)
        }
    }
}

当前线程还能跑，默认就让它接着跑；只有它阻塞、结束或 yield_now 了，才换成「让出次数最少」的可运行线程（同样少时取编号小的）。所以新建的 Schedule 里只有一个 Active、其余都是 Skip，第一轮总是「每个线程尽量一口气跑到阻塞或结束」。其他顺序全部来自 ① 登记的 Pending。

③ 记录或重放这一步

execution.rs:187-209：用默认线程构造 seed，每个线程映射成 Active（默认线程）/ Yield / Disabled（不可运行）/ Skip，交给 path.branch_thread。路径没走到头时 seed 被忽略、按记录返回；走到头时以它新建分支。返回的线程设为 active。path_id = self.path.pos() 在调用前取，就是这个 Schedule 的下标。

④ 死锁

execution.rs:211-226：

// There is no active thread. Unless all threads have terminated, the
// test has deadlocked.
if !self.threads.is_active() {
    let terminal = self.threads.iter().all(|(_, th)| th.is_terminated());

    assert!(
        terminal,
        "deadlock; threads = {:?}",
        self.threads
            .iter()
            .map(|(i, th)| { (i, th.state) })
            .collect::<Vec<_>>()
    );

    return true;
}

没有任何线程可选、又不是全部结束，就是死锁：每个剩下的线程都在等一个不会来的唤醒。loom 的死锁检测就这一处，它不分析等待图，只看「此刻没人能跑」。报错的格式、LOOM_LOCATION 下的阻塞位置，以及死锁 panic 之后展开时常见的 abort，在 死锁、分支上限与位置信息。

⑤ 更新 dpor_vv 与 last access

execution.rs:228-241：

if let Some(operation) = self.threads.active().operation {
    let threads = &mut self.threads;
    let th_id = threads.active_id();

    if let Some(access) = self.objects.last_dependent_access(operation) {
        threads.active_mut().dpor_vv.join(access.version());
    }

    threads.active_mut().dpor_vv[th_id] += 1;

    self.objects
        .set_last_access(operation, path_id, &threads.active().dpor_vv);
}

被选中的线程马上就要执行它挂着的操作，于是：它的 dpor_vv 并入相关的上一次访问（从此「在那次访问之后」），本线程分量加一，然后把自己记为这个对象最新的访问，位置是本分支 path_id。以后别的线程对这个对象做相关操作时，比较的就是这条记录。

注意记录的位置是「这个线程被选中去做这个操作的那个 Schedule」，而不是它登记 operation 的那个。一个线程可能在位置 2 登记了 load、被切走，到位置 7 才被选中执行，它的 load 记在 7。之后回溯到 7，意思是「在这次 load 真正执行之前换别人」。

⑥ 收尾

execution.rs:243-255：所有处于 Yield 状态、且这次没被选中的线程恢复成可运行；切换了线程就打一条 info 日志「~~~~~~~~ THREAD n ~~~~~~~~」；返回「当前线程是否换了」，调用方据此决定要不要 switch()。

在消息传递模型里看回溯

一次 model 的一生 的模型：1 号线程写 UnsafeCell 再 store(true, Release) 一个 flag；0 号线程 load(Acquire) flag、为真再读单元格、最后 join。用一份只加了打印的 loom 副本（exp-loom/pathdump/，在 schedule 开头和 path.backtrack 前各打一行，算法一字未改）看第 1 轮：

SCHED pos=0 cur=0 op=Some((Ref<()>(1), Arc(RefInc)))
SCHED pos=1 cur=0 op=Some((Ref<()>(3), Arc(RefInc)))
SCHED pos=2 cur=0 op=Some((Ref<()>(2), Atomic(Load)))
LOAD pos=3 store_index=0 value=0
SCHED pos=4 cur=0 op=Some((Ref<()>(4), Opaque))
SCHED pos=5 cur=1 op=Some((Ref<()>(2), Atomic(Store)))
BACKTRACK point=2 thread=1
…
SCHED pos=9 cur=1 op=None
BACKTRACK point=8 thread=0

• 位置 2：0 号线程要 load flag（对象 2），1 号线程刚 spawn、能跑但被标 Skip；默认让 0 号接着跑。load 只有一个候选（初值 false），位置 3 是一个单选项的 Load。
• 位置 4：0 号线程 join，等 1 号线程的 Notify（对象 4），阻塞，换 1 号。
• 位置 5：1 号线程要 store flag。flag 上一次访问是 0 号的 load，记在位置 2，它的 dpor_vv 不在 1 号线程的历史里，于是 BACKTRACK point=2 thread=1：在位置 2 把 1 号线程从 Skip 改成 Pending。意思是「在 0 号读 flag 之前，先让 1 号跑」。
• 位置 9：1 号线程结束，0 号线程挂着的 join 等待（Opaque）和 1 号在位置 8 的 notify 相关，也登记一次；但 0 号线程在位置 8 是 Disabled（阻塞在 join 上），按规则把位置 8 上所有 Skip 改成 Pending，而那里没有 Skip，什么也没改。

第 1 轮结束，step 从末尾往前找到位置 2 的 Pending，第 2 轮在那里先跑 1 号线程。6 轮的完整过程在 一次 model 的一生。

实验：哪些操作对会产生回溯

断言：两个线程只做互不相关的操作（读同一个原子、写不同的原子）时，交错数和「只有 spawn 与 join」的空模型一样；做相关的操作（写同一个原子、抢同一把锁）时交错数增加。

方法（exp-loom/runtime/tests/dep.rs）：主线程 spawn 两个线程、依次 join。共享的原子和锁用 Box::leak 得到 &'static，不经过 loom Arc，免得 Arc 的 clone / drop 混进来。每种组合跑一次 loom::model，不设抢占上限，数交错数。

结果（交错数确定，与负载无关）：

空模型就有 3 个交错：主线程的两次 join 都在等子线程结束时的 Notify，这些 Notify 操作之间也有依赖。两次 load、两组互不相干的 fetch_add，交错数和空模型完全一样：这些操作对一个回溯点也没有登记。

用 pathdump 打印同一对模型的回溯记录，对比更直接：两个线程各 load 一次时，第 1 轮的 BACKTRACK 只有 point=3 thread=0 和 point=8 thread=0 两条，都来自主线程 join 的 Notify；各 store 一次时，第 1 轮多了一条 BACKTRACK point=1 thread=2：2 号线程要 store 时，发现 1 号线程在位置 1 store 过同一个原子，于是在位置 1 登记「先让 2 号跑」。

「一个 load、一个 store」比「两个 store」还多 3 个，是因为 load 排在 store 之后时会有两个候选值（初值和新值），多出来的是 Load 分支，不是调度分支。

反向验证：表里本身就是对照。同一个模型骨架，只把两个线程的操作从「不相关」换成「相关」，交错数就从 3 变成 9、12、30，说明计数方法能分辨依赖关系。

不设上限时的规模

不设 LOOM_MAX_PREEMPTIONS 时，回溯点只受上面这些规则约束，交错数随线程数和每线程操作数增长很快。下一章 E2 的表格里有一列「不设上限」的交错数（例如 3 个线程各做 1 次 fetch_add 加 1 次加锁：1 393 998 个），以及设上限 0、1、2、3 时的数字和耗时。是否设上限、设多少，见 抢占上限与剪枝。

小结

• schedule() = ① 对每个线程挂着的操作，取同一对象上相关的上一次访问，没有 DPOR 先后关系就在那次访问的位置把这个线程标成 Pending；② 默认让当前线程接着跑；③ 记录或重放；④ 无人可跑且未全部结束 → 「deadlock; threads = …」；⑤ 更新被选线程的 dpor_vv 与对象的 last access；⑥ 恢复让出的线程。
• dpor_vv 与 causality 是两套时钟：前者只记执行先后，后者记内存模型的同步。
• 依赖规则：原子 load 只和上一次 store / RMW 相关；Arc clone 只和 inspect 相关、drop 和 drop 相关；channel 的 send 和 send、recv 和 recv 相关；锁类对象一律相关；UnsafeCell 不参与。
• Schedule::backtrack：目标线程能跑就标它 Pending，不能跑就把所有 Skip 标 Pending。
• 实测：两线程读同一原子、写不同原子都和空模型一样是 3 个交错；写同一原子 9 个，fetch_add 各两次 30 个。