Rust in Everything
基于 crossbeam-deque 0.8.8

Cell<Buffer<T>> cannot be shared between threads safely

主线位置：一个任务的起点，Worker 和 Stealer 的分工。为什么 Worker 只能有一个主人，见 Chase-Lev 双端队列。

flowchart TB
  w["Worker&lt;T&gt;：Send，!Sync"] --> own["只有主人写 back，push / pop 不用 CAS"]
  w -->|"w.stealer()"| s["Stealer&lt;T&gt;：Send + Sync，可克隆"]
  s --> front["只 CAS front"]
  inj["Injector&lt;T&gt;：Sync"] --> multi["多个线程都能 push"]
  arc["Arc&lt;Worker&lt;T&gt;&gt; 跨线程"] --> err["E0277：Cell&lt;Buffer&lt;T&gt;&gt; cannot be shared"]

现象

想让几个线程都往同一个 deque 里 push：

let w = Arc::new(Worker::<u32>::new_lifo());
let w2 = Arc::clone(&w);
thread::spawn(move || {
  w2.push(1);
})
.join()
.unwrap();
w.pop();

实测

exp-crossbeam/worker-not-sync/（crossbeam-deque 0.8.8，rustc 1.99.1，cargo build --features broken；本机路径已去掉）。三个错误，第一个是：

error[E0277]: `Cell<crossbeam_deque::deque::Buffer<u32>>` cannot be shared between threads safely
   --> src/main.rs:10:17
    |
 10 |     thread::spawn(move || {
    |  ___-------------_^
    | |   |
    | |   required by a bound introduced by this call
 11 | |     w2.push(1);
 12 | |   })
    | |___^ `Cell<crossbeam_deque::deque::Buffer<u32>>` cannot be shared between threads safely
    |
    = help: within `Worker<u32>`, the trait `Sync` is not implemented for `Cell<crossbeam_deque::deque::Buffer<u32>>`
    = note: if you want to do aliasing and mutation between multiple threads, use `std::sync::RwLock`
note: required because it appears within the type `Worker<u32>`
   --> .../crossbeam-deque-0.8.8/src/deque.rs:215:12
    |
215 | pub struct Worker<T> {
    |            ^^^^^^
    = note: required for `Arc<Worker<u32>>` to implement `Send`
note: required because it's used within this closure

另外两个的标题是：

error[E0277]: `*mut ()` cannot be shared between threads safely
error[E0277]: `*mut u32` cannot be shared between threads safely
error: could not compile `worker-not-sync` (bin "worker-not-sync") due to 3 previous errors

报错里没有出现 Worker 不是 Sync 这几个字：编译器报的是 Worker 的字段。Arc<T> 要 Send，需要 T: Send + Sync；Worker 里有三个字段不是 Sync，各报一次。

从哪来

Worker 的定义：

pub struct Worker<T> {
    /// A reference to the inner representation of the queue.
    inner: Arc<CachePadded<Inner<T>>>,

    /// A copy of `inner.buffer` for quick access.
    buffer: Cell<Buffer<T>>,

    /// The flavor of the queue.
    flavor: Flavor,

    /// Indicates that the worker cannot be shared among threads.
    _marker: PhantomData<*mut ()>, // !Send + !Sync
}

unsafe impl<T: Send> Send for Worker<T> {}
  • buffer: Cell<Buffer<T>>：主人自己缓存一份 buffer 指针，不走原子；Cell 不是 Sync。Buffer<T> 里有 *mut T，这是第三个错误里的 *mut u32。
  • _marker: PhantomData<*mut ()>：显式地让它既不 Send 也不 Sync，再用 unsafe impl Send 把 Send 加回来。结果是 Send 不 Sync。

这是算法的要求，不是实现上的偷懒：Chase-Lev 双端队列的 back 只有主人一个写者，push 和 pop 写 back 时不用 CAS（Chase-Lev 双端队列）。两个线程同时 push，会写同一个槽、互相覆盖 back。!Sync 让编译器挡住这种用法。Stealer 只碰 front，用 CAS，所以是 Send + Sync、可以克隆。

怎么改

同一个程序里的三种能编译的写法，输出都是 sum=45：

// 1. Worker 留在本线程，别的线程拿 Stealer
let w = Worker::<u32>::new_lifo();
let s = w.stealer();
let h = thread::spawn(move || {
  let mut sum = 0;
  loop {
    match s.steal() {
      Steal::Success(v) => sum += v,
      Steal::Retry => continue,
      Steal::Empty => break sum,
    }
  }
});

// 2. Worker 是 Send：整个 move 进要当主人的线程
let w = Worker::<u32>::new_fifo();
thread::spawn(move || {
  for i in 0..10 {
    w.push(i);
  }
  // ...
});

// 3. 能编译，但主人被串行化了
let w = Arc::new(Mutex::new(Worker::<u32>::new_lifo()));
  • 「多个线程都要往里放」：那是 Injector 的用途，它是 Sync 的全局 FIFO 队列，任何线程都能 push、steal。rayon 的结构就是「每个工作线程一个 Worker + 一个全局 Injector + 互相持有 Stealer」。
  • 「一个线程放、多个线程取」：Worker + 若干 Stealer（第 1 种）。
  • Arc<Mutex<Worker>>：能编译、能用，但每次 push / pop 都要拿锁，Chase-Lev「主人不用 CAS」的好处没了，不如直接用 Injector 或一个 channel。

小结

  • Worker<T> 是 Send 不是 Sync；放进 Arc 跨线程，报错的是它的字段：Cell<Buffer<T>>、*mut ()、*mut T。
  • 原因是 Chase-Lev 只允许一个主人写 back。
  • 本线程留 Worker、别人拿 Stealer；多个生产者用 Injector。

💬 关联讨论

发起讨论
加载中...