14.3 work stealing 队列的真实实现
14.2 的线程池让所有 worker 抢同一条队列。任务多、核也多的时候，那把锁就成了新的瓶颈：每次出队都要抢锁，抢锁的线程越多，真正干活的时间越少。work stealing 换了个思路——每个 worker 有一条自己的双端队列，平时只操作自己的队列（几乎不抢锁），只有自己空了，才去别人的队列另一头「偷」一个。这一节把它真写出来，并用不均衡负载量出收益。
Chase-Lev 双端队列：两端各取一头
每个 worker 一条双端队列（deque），两端各有用处：
| 端 | 谁用 | 取哪个任务 | 为什么 |
|---|---|---|---|
| bottom（队尾） | 队列的拥有者 | 最后压入的（LIFO） | 刚压入的任务数据还在缓存里，局部性最好 |
| top（队头） | 窃取者 | 最早压入的（FIFO） | 最老的任务往往是「大块」，偷走它最能缓解不均 |
关键约定是：拥有者只碰 bottom，窃取者只碰 top。两端几乎不会撞在一起，所以大多数操作不需要锁。真正需要同步的只有「队列只剩最后一个元素」的瞬间——拥有者和窃取者可能同时想拿它，这时用 Atomics.compareExchange 在 top 上定胜负。
这就是 Chase-Lev 队列（Chase 与 Lev 在 2005 年提出的算法）。它的正确性依赖两点：bottom 由拥有者用原子写发布、窃取者用原子读观察；top 由双方用 CAS 竞争。JS 的 Atomics 默认是顺序一致（seq_cst），刚好省掉了手写内存屏障的麻烦——这也是它能在 JS 里被干净地写出来的原因。
可运行产物：deque、worker 与实测脚本
队列本身是 src/core/deque.ts：一组作用在共享 Int32Array 上的纯函数，push / pop 给拥有者，steal 给窃取者。层级 L1。
// file: src/core/deque.ts
// tier: L1
// verify: npx vitest run src/core/deque.test.ts
//
// helio 内核：Chase-Lev 工作窃取双端队列（work-stealing deque）。
// 所有状态放在一个共享 Int32Array 里，索引运算保证「取」与「窃取」无锁。
// 约定：每个 deque 占 (2 + cap) 个 Int32 字——top、bottom、cap 个槽位。
export const EMPTY = -1;
/** deque d 的 top 字下标。 */
export function topIndex(base: number, cap: number, d: number): number {
return base + d * (2 + cap);
}
/** deque d 的 bottom 字下标。 */
export function bottomIndex(base: number, cap: number, d: number): number {
return base + d * (2 + cap) + 1;
}
/** deque d 的槽位区起始字下标。 */
export function slotIndex(base: number, cap: number, d: number, i: number): number {
return base + d * (2 + cap) + 2 + (i % cap);
}
/** 拥有者从 bottom 端压入（只有本线程调用）。 */
export function push(words: Int32Array, base: number, cap: number, d: number, task: number): void {
const b = Atomics.load(words, bottomIndex(base, cap, d));
Atomics.store(words, slotIndex(base, cap, d, b), task);
Atomics.store(words, bottomIndex(base, cap, d), b + 1);
}
/** 拥有者从 bottom 端弹出（LIFO，只有本线程调用）。 */
export function pop(words: Int32Array, base: number, cap: number, d: number): number {
const bi = bottomIndex(base, cap, d);
let b = Atomics.load(words, bi) - 1;
Atomics.store(words, bi, b);
const t = Atomics.load(words, topIndex(base, cap, d));
if (t <= b) {
const x = Atomics.load(words, slotIndex(base, cap, d, b));
if (t === b) {
// 只剩最后一个：与窃取者竞争 top
if (Atomics.compareExchange(words, topIndex(base, cap, d), t, t + 1) !== t) {
Atomics.store(words, bi, b + 1);
return EMPTY;
}
Atomics.store(words, bi, b + 1);
}
return x;
}
Atomics.store(words, bi, b + 1);
return EMPTY;
}
/** 窃取者从 top 端拿走一个（FIFO，任意线程可调用）。 */
export function steal(words: Int32Array, base: number, cap: number, d: number): number {
const t = Atomics.load(words, topIndex(base, cap, d));
const b = Atomics.load(words, bottomIndex(base, cap, d));
if (t < b) {
const x = Atomics.load(words, slotIndex(base, cap, d, t));
if (Atomics.compareExchange(words, topIndex(base, cap, d), t, t + 1) === t) {
return x;
}
}
return EMPTY;
}
worker 侧 src/core/steal-worker.mjs 先在启动闸门上阻塞，放行后反复 pop 自己的队列；空了就轮流到别的 deque 上 steal。stealing=false 时只取不偷，用来做对照组。
// file: src/core/steal-worker.mjs
// tier: L1
// verify: node src/core/work-stealing.ts
//
// 工作窃取 worker：先在启动闸门上阻塞，放行后反复从自己 deque 的 bottom 端
// 取任务；取空了就轮流到别的 deque 的 top 端窃取。stealing=false 时只取不窃取。
import { parentPort, workerData } from 'node:worker_threads';
import { EMPTY, pop, steal } from './deque.ts';
const { sab, P, cap, workerIndex, costs, stealing } = workerData;
const words = new Int32Array(sab);
const READY = 0, START = 1, DONE = 2;
const DEQUE_BASE = 4;
const RESULTS_BASE = DEQUE_BASE + P * (2 + cap);
const ITER = 300_000;
function spin(units) {
let x = 0x9e3779b9 >>> 0;
const n = units * ITER;
for (let i = 0; i < n; i++) {
x ^= x << 13; x >>>= 0;
x ^= x >>> 17;
x ^= x << 5; x >>>= 0;
}
return x;
}
// 启动闸门：全部就绪后由主线程放行，避免把 worker 冷启动算进 makespan。
Atomics.add(words, READY, 1);
Atomics.wait(words, START, 0);
let processed = 0;
for (;;) {
let id = pop(words, DEQUE_BASE, cap, workerIndex);
if (id === EMPTY && stealing) {
for (let v = 1; v < P && id === EMPTY; v++) {
id = steal(words, DEQUE_BASE, cap, (workerIndex + v) % P);
}
}
if (id === EMPTY) break;
spin(costs[id]);
Atomics.store(words, RESULTS_BASE + id, workerIndex + 1);
processed++;
}
Atomics.add(words, DONE, 1);
parentPort.postMessage({ workerIndex, processed });
主线程 src/core/work-stealing.ts 建 4 条 deque、按段压入 96 个任务，对比「无窃取」与「有窃取」两种模式的 makespan 与每个 worker 的处理数。
// file: src/core/work-stealing.ts
// tier: L1
// verify: node src/core/work-stealing.ts
//
// helio 内核：work stealing 实测。4 个 worker、96 个任务，任务代价呈阶梯分布
// （4 段各 24 个，单价 1/2/4/8），初始按段静态分给 4 个 worker。
// 对比「无窃取」与「有窃取」的墙钟 makespan 与每个 worker 的处理数。
import { Worker } from 'node:worker_threads';
import { performance } from 'node:perf_hooks';
import { push } from './deque.ts';
const P = 4;
const CAP = 128;
const BLOCKS = 4;
const PER_BLOCK = 24;
const M = BLOCKS * PER_BLOCK; // 96
const DEQUE_BASE = 4;
const RESULTS_BASE = DEQUE_BASE + P * (2 + CAP);
/** 任务 id 的代价：第 b 段（24 个）单价 2**b。 */
function costOf(id: number): number {
return 2 ** Math.floor(id / PER_BLOCK);
}
async function run(stealing: boolean): Promise<{
ms: number; perWorker: number[]; results: number[]; costUnits: number;
}> {
const sab = new SharedArrayBuffer((RESULTS_BASE + M) * 4);
const words = new Int32Array(sab);
const costs = Array.from({ length: M }, (_, id) => costOf(id));
Atomics.store(words, 3, P);
// 初始分布：把第 b 段的 24 个任务压进 worker b 的 deque
for (let b = 0; b < P; b++) {
for (let k = 0; k < PER_BLOCK; k++) {
push(words, DEQUE_BASE, CAP, b, b * PER_BLOCK + k);
}
}
const workers = Array.from({ length: P }, (_, i) =>
new Worker(new URL('./steal-worker.mjs', import.meta.url), {
workerData: { sab, P, cap: CAP, workerIndex: i, costs, stealing },
}),
);
// 等全部就绪，再放行并计时
while (Atomics.load(words, 0) < P) await new Promise((r) => setTimeout(r, 1));
const t0 = performance.now();
Atomics.store(words, 1, 1);
Atomics.notify(words, 1, P);
while (Atomics.load(words, 2) < P) await new Promise((r) => setTimeout(r, 1));
const ms = performance.now() - t0;
const results: number[] = [];
for (let i = 0; i < M; i++) results.push(Atomics.load(words, RESULTS_BASE + i));
const perWorker = Array.from({ length: P }, () => 0);
for (const owner of results) if (owner > 0) perWorker[owner - 1]++;
await Promise.all(workers.map((w) => w.terminate()));
return { ms, perWorker, results, costUnits: costs.reduce((a, b) => a + b, 0) };
}
function check(tag: string, r: Awaited<ReturnType<typeof run>>): void {
const done = r.results.filter((x) => x > 0).length;
console.log(
`${tag}: makespan ${r.ms.toFixed(1)} ms | 处理数 ${r.perWorker.join('/')} ` +
`| 覆盖 ${done}/${M}`,
);
}
console.log(`worker=${P} 任务=${M} 总代价=${costOf(0) * 0 + 24 * (1 + 2 + 4 + 8)} 单位`);
const staticRun = await run(false);
check('静态分片(无窃取)', staticRun);
const stealRun = await run(true);
check('work stealing ', stealRun);
console.log(
`加速比 ${(staticRun.ms / stealRun.ms).toFixed(2)}x（makespan ${staticRun.ms.toFixed(1)} -> ${stealRun.ms.toFixed(1)} ms）`,
);
正确性：单线程用例
work stealing 最怕的是「丢任务」或「重复执行」。src/core/deque.test.ts 先用单线程验证两端语义：pop 是 LIFO，steal 是 FIFO，空队列返回 EMPTY。
// file: src/core/deque.test.ts
// tier: L1
// verify: npx vitest run src/core/deque.test.ts
import { describe, expect, it } from 'vitest';
import { EMPTY, pop, push, steal } from './deque.ts';
const BASE = 4;
const CAP = 16;
function make(): Int32Array {
return new Int32Array(BASE + 2 * (2 + CAP));
}
describe('Chase-Lev deque', () => {
it('push 后 pop 是 LIFO（后进先出）', () => {
const w = make();
push(w, BASE, CAP, 0, 10);
push(w, BASE, CAP, 0, 20);
expect(pop(w, BASE, CAP, 0)).toBe(20);
expect(pop(w, BASE, CAP, 0)).toBe(10);
expect(pop(w, BASE, CAP, 0)).toBe(EMPTY);
});
it('窃取从 top 端取，是最早入队的那个（FIFO）', () => {
const w = make();
push(w, BASE, CAP, 1, 1);
push(w, BASE, CAP, 1, 2);
push(w, BASE, CAP, 1, 3);
expect(steal(w, BASE, CAP, 1)).toBe(1);
expect(steal(w, BASE, CAP, 1)).toBe(2);
expect(pop(w, BASE, CAP, 1)).toBe(3);
expect(steal(w, BASE, CAP, 1)).toBe(EMPTY);
});
it('空 deque 上 pop / steal 返回 EMPTY', () => {
const w = make();
expect(pop(w, BASE, CAP, 0)).toBe(EMPTY);
expect(steal(w, BASE, CAP, 0)).toBe(EMPTY);
});
});
本机 npx vitest run src/core/deque.test.ts 三条全过。真正跑并发时还有一道保险：每个 worker 处理完任务会把「谁处理的」写回共享结果数组，主线程最后统计覆盖数——必须是 96/96，一个不多一个不少。
实测：不均衡负载下，静态分片 vs work stealing
worker=4 任务=96 总代价=360 单位
静态分片(无窃取): makespan 119.5 ms | 处理数 24/24/24/24 | 覆盖 96/96
work stealing : makespan 60.5 ms | 处理数 39/27/19/11 | 覆盖 96/96
加速比 1.98x（makespan 119.5 -> 60.5 ms）
两组实验的负载完全一样：96 个任务分成 4 段，单价 1/2/4/8，初始按段静态分给 4 个 worker。也就是说 worker 3 一开始背着的活是 worker 0 的 8 倍。
- 静态分片：每个 worker 只吃自己那段，处理数钉死在 24/24/24/24，谁先干完谁闲着。makespan 由最慢的 worker 3 决定，119.5 ms。
- work stealing：worker 0/1/2 早早清空自己的队列，转头去偷 worker 3 的 top 端。处理数变成 39/27/19/11——worker 3 从 24 个减到 11 个，活被分走了。makespan 掉到 60.5 ms。
同机多跑几次，加速比稳定落在 1.8～2.1 倍。总代价 360 单位、4 核的理想 makespan 是 90 单位；静态分片卡在 192 单位（worker 3 那段），窃取后逼近理想值。这就是 work stealing 的全部意义：把「谁分到多少」从一次静态预算，变成一个运行时会自己收敛的过程。
什么时候窃取才有用
work stealing 不是免费的，代价体现在几处：
- 窃取本身有开销。 每次
steal要在对方的 deque 上做一次 CAS，失败还得重试；所以它只在任务不均、且真有 worker 空闲时才划算。 - 负载越不均衡，收益越大。 上面那组是 8:1 的极端不均才有 1.8 倍；如果初始分布本来就均匀，窃取只会带来纯开销。
- 粒度仍然要够粗。 任务太碎时，14.2 的教训同样成立：窃取抢到的可能只是一个几微秒的任务，收益还不够覆盖 CAS 的成本。
- 共享写仍是禁区。 窃取解决的是「活怎么分」，不解决「两个任务写同一块内存」——后者要靠第 15 章的依赖 DAG 来排开。
到第 15 章，这套 deque 会被封装进一个任务图调度器：每帧把 ECS 系统构造成一张依赖 DAG，拓扑排序后把互不依赖的节点丢进 work stealing 队列，让 10 个核真正转起来。本节把底座打好了，15.1 就来搭那张图。
阅读导航：上一节：14.2 线程池、任务队列与 Atomics 同步 · 下一节：15.1 依赖 DAG 与拓扑执行 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。