ArrayQueue 与 SegQueue

主线位置：主线 1 去掉阻塞的那一半。ArrayQueue::push 就是 array 的 start_send 加 write，SegQueue::push 就是 list 的同一对；队列满了或空了直接返回，不注册、不 park。

flowchart LR
  arr["channel array flavor"] -->|"去掉 SyncWaker、mark 位、Token"| aq["ArrayQueue"]
  aq -->|"多一个"| fp["force_push：满了挤掉最旧的"]
  lst["channel list flavor"] -->|"去掉 SyncWaker、断开"| sq["SegQueue"]
  lst -->|"MARK_BIT 改名"| hn["HAS_NEXT：只剩 head 上的含义"]
  aq --> mut["push_mut / pop_mut：&mut self，免原子（0.3.13）"]
  sq --> mut

crossbeam-queue 只有两个文件：array_queue.rs 645 行，seg_queue.rs 713 行（含文档注释）。两个队列都是 MPMC、&self 方法、Send + Sync（T: Send）。没有阻塞版本：想要「空了就等」的语义，用 channel。

ArrayQueue：没有 mark 位的 stamp 环

pub struct ArrayQueue<T> {
    head: CachePadded<AtomicIndex>,
    tail: CachePadded<AtomicIndex>,
    buffer: Box<[Slot<T>]>,
    cap: usize,
    one_lap: Index,
}

和 array channel 逐项比：

one_lap 取 cap + 1 的下一个 2 的幂（new 里的注释是「One lap is the smallest power of two greater than cap」）。判定规则和 channel 完全相同：stamp == tail 可写，stamp == head + 1 可读，stamp + one_lap == tail + 1 时加 fence(SeqCst) 读 head 判满。

push_or_else 把「看起来满」时怎么办做成了参数 f：

} else if stamp.wrapping_add(self.one_lap) == tail + 1 {
    atomic::fence(Ordering::SeqCst);
    value = f(value, tail, new_tail, slot)?;
    backoff.spin();
    tail = self.tail.load(Ordering::Relaxed);
}

f 返回 Err(v) 就把元素还给调用方结束，返回 Ok(v) 就继续循环。push 的 f 读 head，真满就 Err。

force_push：满了挤掉最旧的

force_push 的 f 不放弃，而是替消费者把 head 推前一格，自己占用这个槽：

pub fn force_push(&self, value: T) -> Option<T> {
    self.push_or_else(value, |v, tail, new_tail, slot| {
        let head = tail.wrapping_sub(self.one_lap);
        let new_head = new_tail.wrapping_sub(self.one_lap);

        // Try moving the head.
        if self
            .head
            .compare_exchange_weak(head, new_head, Ordering::SeqCst, Ordering::Relaxed)
            .is_ok()
        {
            // Move the tail.
            self.tail.store(new_tail, Ordering::SeqCst);

            // Swap the previous value.
            let old = unsafe { slot.value.get().replace(MaybeUninit::new(v)).assume_init() };

            // Update the stamp.
            slot.stamp.store(tail + 1, Ordering::Release);

            Err(old)
        } else {
            Ok(v)
        }
    })
    .err()
}

满的时候 head 恰好比 tail 落后一圈，所以 head = tail - one_lap。CAS head 成功，就等于「以消费者的身份读走了最旧的那个」，接着把 tail 也推前一格，原地换出旧值、写入新值，stamp 置为 tail + 1（新的一圈、有数据）。旧值作为 Some(old) 还给调用方（借用 Err 通道返回，最后 .err() 转回来）。和真消费者抢 head 的 CAS 失败，就说明刚好有人读走了一个，不满了，按普通 push 重试。

这适合「只保留最近 N 个」的场景：日志尾巴、最近的采样点。channel 没有对应的操作。

len：要等 tail 稳定

len 和 array channel 一字不差：

loop {
    // Load the tail, then load the head.
    let tail = self.tail.load(Ordering::SeqCst);
    let head = self.head.load(Ordering::SeqCst);

    // If the tail didn't change, we've got consistent values to work with.
    if self.tail.load(Ordering::SeqCst) == tail {
        let hix = (head & (self.one_lap - 1)) as usize;
        let tix = (tail & (self.one_lap - 1)) as usize;
        return if hix < tix { tix - hix }
            else if hix > tix { self.cap - hix + tix }
            else if tail == head { 0 }
            else { self.cap };
    }
}

head 和 tail 是两个原子，不能一次读出一对一致的值。读 tail、读 head、再读 tail：两次 tail 相同，说明这段时间没有生产者推进，head 只会前进，算出的长度至少在某一刻成立过。生产者一直在推进时，第二次读总是不同，循环就一直转；每一轮是 3 次 SeqCst 读，其中两次碰的是生产者在写的那条缓存行。

sqlx 卷里那条注释说的就是这件事（获取连接）：idle_conns 是 ArrayQueue，sqlx 另外维护一个原子计数 num_idle，理由是「ArrayQueue::len() 要等头尾停止变化，高并发时可能很慢」。这个说法和源码对得上：len 没有上界，高并发下会多转几圈。

is_empty / is_full 只读一次 head、一次 tail，不循环，结果可能一出来就过时。

push_mut / pop_mut（0.3.13）

0.3.13（#1191）给两个队列都加了 &mut self 版本：独占引用说明没有别的线程在用，可以用 get_mut() 直接读写原子里的值，不做任何原子操作，也不需要 stamp 判定（push_mut、pop_mut）。适合「单线程建好、再交给多线程」或「多线程用完、单线程收尾」的阶段。

SegQueue：list channel 去掉阻塞

pub struct SegQueue<T> {
    head: CachePadded<Position<T>>,
    tail: CachePadded<Position<T>>,
    _marker: PhantomData<T>,
}

常量和 list channel 一样：LAP = 32、BLOCK_CAP = 31、SHIFT = 1，槽状态 WRITE / READ / DESTROY。唯一改名的是最低位：

/// Indicates that the block is not the last one.
const HAS_NEXT: Index = 1;

channel 的 MARK_BIT 在 tail 上表示断开、在 head 上表示后面还有块；队列没有断开，只剩后一个含义，所以叫 HAS_NEXT。push 和 list 的 start_send 加 write 是同一段逻辑：块尾 snooze、倒数第二个槽提前分配下一块、第一块懒分配；pop 读完用 READ / DESTROY 接力释放块（见 list）。SegQueue::new 是 const fn，可以放进 static。

0.3.12 的「Fix stack overflow when pushing large value」（#1146、#1147、#1159）就是 list channel 那次「Block 直接在堆上建」的同款修复。

len 同样是「读 tail、读 head、再读 tail」的循环，算法上多两步：落在块尾空序号上的索引往后挪一格；把两个索引平移到 head 所在的块从 0 开始，再减去中间经过的块数（每块有一个不存元素的序号）。

两个队列和 channel 怎么选

吞吐上，SegQueue 每 31 个元素要一次分配，文档自己说「somewhat slower than ArrayQueue」。本卷没有单独测队列，channel 一侧的数字见 compare。

谁在用

• pingora：pingora-pool 的每组连接有一个 16 格的 ArrayQueue 当热队列，满了再落进 Mutex<HashMap>；ArrayQueue 先进先出，所以同一组里最早放回的连接先被复用（连接池）。TinyUFO 的 small / main 两条队列是 SegQueue<Key>（TinyUFO）。本站网关 crates/gateway/Cargo.lock 里锁的是 crossbeam-queue 0.3.12。
• sqlx：连接池的 idle_conns 是容量为 max_connections 的 ArrayQueue，长度另用原子计数，原因见上面的 len。
• 本站框架：sqlx-core 带进来 crossbeam-queue 0.3.12。0.3.12 到 0.3.14 的变化是 push_mut / pop_mut（新 API）、32 位平台的 64 位索引和 MSRV 政策，对 64 位平台上的现有调用没有行为变化。

小结

• ArrayQueue 是 array channel 的 stamp 环去掉阻塞、断开和 Token：one_lap 直接取 cap + 1 的下一个 2 的幂，判定等式不变。
• force_push 在满的时候以消费者身份 CAS 推进 head，原地换出最旧的元素；抢不过真消费者就按普通 push 重试。
• len() 读 tail、head、tail，直到两次 tail 相同；生产者忙时可能转很多圈，sqlx 因此另记计数。
• SegQueue 是 list channel 去掉阻塞：31 槽一块、READ / DESTROY 接力释放，最低位改名 HAS_NEXT。
• 0.3.13 加了 &mut self 的 push_mut / pop_mut，独占时不做原子操作。