Arc、mpsc 与 future

主线位置：主线 1 里 Execution::schedule 用的「哪两个操作相关」（DPOR 的依赖关系）在 Arc 和 channel 上的具体定义；主线 4 里 channel 断开语义与 std 的差别；block_on 是把 Notify 用作 waker 的一个实例。

flowchart LR
  subgraph arc["rt::Arc 的依赖关系"]
    inc["RefInc（clone）"]
    dec["RefDec（drop / get_mut / try_unwrap）"]
    ins["Inspect（strong_count）"]
  end
  inc <-->|相关| ins
  dec <-->|相关| ins
  dec <-->|相关| dec
  inc -.-|不相关| inc
  inc -.-|不相关| dec
  subgraph ch["rt::Channel"]
    snd["MsgSend"] <-->|相关| snd
    rcv["MsgRecv"] <-->|相关| rcv
  end

Arc：三种动作

loom::sync::Arc<T> 有两个字段：value: std::sync::Arc<T> 存数据，obj: std::sync::Arc<rt::Arc> 是计数对象（sync/arc.rs:8-12）。真实的引用计数由 std 的 Arc 维护，rt::Arc 另记一份 ref_cnt，用来做调度、同步和泄漏检查（泄漏检测）。

rt::Arc 把操作分成三类（rt/arc.rs:32-47）：

ref_dec：

// Decrement the ref count
state.ref_cnt -= 1;
...
// Synchronize the threads.
state
    .synchronize
    .sync_store(&mut execution.threads, Release);

if state.ref_cnt == 0 {
    // Final ref count, the arc will be dropped. This requires
    // acquiring the causality
    //
    // In the real implementation, this is done with a fence.
    state.synchronize.sync_load(&mut execution.threads, Acquire);
    true
} else {
    false
}

和 std Arc 的 drop 是同一个协议：每次减计数都 Release，最后一个减到 0 的线程再 Acquire（std 用 fence(Acquire)），保证所有持有者对数据的访问都发生在析构之前。loom 用一个 Synchronize 累积所有 release 的时钟，最后一次 acquire 把它们全部并进来。

Arc::new 不是调度点（只在对象仓库里插一项）；clone、drop、strong_count、get_mut 都是。

依赖关系：clone 和谁都不冲突

last_dependent_access：

pub(super) fn last_dependent_access(&self, action: Action) -> Option<&Access> {
    match action {
        // RefIncs are not dependent w/ RefDec, only inspections
        Action::RefInc => self.last_ref_inspect.as_ref(),
        Action::RefDec => self.last_ref_dec.as_ref(),
        Action::Inspect => match self.last_ref_modification {
            Some(RefModify::RefInc) => self.last_ref_inc.as_ref(),
            Some(RefModify::RefDec) => self.last_ref_dec.as_ref(),
            None => None,
        },
    }
}

DPOR 只在「相关」的两次访问之间加回溯点。两次 clone 谁先谁后，结果一样（计数都加 2），不相关；clone 和 drop 也被当作不相关：clone 的线程手里至少有一个引用，计数不可能在这两次操作之间归零，交换顺序不改变「谁是最后一个」。drop 和 drop 相关，因为谁先减到 0 谁负责析构。strong_count 读的是计数，和上一次修改相关。

实测（exp-loom/primitives/tests/prim.rs，两个子线程，主线程 join）：

第三行与第二行做的是「同样的事」，但原子的 RMW 之间两两相关，交错数是 Arc 版本的 3 倍。Arc 的计数在 loom 里不是原子变量，而是一个有专门依赖规则的对象，这是交错数小的原因。

其他 API

• 没有 Weak，也没有 Arc::make_mut、new_cyclic。依赖 Weak 的代码在 loom 下要么换成别的写法，要么那部分不进模型。
• unsized 只能 from_std：Arc<[T]>、Arc<dyn Trait> 不能靠强制转换得到（CoerceUnsized 不稳定），要先建一个独占的 std Arc 再 Arc::from_std（102-118 行）；不独占就 panic「Arc provided to from_std is not unique」。
• into_raw / from_raw：Execution.arc_objs 用数据指针做键记住对应的计数对象，from_raw 查表找回来（207-212 行）；increment_strong_count / decrement_strong_count 建在它们之上。block_on 的 waker 就是这么实现的（见下）。

mpsc：每条消息一个同步点

loom 只有无界的 mpsc::channel，没有 sync_channel。Sender / Receiver 各持一个 std channel 端和一个共享的 rt::Channel（sync/mpsc.rs:5-20）。

rt::Channel 的字段注释讲了为什么同步点要按消息分（rt/mpsc.rs:20-38）：

/// The `mpsc` channels have a guarantee that the messages will be received
/// in the same order in which they were sent. Therefore, if thread `t1`
/// managed to send `m1` before `t2` sent `m2`, the thread that received
/// `m2` can be sure that `m1` was already sent and received. In other
/// words, it is sound for the receiver of `m2` to know that `m1` happened
/// before `m2`. That is why we have a single `sender_synchronize` for
/// senders which we use to "timestamp" each message put in the channel.
/// However, in our example, the receiver of `m1` does not know whether `m2`
/// was already sent or not and, therefore, by reading from the channel it
/// should not learn any facts about `happens_before(send(m2), recv(m1))`.
/// That is why we cannot use single `Synchronize` for the entire channel
/// and on the receiver side we need to use `Synchronize` per message.
sender_synchronize: Synchronize,
/// A synchronization point per message synchronizing the receiving thread
/// with the channel state at the point when the received message was sent.
receiver_synchronize: VecDeque<Synchronize>,

send 把当前线程 release 进 sender_synchronize，再把它的快照推进队列；recv 弹出队首那份快照做 acquire（69-133 行）。收到第 k 条消息的线程，看到的是第 k 条发送时刻为止所有发送者的历史，看不到更晚的发送。如果整个 channel 只用一个同步点，收第一条消息的线程会「看到」还没被它收到的后续发送，多出原本不存在的 happens-before，可能掩盖错误。

依赖关系：发送只和上一次发送相关、接收只和上一次接收相关（165-170 行）。实测（mpsc_two_senders）：两个线程各发一条、主线程收两条，6 个交错，两种接收顺序 (1, 2)、(2, 1) 都出现。

断开语义与 std 不同

recv 在 channel 为空时用 branch_disable 把自己标成阻塞，send 让消息数从 0 变 1 时唤醒等待者。rt 层只数消息，不知道发送方或接收方是否已经 drop。实测（exp-loom/primitives/tests/prim.rs）：

两条都是假阳性：用「发送方全部 drop」通知接收方退出，是 std channel 最常见的关闭方式，在 loom 里会被报成死锁。绕法是在模型里显式发一条「结束」消息。Receiver 的 Drop 会把剩余消息全部 recv 掉（sync/mpsc.rs:82-89），所以正常 drop 的接收方不会触发泄漏，只有 drop 之后还在发送才会。

try_recv 先看消息数（不是调度点），为空直接返回 Empty，否则走 recv。

调用即 panic 的桩

它们存在只是为了让切换层能编译：被测代码 use crate::sync::Barrier 在两种 cfg 下都能编过，但模型一旦走到就失败。

future：block_on 与 AtomicWaker

开 futures feature 才有 loom::future。

block_on：

pub fn block_on<F>(f: F) -> F::Output
where
    F: Future,
{
    pin_mut!(f);

    let notify = Arc::new(rt::Notify::new(false, true));

    let waker = unsafe {
        mem::ManuallyDrop::new(Waker::from_raw(RawWaker::new(
            &*notify as *const _ as *const (),
            waker_vtable(),
        )))
    };

    let mut cx = Context::from_waker(&waker);

    loop {
        match f.as_mut().poll(&mut cx) {
            Poll::Ready(val) => return val,
            Poll::Pending => {}
        }

        notify.wait(location!());
    }
}

waker 指向一个 loom Arc<rt::Notify>：clone 是 Arc 的 RefInc、wake 是 notify、drop 是 RefDec（44-78 行）。Notify 允许伪唤醒（spurious = true），所以 block_on 会在没人 wake 时多 poll 一次，对应 0.2 时代的「futures::block_on polls spuriously (#59)」。一个 future 在 Pending 之后如果没有安排任何唤醒，notify.wait 就会永远阻塞，loom 报死锁。

AtomicWaker 是 tokio / futures 里同名类型的 mock：一个 std Mutex<Option<Waker>> 存 waker，一个 rt::Mutex（seq_cst = false）表示「正在注册或取出」。register 用 try_acquire_lock，抢不到就先 waker.wake() 再 yield_now() 返回（注释「this is a spin lock」）；wake 用阻塞的 acquire_lock 取出 waker。

实验 E18：丢唤醒

（exp-loom/primitives/tests/prim.rs，与 exp-loom/survey/tests/extra.rs 的同名模型一致）一个线程置标志再 AtomicWaker::wake()；主线程 block_on 一个 poll_fn：

第二种是经典的丢唤醒：检查完标志、还没注册时，对方置位并 wake，那时 AtomicWaker 里没有 waker，唤醒落空；主线程随后注册、返回 Pending、在 notify.wait 上永远阻塞。子线程已结束（Terminated），主线程 Blocked，正是死锁检测的条件。死锁报出后的 abort 是 死锁、分支上限与位置信息 讲的展开问题，不影响结论；要用 #[should_panic] 断言这类模型失败行不通，需要单独一个进程跑。

tokio 在 loom 下用的是它自己的 AtomicWaker 实现（经 crate::loom 切换层换成 loom 原子），不是这里的 mock。

小结

• Arc：std Arc 存数据，rt::Arc 记计数；clone = RefInc，drop / get_mut / try_unwrap = RefDec，strong_count = Inspect；drop 每次 release、归零时 acquire，和 std 的协议一致。
• RefInc 只和 Inspect 相关，RefDec 和 RefDec 相关：两线程 clone + drop 30 个交错，同样形状的原子 RMW 90 个。没有 Weak；unsized 靠 from_std。
• mpsc：只有无界 channel；每条消息一个 Synchronize 快照，避免多出 happens-before。rt 层只数消息，发送方断开后 recv 被报成死锁、接收方 drop 后 send 被报成「Messages leaked.」，都是假阳性。
• recv_timeout、Barrier 能编译、一调用就 unimplemented!。
• block_on 用一个允许伪唤醒的 Notify 当 waker；AtomicWaker 是带 yield_now 的自旋锁 mock。先注册后检查 180 个交错通过，先检查后注册报死锁（E18）。