Rust in Everything
基于 crossbeam-channel 0.5.17

最小有界 channel

主线位置：主线 1「一条消息」的缩小版。非阻塞部分对照 array，阻塞部分对照 Context 与 Waker。代码在 exp-crossbeam/hands-on/src/mini_channel.rs。

sequenceDiagram
  participant S as 发送方
  participant C as 环
  participant W as senders 等待列表
  participant R as 接收方
  S->>C: try_send：满
  S->>S: Backoff snooze，约 11 次后放弃自旋
  S->>W: register：挂上自己，claimed = false
  S->>C: is_full() 复查：仍满
  S->>S: park，直到 claimed 为真
  R->>C: try_recv 成功，stamp 前进一圈
  R->>W: notify_one：CAS 某个条目的 claimed，unpark
  S->>W: unregister
  S->>C: 回到 try_send

环：从 ArrayQueue 搬过来

每个槽一个 stamp，head / tail 低位是下标、高位是圈数，one_lap = (cap + 1).next_power_of_two()。和 ArrayQueue 一样不需要 mark 位，因为这个 channel 没有断开。

pub fn try_send(&self, msg: T) -> Result<(), T> {
  let backoff = Backoff::new();
  let mut tail = self.tail.load(Ordering::Relaxed);
  loop {
    let slot = &self.buffer[tail & (self.one_lap - 1)];
    let stamp = slot.stamp.load(Ordering::Acquire);
    if stamp == tail {
      // empty in this lap: claim it
      match self.tail.compare_exchange_weak(tail, self.next(tail), Ordering::SeqCst, Ordering::Relaxed) {
        Ok(_) => {
          unsafe { slot.msg.get().write(MaybeUninit::new(msg)) };
          slot.stamp.store(tail + 1, Ordering::Release);
          self.receivers.notify_one();
          return Ok(());
        }
        Err(t) => {
          tail = t;
          backoff.spin();
        }
      }
    } else if stamp.wrapping_add(self.one_lap) == tail + 1 {
      // still holds last lap's message: full unless head moved
      atomic::fence(Ordering::SeqCst);
      let head = self.head.load(Ordering::Relaxed);
      if head.wrapping_add(self.one_lap) == tail {
        return Err(msg);
      }
      backoff.spin();
      tail = self.tail.load(Ordering::Relaxed);
    } else {
      // another sender claimed it and is still writing
      backoff.snooze();
      tail = self.tail.load(Ordering::Relaxed);
    }
  }
}

三个分支就是 array 那张表：stamp == tail 可写；stamp + one_lap == tail + 1 且 head + one_lap == tail 才是真满；其余是别的发送方占了槽还没写完。CAS 失败直接用返回的当前值继续，没有再 load 一次：vs-std 的二分实验里，std list 在这里多读一次是 4P1C 无界吞吐差 4 倍的主要来源。try_recv 是镜像，读完把 stamp 置成 head + one_lap，然后 senders.notify_one()。

Drop 只是 while self.try_recv().is_some() {}，把还在缓冲里的消息析构掉。

阻塞：注册、复查、park

Waiters 是一把锁下的 VecDeque<Entry>，外加一个 is_empty: AtomicBool 让没人等时 notify_one 不碰锁（对应 SyncWaker 的同名字段）。每个条目带一个共享的 claimed 标志：

struct Entry {
  thread: Thread,
  claimed: Arc<AtomicBool>,
}

fn notify_one(&self) {
  if self.is_empty.load(Ordering::SeqCst) {
    return;
  }
  let mut list = self.list.lock().unwrap();
  while let Some(e) = list.pop_front() {
    if e.claimed.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire).is_ok() {
      e.thread.unpark();
      break;
    }
  }
  self.is_empty.store(list.is_empty(), Ordering::SeqCst);
}

send：

pub fn send(&self, mut msg: T) {
  loop {
    // 1. spin / yield a few times
    let backoff = Backoff::new();
    loop {
      match self.try_send(msg) {
        Ok(()) => return,
        Err(m) => msg = m,
      }
      if backoff.is_completed() {
        break;
      }
      backoff.snooze();
    }
    // 2. register, then re-check: a receiver may have made room in between
    let claimed = self.senders.register();
    if RECHECK && !self.is_full() && claimed.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire).is_ok() {
      self.senders.unregister(&claimed);
      continue;
    }
    // 3. park until a receiver claims our entry (or we already were claimed)
    while !claimed.load(Ordering::Acquire) {
      thread::park();
    }
    self.senders.unregister(&claimed);
  }
}

（RECHECK 平时是 true，--features no_recheck 时为 false，用于下面的反向验证。）

几处和 crossbeam 一一对应：

这里crossbeam防的是什么
第 1 段 Backoff 循环array send 里的内层循环对面马上就会腾出位置时，不必付一次 park / unpark
register 后 is_full() 复查「Has the channel become ready just now?」丢唤醒：接收方在「我判断满」和「我挂上去」之间读走了一条，它 notify 时我还不在列表里
复查成功时 CAS 自己的 claimedcx.try_select(Selected::Aborted)放弃之后，notify_one 会跳过我、去叫下一个，这次通知不会浪费在一个本来就要成功的线程上
notify_one 里 CAS 别人的 claimedWaker::try_select CAS 对方的 Context同一个等待者只被选中一次；唤醒和放弃互斥
while !claimed { park() }Context::wait_until 的循环park 可能假醒；unpark 先于 park 时 park 立即返回
醒来后回到 try_send醒来后重新 start_send被叫醒只说明「刚才有空位」

复查为什么能和 notify_one 配上：发送方 register 里 is_empty.store(false, SeqCst)，之后 is_full() 用 SeqCst 读 head、tail；接收方先 SeqCst CAS head，之后 notify_one 用 SeqCst 读 is_empty。四个都是 SeqCst，在全局顺序里，要么发送方的 store(false) 在接收方的 load 之前（接收方看到有人在等，去叫），要么接收方的 CAS 在发送方读 head 之前（发送方复查时看到不满，不睡）。两边都错过对方的情况不存在。crossbeam 的 SyncWaker 也是同样的理由要求 is_empty 用 SeqCst（Context 与 Waker）。

测试与反向验证

tests/channel.rs：容量 1、2、16、1024，形状 1P1C、4P1C、1P4C、4P4C，每个生产者发 5 万个互不相同的数，消费者按总数分摊着收，最后核对总和，并确认环已空；另有一个 60 秒看门狗，超时就 abort。另外两个小测试检查满 / 空的边界和 Drop 析构缓冲消息。

正常构建连跑 6 次全部通过，整组 16 种形状约 0.37 秒（每次）。

反向验证：--features no_recheck 去掉复查，其余不动，同一个测试跑 3 次：

watchdog: test still running after 60s (lost wakeup?)
--
watchdog: test still running after 60s (lost wakeup?)
--
cap=   1 1P1C ok in 7.233667ms
watchdog: test still running after 60s (lost wakeup?)

前两次卡在第一个形状（容量 1、1P1C），第三次过了它、卡在容量 1、4P1C。容量 1 时「满」和「空」来回切换最频繁，丢唤醒的窗口最容易被撞上。去掉复查以后没有任何东西会把睡着的线程叫醒，表现就是卡死而不是变慢，这正是 register 之后那一行存在的理由。

省掉了什么

省掉的crossbeam 怎么做
断开tail 的 mark 位；Counter 分别数两端，最后一方释放（API 与 flavor）
超时wait_until(deadline) 里 park_timeout；注意先尝试后看表（常见坑）
select一个线程本地 Context 同时挂在多个 channel 上，选中状态是一个 AtomicUsize（select）
每次阻塞分配一个 Arc<AtomicBool>Context 缓存在线程本地，阻塞前 reset
等待者列表用 std::sync::Mutexutils::Mutex，忽略 poison
接收端走光时丢弃消息0.5.17 的 array 也没有，main 的 #1121 才有（常见坑）

小结

  • 非阻塞部分直接搬 ArrayQueue 的 stamp 环：可写、真满、别人正在写三个分支，CAS 失败用返回值。
  • 阻塞部分四步：Backoff 自旋 → 注册 → 复查 → park；复查和 notify 两侧都用 SeqCst，才能排除丢唤醒。
  • 每个等待者一个 claimed 标志，唤醒方和放弃方各 CAS 一次，保证一次通知只落在一个真正在等的线程上。
  • 压力测试 16 种形状通过；去掉复查，3 次测试 3 次卡死。

💬 关联讨论

发起讨论
加载中...