Steal、Retry 与批量窃取

主线位置：任务主线里「窃取者拿到结果之后」。steal 返回的不是 Option<T>，而是三态的 Steal<T>；调度循环怎么处理 Retry，决定了窃取是否可靠。批量窃取把「偷一个」变成「偷一批搬进自己的队列」。

flowchart TB
  attempt["steal() / steal_batch*()"] --> r{"结果"}
  r -->|"Success(t)"| run["执行 t"]
  r -->|"Empty"| next["队列确实空：去下一个来源，或者睡觉"]
  r -->|"Retry"| again["有东西，但 CAS 输了或 buffer 刚换：重试"]
  again --> attempt

Steal：三态

第 2100 行：

#[must_use]
#[derive(PartialEq, Eq, Copy, Clone)]
pub enum Steal<T> {
    /// The queue was empty at the time of stealing.
    Empty,

    /// At least one task was successfully stolen.
    Success(T),

    /// The steal operation needs to be retried.
    Retry,
}

Retry 的含义是「这一次没偷到，但不能断定队列是空的」。它从哪来：

Empty 才是「我看的那一刻队列是空的」。#[must_use] 防止调用者直接忽略结果。

几个组合子：is_empty / is_success / is_retry / success()（转成 Option），以及 or_else（第 2209 行）：

pub fn or_else<F>(self, f: F) -> Steal<T>
where
    F: FnOnce() -> Steal<T>,
{
    match self {
        Steal::Empty => f(),
        Steal::Success(_) => self,
        Steal::Retry => {
            if let Steal::Success(res) = f() {
                Steal::Success(res)
            } else {
                Steal::Retry
            }
        }
    }
}

注意 Retry 的「粘性」：前一个来源返回了 Retry，后一个来源返回 Empty，合起来仍是 Retry。只要有一处「可能还有」，整体就不能说「空」。

FromIterator：一串来源合成一个结果

第 2237 行：

impl<T> FromIterator<Steal<T>> for Steal<T> {
    /// Consumes items until a `Success` is found and returns it.
    ///
    /// If no `Success` was found, but there was at least one `Retry`, then returns `Retry`.
    /// Otherwise, `Empty` is returned.
    fn from_iter<I>(iter: I) -> Steal<T>
    where
        I: IntoIterator<Item = Steal<T>>,
    {
        let mut retry = false;
        for s in iter {
            match &s {
                Steal::Empty => {}
                Steal::Success(_) => return s,
                Steal::Retry => retry = true,
            }
        }

        if retry {
            Steal::Retry
        } else {
            Steal::Empty
        }
    }
}

碰到第一个 Success 就返回，迭代器是惰性的，后面的窃取者根本不会被调用。crate 文档里的 find_task 示例就是靠这个写成的（lib.rs 第 52–76 行）：

fn find_task<T>(
    local: &Worker<T>,
    global: &Injector<T>,
    stealers: &[Stealer<T>],
) -> Option<T> {
    // Pop a task from the local queue, if not empty.
    local.pop().or_else(|| {
        // Otherwise, we need to look for a task elsewhere.
        iter::repeat_with(|| {
            // Try stealing a batch of tasks from the global queue.
            global.steal_batch_and_pop(local)
                // Or try stealing a task from one of the other threads.
                .or_else(|| stealers.iter().map(|s| s.steal()).collect())
        })
        // Loop while no task was stolen and any steal operation needs to be retried.
        .find(|s| !s.is_retry())
        // Extract the stolen task, if there is one.
        .and_then(|s| s.success())
    })
}

本地 pop → 全局批量窃取 → 逐个窃取其他线程，整体结果是 Retry 就从头再来一遍，直到得到 Success 或确定的 Empty。

rayon 怎么写

rayon-core 1.13.0 没有用这个 collect 惯用法，而是手写了等价的循环（registry.rs 第 873–904 行）：

loop {
    let mut retry = false;
    let start = self.rng.next_usize(num_threads);
    let job = (start..num_threads)
        .chain(0..start)
        .filter(move |&i| i != self.index)
        .find_map(|victim_index| {
            let victim = &thread_infos[victim_index];
            match victim.stealer.steal() {
                Steal::Success(job) => Some(job),
                Steal::Empty => None,
                Steal::Retry => {
                    retry = true;
                    None
                }
            }
        });
    if job.is_some() || !retry {
        return job;
    }
}

语义和 FromIterator 一样（第一个成功就停、有 Retry 就整轮重来），多了一个随机起点，避免所有空闲线程都先去偷同一个受害者。全局注入队列和本线程的广播队列也是「Retry 就继续循环」（第 450–458 行、第 749–763 行）。rayon 只用单个 steal()，没有用批量窃取。rayon 卷会展开这个调度循环。

实验 E4：窃取不丢不重，以及 Retry 不能当 Empty

断言：（1）并发 push / pop / steal 下每个任务恰好被取一次；（2）把 Retry 当成 Empty 退出的窃取者，会让工作集中到少数线程上。

规划阶段已经跑过第一条（exp-crossbeam/survey/src/bin/deque_steal.rs）：主人 push 200 万个（每 push 3 个 pop 1 个），11 个窃取者循环 steal，FIFO 和 LIFO 都取到 2000000 个、总和正确；Retry 约 570 万次，比成功次数还多。

方法（exp-crossbeam/deque-retry/）：主人先 push 好 n 个任务，11 个窃取者用 Barrier 同时出发，循环 steal：

• 正确写法：Success 继续，Retry 继续，Empty 退出。
• 错误写法：Retry 和 Empty 都退出。

另一组让主人在窃取者出发的同时也从自己那头 pop。每个任务可以附带一段空转（2000 次整数乘加，用 black_box 防优化），模拟真实工作。统计每个线程拿到多少、总共多少、Retry 多少次、耗时。release，每组 3 次。耗时那一批测量开始时 1 分钟负载 2.89、结束时 4.34（包括本实验自己的 12 个线程）。M2 Max（8P + 4E）。

结果（3 次的范围）：

所有组合里任务都一个不少（「剩余」全为 0），也没有重复。差别在谁干的活：

• 错误写法下，11 个窃取者里有 10 个在第一次撞车时就退出了，剩下一个独自干完全部（每任务有工作量时耗时 8.6 倍）。主人也 pop 时，11 个窃取者全部很快退出，99.8% 的任务由主人一个人做完。
• 任务没有工作量时，错误写法反而快 20 倍：单线程独占时没有任何 CAS 竞争，而 12 个线程争一个 front 每次成功前平均要撞约 1.5 次（100 万次成功对应约 150 万次 Retry）。这说明的是「对微小任务逐个窃取不划算」，不是错误写法更好。

在调度器里，「窃取者以为没活干」的后果是它去睡觉；这里的实验没有睡眠机制，所以表现为「过早退出」。不管哪种，并行度都没了，而程序仍然算出正确结果，只是慢，这是这类 bug 难被发现的原因。常见坑有一条专门讲它。

反向验证：同一组任务、同样的线程数，正确写法下每个窃取者拿到约 1 万个（10 万 ÷ 11 ≈ 9 091，加上竞争带来的不均），退出前一律看到的是 Empty；错误写法的「因 Retry 退出」计数是 10 或 11，说明退出确实是 Retry 造成的，而不是队列真的空了。

批量窃取

单个 steal 每偷一个任务都要 pin、fence、CAS 一次。批量窃取一次 CAS 认领一批，搬进调用者自己的 Worker：

steal_batch_with_limit（第 761 行）的骨架：

1. 偷自己（Arc::ptr_eq(&self.inner, &dest.inner)）：直接按 dest 是否为空返回 Empty / Success(())，什么都不搬。这是 0.7.3 的 #448（「Stop stealing from the same deque」）。
2. 读 front、fence（借 pin）、读 back，和 steal 一样。
3. batch_size = min((len + 1) / 2, limit)，即向上取整的一半；dest.reserve(batch_size)，不够就扩容。
4. 按源队列的 flavor 分两种做法（下面）。
5. Release fence + 写 dest.back，把搬进来的任务发布给 dest 的窃取者。

steal_batch_with_limit_and_pop 的 batch_size 是 min((len - 1) / 2, limit - 1)，再加上直接返回的那一个，总数同样不超过 limit。

源是 FIFO：一次 CAS 认领一批

Flavor::Fifo => {
    // Copy the batch from the source to the destination buffer.
    match dest.flavor {
        Flavor::Fifo => {
            for i in 0..batch_size {
                unsafe {
                    let task = buffer.deref().read(f.wrapping_add(i));
                    dest_buffer.write(dest_b.wrapping_add(i), task);
                }
            }
        }
        Flavor::Lifo => {
            for i in 0..batch_size {
                unsafe {
                    let task = buffer.deref().read(f.wrapping_add(i));
                    dest_buffer.write(dest_b.wrapping_add(batch_size - 1 - i), task);
                }
            }
        }
    }

    // Try incrementing the front index to steal the batch.
    // If the buffer has been swapped or the increment fails, we retry.
    if self.inner.buffer.load(Ordering::Acquire, guard) != buffer
        || self
            .inner
            .front
            .compare_exchange(
                f,
                f.wrapping_add(batch_size),
                Ordering::SeqCst,
                Ordering::Relaxed,
            )
            .is_err()
    {
        return Steal::Retry;
    }

    dest_b = dest_b.wrapping_add(batch_size);
}

（第 808–847 行）

先把整批拷进 dest 的 buffer（写在 dest.back 之后，dest 的窃取者还看不见），再一次 CAS 把 front 推过整批。CAS 失败就当没拷过：dest.back 没动，那些槽等于没写。FIFO 源的主人也从 front 取（fetch_add），所以一次 CAS 就能独占整段。

源是 LIFO：只能一个一个来

LIFO 源的主人从 back 取，而且 len > 0 时不碰 front（Chase-Lev 双端队列）。如果窃取者一次把 front 推过 k 个，主人可能同时已经从 back 那头把其中几个拿走了，CAS front 照样成功，同一个任务就被取了两次。所以 LIFO 源只能逐个偷：每偷一个之前都重新 fence、重新读 back、确认还够，再 CAS front + 1（第 850–923 行）。中途 CAS 失败就停下，已经偷到的算数；一个都没偷到才返回 Retry。

搬过去以后的顺序

批量窃取要保证：dest 的主人 pop 这批任务的顺序，和源队列的主人本来会 pop 它们的顺序一致。所以写进 dest 时要看两边的 flavor：FIFO 源搬进 LIFO 目标时倒着写；LIFO 源搬进 FIFO 目标时偷完再整体翻转一次。这是 2019 年 1 月 4d574d40（「Use correct order when stealing between FIFO and LIFO queues」）定下的。实测（deque-retry order，源队列 push 1 到 8，steal_batch 一次）：

窃取者总是从 front（最老的一头）偷，所以偷到的是 1 到 4；四种组合下目标队列 pop 的相对顺序都和源主人一致。

上限 32 是怎么来的

MAX_BATCH = 32（第 39 行）。这个上限从一开始就有：2018 年 11 月合并进仓库时是 128，2019 年 1 月 13 日 e97ac760（「Polish up」）改成 32；1 月 17 日 af8ce60f（「Tweak batch size」）让 Injector::steal_batch_and_pop 用 MAX_BATCH + 1。2022 年 8 月 34dd9d5d（「limit batch steal size」，#903，0.8.3）没有改这个数，而是加了 *_with_limit 系列方法让调用者自己定上限，原来的四个方法改成用 MAX_BATCH 调它们；Injector 那里的 MAX_BATCH + 1 也是这时加上了 TODO 注释（Injector）。

实验 E4（续）：steal 与 steal_batch_and_pop

方法（deque-retry drain_batch）：主人先 push 100 万个，11 个窃取者各带一个自己的 FIFO Worker，循环调 steal() 或 steal_batch_and_pop(&local)，拿到后把 local 里的也 pop 干净，Empty 时退出。统计返回 Success 的调用次数、Retry 次数、耗时。每组 3 次，负载同上一批。

• FIFO 源：100 万 ÷ 31 255 ≈ 32，每次正好拿满上限，三次运行完全一样；耗时降到 1/27。
• LIFO 源：批量窃取几乎没有收益，平均每次成功只拿到 3.4 – 3.6 个，Retry 反而更多。原因就是上一节：LIFO 源只能逐个 CAS，11 个线程在同一个 front 上一个一个地抢。

所有组合都取到 1000000 个、总和正确。换成规划阶段 survey 那种「主人边 push 边 pop、队列几乎总是快空」的形状（deque-retry batch），两种方法的成功调用次数差不多（约 142 万对 134 万），批量窃取派不上用场；批量的收益只在源队列攒了足够多任务时出现。

小结

• Steal<T> 有三态：Retry 是「有东西但撞车了」，不能当 Empty。or_else 和 FromIterator 都让 Retry 有粘性：只要有一个来源说「可能还有」，整体就是 Retry。
• FromIterator 碰到第一个 Success 就停；crate 文档的 find_task 用它写成「本地 → 全局 → 逐个窃取，Retry 就整轮重来」。rayon 手写了等价的循环，并且随机选起点。
• 实验 E4：把 Retry 当 Empty，任务不丢不重，但 11 个窃取者里 10 个第一次撞车就退出，有工作量时耗时 8.6 倍；主人也在 pop 时 99.8% 的任务落到主人一个人头上。
• 批量窃取一次最多 32 个（*_with_limit 可调）。FIFO 源一次 CAS 认领整批，实测调用次数降到 1/32、耗时降到 1/27；LIFO 源只能逐个 CAS，几乎没有收益。
• 搬过去的一批，目标主人 pop 的顺序和源主人本来的顺序一致。