Lecture 24: Graph Processing and Distributed Machine Learning — 图处理与分布式机器学习
Lecture 24: Graph Processing and Distributed Machine Learning — 图处理与分布式机器学习
讲义对应：CS 425 FA2026 Lecture 24。本章对应课程 Lecture 26「Graph Processing and Machine Learning」（2026-11-19，The New Age 模块），主要素材为课程讲义
L26.FA25.pdf（30 页：页 1-15 为 Distributed Graph Processing / Google Pregel，页 16-30 为 Machine Learning）。讲义明确指出这两部分都在考试范围内、且不可选（Graph Processing (in syllabus)、Machine Learning (in syllabus)；课程网站上的 Spark 视频亦标注为 In Syllabus, NOT optional），期末范围是 Lectures 1-29，因此本章是全课程收官阶段的必考内容。 前置章节：Lecture 3/5（Ch.5）MapReduce（L4.FA25.pdf/L3.FA26.pdf：map/reduce、shuffle、Map 与 Reduce 之间的屏障、HDFS 落盘、straggler 与 backup task）——本章的 BSP 屏障、”重算代替恢复”、straggler 处理都直接沿用那里的概念；Lecture 26-B（Ch.24 补充）Spark（L26.B.FA25.pdf：RDD、lineage 容错、GraphX 的 Gather-Apply-Scatter）——Spark 讲义自己就把 Pregel 列为”迭代式 MapReduce”这一动机的产物；Lecture 19/20（Ch.17/Ch.20）Paxos 与复制控制——参数服务器副本一致性的背景；Lecture 5（Ch.7）故障检测——Pregel 的 ping 心跳。 教材对应：Coulouris 5th Ed. 分布式计算范式与云章节（Ch. 1-2 关于大规模数据处理的讨论）；补充：Malewicz et al. Pregel: A System for Large-Scale Graph Processing（SIGMOD 2010，课程指定阅读）、Dean et al. Large Scale Distributed Deep Networks（NeurIPS 2012，讲义页 21 引用）、Abadi et al. TensorFlow: A System for Large-Scale Machine Learning（OSDI 2016，讲义页 22-24 引用）。 阅读材料：课程指定 Pregel 论文（本章以讲义口径为准；讲义未给出的论文级细节均标注”补充说明”）；可选：Valiant, A Bridging Model for Parallel Computation（1990，BSP 模型原始论文）、Gonzalez et al. PowerGraph（OSDI 2012，vertex-cut）、Li et al. Parameter Server（OSDI 2014）、McMahan et al. Federated Learning / FedAvg（AISTATS 2017）。
24.1 概述
本章回答两个在 2010 年之后迅速合流的问题：（1）当图大到一台机器装不下时，如何用一整个集群去”想”这张图？（2）当模型和数据大到一台机器装不下时，如何用一整个集群去”训练”它？ 第一个问题的答案是 Google 的 Pregel：它把 Valiant 1990 年提出的 BSP（Bulk Synchronous Parallel，整体同步并行） 模型搬到图上，提出”像顶点一样思考（think like a vertex）“的编程范式——程序员只需写”一个顶点收到消息后该做什么”，全局的并行、通信、同步与容错全部由系统负责。第二个问题的答案是分布式机器学习（Distributed Machine Learning）：以参数服务器（Parameter Server）或 AllReduce 为通信骨架，把训练在数据并行（Data Parallelism）、模型并行（Model Parallelism）两个维度上切开，并在”同步“与”异步“之间做权衡。
为什么这两件事要放在同一讲？因为它们共享同一个核心矛盾：计算单元之间必须频繁交互——图处理中交互发生在邻居顶点之间（消息），机器学习中交互发生在梯度之间（聚合）。交互就意味着通信与同步，而一旦有同步就会遇到最慢的参与者拖累所有人（straggler），一旦取消同步就会失去确定性。因此本章反复出现的三条主线是：BSP 屏障给出确定性与简单性（代价是 straggler 与收敛慢）、异步执行给出速度（代价是确定性与收敛性的损失）、而 SSP（有界延迟） 与 Local SGD 都是在两端之间找折中的工程答案。
本讲在整门课中的位置：Lecture 3/5 的 MapReduce 给出了”分而治之 + 屏障 + 重算容错“的批量计算骨架；Lecture 5 的 GFS/HDFS 给出了持久化存储；Lecture 6-8 的 gossip 与 Chord 给出了哈希分区（Pregel 的默认分区策略就是 hash(vertexID) mod N，讲义页 10 明确回指 “Remember consistent hashing from P2P systems?!”）；Lecture 7 的故障检测给出了 ping 心跳；本章把这些零件组装成两类迭代式系统，并第一次把”机器学习的训练过程”本身当作一个分布式系统问题来研究——这也是 2012 年之后工业界最大的分布式系统战场。
24.2 核心概念与分布式机制图解
24.2.1 为什么图处理需要专门的系统（Why Graph Processing Needs Its Own System）
定义与目的：图处理（Graph Processing）指的是在一张图 $G=(V,E)$ 上反复执行”用邻居的值更新自己的值“这类操作，直到满足终止条件。讲义页 6 把它抽象为图算法的典型结构（Gather-Apply-Scatter）：每个顶点持有一个值；每一轮迭代中，每个顶点 (1) Gather：从直接邻居收集值（例如 $B\!\to\!A, C\!\to\!A, D\!\to\!A$）；(2) Apply：用自己的旧值和邻居的值做计算；(3) Scatter：更新自己的值并发送给邻居（例如 $A\!\to\!B,C,D,E$）。迭代在固定轮数后或顶点值不再变化时结束。
图数据的普遍性（讲义页 3-4 的原口径）：Internet 图（顶点是路由器/交换机，边是链路）、World Wide Web（顶点是网页，边是网页上的 URL 链接，因为边是单向的所以叫有向图）、社交图（Facebook、Twitter、LinkedIn）、生物图（脑神经元、DNA 相互作用图、生态系统图）等等。我们需要从这些图里推导性质、汇总统计：例如求顶点对之间的最短路（Internet 用于路由，LinkedIn 用于”几度分隔”）、做匹配（matching）（match.com 的约会图）、以及 PageRank（Web 图；Google、Bing、Yahoo 的搜索都依赖它）。
为什么难（讲义页 5）：图的规模太大——人类社交网络有数亿顶点、数十亿条边（讲义口径）；WWW 有数百万顶点与边。在单台服务器上存储与处理整张图很困难：单机要么很慢，要么非常贵（性能价格比极低）。因此必须使用分布式集群/云。
- 为什么 MapReduce 不适合图算法（必须讲透的四点）：讲义页 7 直接给出了结论——”Multi-stage Hadoop：每个 stage 等于一次图迭代；在 reduce 阶段把顶点 ID 当作 key；每个 stage 结束时要把所有顶点通过网络传给邻居顶点；所有顶点值都要写进 HDFS；非常慢（Very slow!）“。把这句话拆开，是四个结构性错配：
- 图算法天生迭代（PageRank 迭代到收敛、最短路反复松弛、连通分量不断传播），而 MapReduce 的每次迭代都是一个独立的 job：Map 输出落本地盘、Reduce 输出落 HDFS、job 之间靠调度器重新分配任务。若算法需要 30 轮迭代，就要重启 30 次 job、把整张图往返读写 30 次磁盘。I/O 与调度开销（秒级到分钟级）远大于每轮真正的计算（往往是毫秒级）。
- 访问模式不规则：真实图的度数服从幂律分布（power law）——绝大多数顶点只有几条边，少数顶点有上百万条边。MapReduce 的 map/reduce 任务按数据块静态切分，无法按”度数”切分，于是负载极度倾斜：一个任务要处理百万条边，其他任务几秒钟就跑完，剩下的时间全在等它（这又回到 Lecture 5 的 straggler 问题）。
- 收敛条件是动态的：有的顶点第 3 轮就不再变化，有的第 50 轮还在变。MapReduce 的每个 job 都是全量重算（all vertices），无法表达”只重算还没收敛的那部分顶点”。真正高效的图系统需要活跃顶点集合（active set）这一概念。
- 图分割（graph partitioning）困难：要把图放到 $N$ 台机器上，理想目标是最小化跨机器的边（edge cut），因为每条跨机边都意味着一次网络消息。而最小割图分割本身是 NP-hard 的（讲义页 10 的口径更务实：hash 分区简单但可能不均；基于局部性的分区能减少机器间通信量，但”一些聪明的基于局部性的方案可能花掉大量前期时间，却带不来足够收益”）。
图处理系统的设计目标：由此推出五条设计目标，它们正是 Pregel 的设计清单：(1) 原生支持迭代（不要让每次迭代都重启作业）；(2) 让图的拓扑常驻内存，避免反复落盘（这与 Lecture 26-B 中 Spark 用 RDD 缓存中间结果、避免 MapReduce “expensive save to disk for fault tolerance” 是同一个动机）；(3) 容错（集群规模到几百上千台，故障是常态）；(4) 容忍负载倾斜；(5) 可扩展（顶点与边的规模、以及机器数目两个方向）。
- 并行图处理的抽象层级（四种”以谁为中心”的抽象）：
| 抽象层级 | 代表系统 | 程序员写什么 | 优点 | 缺点 |
|---|---|---|---|---|
| 顶点为中心（vertex-centric） | Pregel、Giraph、GraphX、Piccolo | 一个顶点的 Compute() | 直观（”像顶点一样思考”）、天然表达 BFS/PageRank/连通分量 | 高 degree 顶点成为热点；同步屏障代价高 |
| 边为中心（edge-centric） | X-Stream、Chaos、Scatter-Gather | 一条边上的 scatter/gather | 顺序访问边数组，外存友好（顺序 I/O） | 需要按边反复扫描，随机度低但工作量大 |
| 块/分片为中心（block-centric） | Giraph Unchained、Blogel | 一个块内部的子图算法 | 块内可以同步、块间异步，减少消息 | 编程模型更复杂，需要块划分 |
| 矩阵为基础（matrix-based） | GraphBLAS、线性代数视角 | 稀疏矩阵-向量乘 | 复用成熟的高性能稀疏线性代数库 | 表达”动态收敛/活跃集”不自然 |
关于”矩阵视角”要补一句直觉：PageRank 的迭代 $\text{PR}^{(k+1)} = \frac{1-d}{N}\mathbf{1} + d\,A^{\top}D^{-1}\text{PR}^{(k)}$ 本质上就是一次稀疏矩阵-向量乘（SpMV）加一次线性组合，”顶点为中心”与”矩阵为基础”描述的是同一件事的不同写法；这也解释了为什么 GNN 与图计算可以共用一套运行时。
- 关键假设与系统模型（本章图处理部分的公共假设）：集群由 $N$ 台同构机器组成（worker），机器之间只通过消息通信，没有共享内存；机器可能崩溃（crash-stop / crash-recovery），模型是fail-stop 而非拜占庭（对应 Lecture 17 的故障模型分类）；网络是可靠但延迟有界但未知的（部分同步）；持久状态放在分布式存储（GFS/BigTable/HDFS）上，临时状态放本地磁盘（讲义页 11 的原话）。
24.2.2 BSP 与超级步：Pregel 的执行模型（Bulk Synchronous Parallel and Supersteps）
定义与目的：BSP（整体同步并行）由 Valiant 于 1990 年提出（讲义页 8 明确标注 “Originally by Valiant (1990)”），是 Pregel 的执行骨架。系统把时间切成一段段超级步（superstep），每个超级步由三个阶段组成，阶段之间由一个全局屏障（barrier）分开：(1) Receive（接收）：顶点读取上一个超级步投递给它的消息；(2) Compute（计算）：顶点执行用户定义的
Compute()，它可以更新自己的值、向其他顶点发消息、或投票停机（vote to halt）；(3) Send（发送）：消息被投递给目标顶点。讲义页 12 给出了 Pregel 的执行流程与关键时序约定：“消息可以随时发送，但必须在本次迭代结束（即屏障）之前投递完成”——也就是本章最常被考到的规则：第 $k$ 个超级步发出的消息，只能在第 $k+1$ 个超级步被收到。直观解释（”它是什么？”）：把每个顶点想成一个只能跟邻居说话的村民。每一轮里，所有村民同时开口：先听上一轮邻居寄来的信（Receive），再自己算一算（Compute），然后把新写的信塞进信封（Send）。这一轮里谁都不许提前拆看这一轮刚寄出的信——所有信统一在本轮结束时由邮局（系统）投递，下一轮才被拆开。等所有人都说完，村长喊一声”本轮结束”（屏障），大家才一起进入下一轮。这个”一起停一下再一起走”的纪律带来两个巨大好处：结果与调度顺序无关（确定性），以及任何一台机器的进度都能被系统精确地记账（便于容错）。
机制图解（本章最重要的图：BSP 超级步 + 屏障 + 消息在超级步末投递）：
每个超级步的三个阶段（每个 worker 都对自己分区里的【活跃顶点】做这三件事）
Receive : 读 inbox —— 也就是上一步（k-1）投递过来的消息
Compute : 调用用户函数 compute()，可以改值 / 发消息 / 投票停机 / 提供聚合值
Send : 消息进入 outbox，【本步不投递】，留到超级步结束时统一投递
超级步 k BARRIER 超级步 k+1
---------------------------- ------- -------------
W0 [v1 v4 v7] compute() -> outbox ---+
W1 [v2 v5 v8] compute() -> outbox ---+---> 所有 worker 到达屏障
W2 [v3 v6 v9] compute() -> outbox ---+ 屏障耗时 = 最慢 worker 的耗时
|
+---> outbox 在【超级步结束时】统一投递
目标顶点的 inbox 在 k+1 步被读取
时间轴 t_k -------------------------> t_k + max_w(load_w) -------------> t_{k+1}
规则 1（时序）：k 步发出的消息，最早只能在 k+1 步被接收
规则 2（屏障）：k+1 步开始时，k 步的全部计算与投递都已【确定完成】
规则 3（终止）：没有【活跃顶点】且没有【在途消息】时，整个作业结束
规则 4（容错）：屏障把所有 worker 的进度对齐到同一个超级步编号上，
因此"从检查点重放"有唯一确定的含义
屏障的代价与收益：屏障的代价是最慢的 worker 决定整个超级步的长度（讲义页 12 的”当所有 worker 都到达迭代屏障时，leader 才开始下一次迭代”，以及 Lecture 5 中 “The slowest task slows the entire job down” 的同一现象）；屏障的收益是确定性——因为每个顶点在超级步 $k$ 收到的消息集合只取决于超级步 $k-1$ 结束时各顶点的值，与消息在网络中的到达顺序、与 worker 内部的执行顺序都无关。
关键假设与系统模型：BSP 假设”超级步的时长可以不同，但屏障本身不出错”；故障被抽象为”某个 worker 在某个超级步内消失”，由系统在下一个检查点边界统一处理（见 24.2.6）。
24.2.3 顶点为中心的编程模型（Vertex-Centric Programming Model）
定义与目的：讲义页 8-9 用一句话概括 Pregel 的设计哲学：“Think like a vertex”（像顶点一样思考）。具体做法是：把每个顶点分配到一台服务器（因为一个顶点可能有上百万条边，所以顶点是最合适的分配单位，而不是边）；每台服务器因此拿到顶点的一个子集；在每个迭代（超级步）中，每台服务器对自己负责的顶点执行 Gather-Apply-Scatter。程序员只需要写
Compute()——”我这个顶点收到这些消息后应该做什么”，而不需要写如何并行、如何通信、如何同步、如何容错。直观解释（”它是什么？”）：写图算法通常有两种痛苦的写法：全局视角（”我要如何把整个算法切分到 500 台机器上并调度它们？”）与矩阵视角（”我要如何把算法写成一系列稀疏矩阵运算？”）。Pregel 提供第三种：局部视角——你只回答”一个顶点在收到邻居消息后应该做什么“。这就像写社会调查问卷：你不用设计”如何让 800 个调查员互不干扰”，你只需写”每个受访者拿到邻居的答案后，怎么改自己的答案”。全局行为是局部规则的涌现。
- 模型要素（这是考点级细节）：
- 顶点（Vertex）：有唯一 ID（字符串）、可变的用户定义值（例如 PageRank 值、到源点的距离）、以及出边邻接表。
- 边（Edge）：由目标顶点 ID 与可变的边值（例如权重）组成；顶点通过 ID 向任意顶点发消息（不要求邻居在同一台机器上）。
- 消息（Message）：顶点之间传递的用户定义值；Pregel 的默认做法是值传递（不共享内存指针），从而天然支持跨机器。
- 超级步（Superstep）：BSP 的同步单位，见 24.2.2。
- 活跃状态（active / halted）：顶点有”活跃”与”停机”两态，见下面的状态机。
- 机制图解（Pregel 的顶点状态机——”投票停机”与”被唤醒”）：
compute() 里调用 VoteToHalt()
+----------+ --------------------------------> +----------+
| ACTIVE | | HALTED |
+----------+ <-------------------------------- +----------+
收到任意一条消息 -> 系统【唤醒】它
（wake-up: halted -> active）
状态含义：
ACTIVE : 本超级步会被执行；可以发消息、改自己的值、提供聚合值
HALTED : 本超级步不被执行；但仍然【可以接收】消息
终止判定：某一时刻【所有顶点都已停机】且【系统中没有在途消息】=> 全局终止
陷阱提醒：只"我自己的值不变了"就停机是不够的——邻居之后可能纠正你（见 24.3.4 的 SSSP）
终止条件（必须精确）：讲义页 12 的原话是 “Computation halts when, in some iteration: no vertices are active and when no messages are in transit“——当某一轮中没有任何活跃顶点、且没有任何在途消息时，计算停止。两个条件缺一不可：只判断”没有活跃顶点”是不够的，因为停机顶点的消息可能还在网络上；只判断”没有在途消息”也是不够的，因为有些顶点可能仍在自我迭代。
投票停机与唤醒（Pregel 终止机制的精妙之处）：顶点在
Compute()中主动投票停机后就不再被系统执行。但如果它在后续超级步收到了新消息，系统必须把它唤醒（重新变为活跃）。为什么？因为”我现在算出来的值不再变化”并不等于”我最终的值已经确定”——邻居稍后可能算出一个更小的距离/更大的权重，此时我必须被重新激活去修正自己。正是这个机制让 Pregel 能表达 Bellman-Ford 式的多轮松弛、以及”局部收敛但全局未收敛”的算法（24.3.4 会打印真实日志演示这一点）。关键假设与系统模型：
Compute()是确定性函数（同样的输入给出同样的输出）——这是 Pregel 能保证确定性的前提；用户函数不能读取未定义的外部状态（例如随机数、本地时间、未同步的文件），否则”相同的输入产生相同输出”这一前提就被破坏。
24.2.4 顶点分配与图分割：edge-cut vs vertex-cut（Partitioning）
定义与目的：分区（partitioning）决定哪个顶点属于哪台机器。讲义页 10 给了两类做法：基于哈希（Hash-based）——
Hash(vertex id) modulo number of servers（并回指 P2P 系统的一致性哈希）；基于局部性（Locality-based）——把邻接关系密集的顶点尽量放到同一台服务器上，减少每次迭代后服务器之间的通信量。讲义同时警告：“一些’聪明’的基于局部性的方案可能占用大量前期时间，却带不来足够的收益”——分区是一次性成本（启动时）换取每一轮的收益，所以要算清楚：如果只能跑 5 轮，复杂的图分割可能得不偿失。直观解释（”它是什么？”）：把集群想成几间办公室，顶点是需要频繁打电话的人。Edge-cut（切边）的做法是”按人分“——每个人只属于一间办公室，但他的朋友可能分布在别的办公室，于是跨办公室的电话（跨机边）就是通信开销；Vertex-cut（切点）的做法是”按关系分“——某个社交达人的关系被复制到多间办公室（他被”切开”），这样他每次只需跟同办公室的人讲话，但代价是他的状态需要在多份副本之间同步。
机制图解（edge-cut vs vertex-cut，以及它们对幂律图的影响）：
Edge-cut（切边，Pregel / Giraph / GraphX 的默认做法）
规则：每个顶点只属于一台机器；把边切开（edge cut）
machine A | machine B
[A]---[B]---[C] | [D]---[E]
\ | |
\-- cross-machine ---|-------------+ （跨机边 = 一条消息 = 一次网络传输）
* 幂律图的问题：顶点 H 有 10^6 条边，H 落到哪台机器，
哪台机器就背 10^6 条边 + 每轮 10^6 条消息 -> 负载倾斜 + 通信热点
Vertex-cut（切点，PowerGraph 的默认做法）
规则：把【高 degree 顶点】切开，复制到多台机器；边不跨机器（edge 归属确定）
machine A | machine B
[A]---[B]---[H_A] | [D]---[E]---[H_B]
\ | /
+---- sync the STATE of H only, not its edges ----+
（只同步 H 的状态，不同步它的边）
* 优势：把"一个超级明星顶点"的边负载摊到多台机器上，
让每台机器处理大致相同的边数 -> 解决幂律图（power-law）的倾斜
* 代价：H 的值被复制成多份，需要额外的一致性维护（master/mirror 机制）
为什么幂律分布是关键：真实图（Web、社交、引文图）中顶点度数近似服从幂律 $P(d)\propto d^{-\gamma}$（$\gamma$ 常在 2-3 之间）。于是平均度数很小，但最大度数极大：顶点数只有几亿，最大 degree 却可达数百万。Edge-cut 在数学上就无法把这样的图均分（想象一个 100 万度的星形顶点，无论放哪台机器，那台机器都要处理 100 万条边），而 vertex-cut 通过”复制超级顶点、摊开它的边”来达成均衡——这是 PowerGraph 相对 Pregel 的核心改进（补充说明：讲义只列出 PowerGraph 的名字，未展开其机制；本节按 PowerGraph 论文口径补充，供理解”为什么后续系统要改分割方式”）。
关键假设与系统模型：分区在作业启动时完成（静态分区），运行期不变（除非发生故障重分配）；默认哈希分区是无状态的，因此任何 worker 都能算出任意顶点 ID 的归属——这正是”顶点可以给任意 ID 发消息”这一 API 能成立的原因。
24.2.5 Combine 与 Aggregator：两种”减少通信”的机制
定义与目的：组合器（Combiner）是针对同一个目标顶点的多条消息在本地先合并（用用户提供的
combine()），从而减少网络传输量；聚合器（Aggregator）是全局归约（例如全局最大值、全局求和、全局计数），让顶点能读到”整个图的汇总信息”，用于全局协调（判断是否收敛、动态调整行为）。- 直观解释（”它是什么？”）：
- Combiner 就像宿舍楼里的快递代收：同寝室 4 个人的包裹不必各派一辆车送到宿舍楼下，而是在校门口先合并成一个大包再送（这正是 Lecture 5 里 MapReduce combiner 的同一思想：Map 端本地预聚合，减少 shuffle 数据量）。所以在 Pregel 中，向同一个顶点发送”和/最大值”这类可合并的消息，能被压缩成一条。
- Aggregator 就像班级群里的投票统计：每个人把自己的选项交给班长，班长在下一轮公布”全班最高分/总人数”，于是每个人都能基于全局信息决定下一步（例如”全班最大残差已小于 $10^{-3}$，可以收工了”）。
- 机制图解（Combine 的本地合并 + Aggregator 的超级步边界传递）：
超级步 k：多个顶点都把消息发给同一个目标 T（T 在 machine 2）
machine 1: v5 --msg 0.31--> T raw messages = 4
v9 --msg 0.12--> T
(combine = sum) ===> 合并成一条 0.43 ------------------+
|
machine 2: v7 --msg 0.05--> T v
v8 --msg 0.09--> T ===> 合并成一条 0.14 -------> T 的 inbox: [0.43, 0.14]
delivered messages = 2
要求：combine() 必须满足【交换律 + 结合律】（本例 sum/min/max 都满足），
否则"先合并还是后合并"会改变结果 -> 破坏确定性。
Aggregator：超级步 k 中被提供的值，在超级步 k+1 可读
k 步: v1 提供 0.014 | v2 提供 0.003 | ... --> 系统归约 --> max = 0.014
k+1 步: 每个顶点调用 GetAggregatedValue("max_delta") 得到 0.014
-> 若 < eps 则 VoteToHalt()（全局一致的停机判据！）
为什么 Aggregator 对”正确收敛”至关重要：第 24.4 节的实验会给出一个真实的反例：如果每个 PageRank 顶点只根据”我自己的变化量小”就停机（局部判据），那么它停机后不再向邻居发送自己那份权重，邻居的求和就少了一项，整个图的权重总量会泄漏（实验里那个「局部停机」版本跑到 200 个超级步仍未收敛，$\sum \text{PR} = 0.813566 \ne 1$，权重泄漏 18.6%）；而用 Aggregator 做全局最大残差判据时，所有顶点读到同一个全局值、在同一个超级步一起停机，$\sum \text{PR} = 1.000000$，与 200 轮 Jacobi 精确解的最大误差仅 $1.24\times10^{-4}$。
关键假设与系统模型：Combine 是每个 worker 本地的优化（只能合并”落在同一台机器上、同一超级步、同一个目标”的消息）；Aggregator 是全局的，需要所有 worker 参与归约，因此天然与屏障绑定（这也是它便宜的原因：屏障本来就要同步一次）。
24.2.6 Pregel 的容错：检查点 + 重算（Fault Tolerance by Checkpointing）
定义与目的：Pregel 采用 leader/worker（Master/Worker） 结构（讲义页 11）：leader（一台服务器）维护 worker 列表、监控 worker 并在故障时重启它们、提供 Web-UI 监控作业进度；worker 处理自己那份顶点、与其他 worker 通信；持久数据存在分布式存储（GFS/BigTable）上，临时数据存在本地磁盘。容错机制分三步（讲义页 13 原口径）：
- 检查点（Checkpointing）：周期性地（在某个迭代开始时）由 leader 指示所有 worker 把自己分区的状态保存到持久存储——例如顶点值、边值、收到的消息。
- 故障检测（Failure detection）：leader 用周期性 ping 消息（leader $\to$ worker）判断 worker 是否存活（与 Lecture 5/7 的失败检测器同一思路：超时即怀疑）。
- 恢复（Recovery）：leader 把图分区重新分配给当前可用的 worker，所有 worker 从最近一次可用的检查点重新加载分区状态，然后从该检查点对应的超级步继续。
直观解释（”它是什么？”）：检查点像游戏存档。存档太频繁，每次存档本身很慢（要写几百 GB）；存档太稀疏，一旦挂了就要从很久以前重打。恢复的哲学是”用重算代替恢复状态“——这一点与 Lecture 5 中 MapReduce 的作法完全一致（MapReduce 也是扔掉失败的 task，重新执行它，而不是去恢复它的中间状态）。Pregel 只需要”回到上一个存档”，因为 BSP 的屏障保证了执行的可重放性：所有 worker 都停在同一个超级步边界上，从那里重放必然得到相同结果（24.3.1 会证明确定性）。
机制图解（worker 失败 → 分区重分配 → 从检查点重算）：
超级步: 4 5 6 7 8
worker0: [====] [====] [====] [====] [====]
worker1: [====] [====] [ X ] <-- worker1 在第 5 步崩溃（ping 超时）
worker2: [====] [====] [====] [====] [====]
^ ^
| |
checkpoint@4 故障点（第 5 步）
1) leader 检测到 worker1 无响应（ping 超时）
2) leader 把 worker1 的分区【重新分配】给存活的 worker0 / worker2（并均分负载）
3) 所有 worker 回滚到 checkpoint@4，重新执行超级步 4、5（重算 2 个超级步）
4) 新 worker 从第 6 步继续 —— 最终结果与无故障时【逐位相同】（BSP 确定性）
代价模型：故障恢复成本 ≈ (检查点间隔 N) x (每个超级步的成本) + 分区迁移成本
检查点间隔 N 越小 -> 重算越少，但检查点开销越大（存在最优 N）
confined recovery（补充说明，论文口径）：只重算"丢失分区所需的那些消息"，
而不是让全系统回滚，可进一步降低恢复成本
- 检查点频率的权衡（本章要求做表对比）：
| 检查点间隔 $N$ | 检查点开销（正常路径） | 故障后重算量 | 适用场景 |
|---|---|---|---|
| $N$ 很小（如 1） | 每个超级步都要写全量分区状态到持久存储，开销可能超过计算本身 | 最多 1 个超级步 | 集群不稳定、单步计算极快、状态很小 |
| $N$ 中等（如 5-10） | 均摊后约 $(1/N)$ 的额外 I/O | 最多 $N$ 个超级步 | 通常的工程折中，也是 Pregel 的默认策略 |
| $N$ 很大 | 几乎无额外开销 | 故障后重算很多超级步（可能比重新跑一遍还慢） | 集群非常稳定、单步很贵、状态很大 |
补充说明：讲义页 13 明确列出的检查点内容包括”顶点值、边值、收到的消息“三样；把”在途/已收消息“也纳入检查点是必要的，因为一个顶点在超级步 $k+1$ 的行为取决于它在 $k$ 步末收到的消息，如果只存顶点值不存消息，重放就会丢失”唤醒”事件，结果将与无故障时不同。第 24.4 节的实验正是把这三样（顶点值、边表、收件箱）连同活跃标志一起存档，才得到了与无故障运行逐位一致的恢复结果。
关键假设与系统模型：故障模型是 fail-stop / crash-recovery（进程崩溃后不发送错误消息，重启后从持久存储加载）；不处理拜占庭故障（对应 Lecture 17：拜占庭需要 $3f+1$ 才可行，而图处理系统的前提是”机器不会说谎”）。leader 自身故障的处理在讲义中未展开（补充说明：论文口径是”leader 也周期性检查点自己的状态，由其他 worker 接管”，本章不展开）。
24.2.7 同步的代价与异步图处理（Asynchronous Graph Processing）
定义与目的：异步执行（asynchronous execution）指顶点一旦被激活就立即计算并立即把新值传播给邻居，不等待任何全局屏障。它是对 BSP 屏障代价的直接回应。
- 同步 BSP 的三项代价：
- straggler 拖累所有人：超级步的时长等于最慢 worker 的时长，只要有一个顶点（例如某个百万度的 hub，或某台被其他作业干扰的机器）慢下来，所有机器都在屏障处空等。
- 收敛慢的顶点拖住整个超级步：只要有一个顶点还在变，全局就无法进入下一轮；在幂律图上，”尾巴顶点”会让收敛轮数被最慢的那条链拉长。
- 消息量随超级步线性增长：每轮所有活跃顶点都要发消息，总消息数 $\approx$ 轮数 $\times$ 活跃顶点数（第 24.4 节的实验里，7 个顶点 12 轮就产生 132 条消息）。
- 异步执行的优点与代价：
- 优点：收敛更快。因为更新立即传播——这在数值上等价于Gauss-Seidel 迭代，而同步 BSP 是 Jacobi 迭代。直觉上：Jacobi 中”第 10 跳的信息”必须等 10 个超级步才能传出去；Gauss-Seidel 中它沿着一条链一次扫描就传到底。研究文献（GraphLab）报告异步可达 10-100× 加速（补充说明：这一数字来自 GraphLab 系列论文，不在课程讲义中；第 24.4 节的实验中异步在直径较大的小世界图上取得了 2.25× 的时间与计算量双降，与”传播速度决定收敛”这一机理一致）。
- 缺点：失去确定性。结果可能因调度顺序（FIFO/LIFO/优先级）而不同：实验中同一张图、同一算法，FIFO 调度用 800 次顶点重算收敛，LIFO 调度却用了 21340 次（慢 27 倍），最终数值也有 $5.75\times10^{-5}$ 的差异。
- 缺点：需要串行化机制。顶点可能被并发更新，因此需要顶点级锁、或按颜色/分区的无锁调度（相邻顶点不同时更新）、或用细粒度原子操作保证单顶点更新的原子性。
- 缺点：调试困难、容错更难。因为不存在”所有机器都对齐在同一个超级步”的时刻，无法简单地”回滚到检查点再重放”——异步系统必须靠日志（logging）、周期性一致性快照（如 Chandy-Lamport 式的分布式快照，见 Lecture 13）或Lineage 重算（见 Lecture 26-B 的 RDD）来恢复。
- 机制图解（同步 vs 异步：三台 worker 的时间线）：
同步 BSP（每轮都在屏障处等最慢者）
W0: |==k0==| |==k1==| |==k2==|
W1: |==k0==| |==k1==| |==k2==|
W2: |====k0====| |====k1====| |====k2====| <- 慢 (straggler)
^-------------------^-------------------^
屏障：每轮耗时 = max(三个 worker 的耗时) -> straggler 被【乘以轮数】
异步（谁先算完谁先传播，无人等待）
W0: |=k0=|=k1=|=k2=|=k3=|=k4=|=k5=|
W1: |=k0=|=k1=|=k2=|=k3=|=k4=|
W2: |==k0==|==k1==|==k2==|==k3==| (慢，但只慢自己)
^ 更新立即可见 -> 信息传播快；但没有全局"同一时刻"，结果依赖调度顺序
有界异步 SSP（staleness 上界 s=2；用于分布式 ML，见 24.2.11）
W0: |=k0=|=k1=|=k2=|==WAIT==|=k3=|=k4=|
W1: |=k0=|=k1=|=k2=|=k3=|=k4=|=k5=| 领先超过 s 个版本 -> 必须停下等待
W2: |==k0==|==k1==|==k2==|==k3==|==k4==|
混合方案（hybrid）：同步/异步可切换（同一系统提供两种模式，按图结构选择）；分层同步（局部同步 + 全局异步，例如块内同步、块间异步——Blogel 的 block-centric 模型正是这个思想）；以及按连通性分区后分区内用同步、分区间用异步。
- 收敛加速的通用思想（必须总结的四条）：
- 优先级调度：先算”重要”的顶点。例如 PageRank 的残差优先（residual-based prioritization）——总是更新”当前值与正确值差距最大”的顶点；SSSP 用优先队列（Dijkstra 风格）总是先扩展距离最小的顶点。这本质上是把”同步的全量迭代”换成”按影响排序的增量迭代”。
- Delta-based 计算：只传播变化的增量（$\Delta$），而不是每个顶点都重算并重发。第 24.4 节的异步引擎中加入了
delta > 1e-7才继续传播的剪枝，这是异步能在 480 次重算内（同步需要 1080 次）收敛的关键之一。 - 消息组合（Combiner）：见 24.2.5，把”同一目标的多条消息”在本地合并。
- 剪枝（pruning）：明确”什么时候可以停止传播”。例如 SSSP 中”只有当距离严格变小时才传播”；PageRank 中”残差小于阈值就停止”；连通分量中”只有拿到更小的分量 ID 才传播”。剪枝是活性层面最重要的技巧（见 24.3.4 的正确性论证）。
- 同步 vs 异步全面对比：
| 维度 | 同步 BSP | 异步 |
|---|---|---|
| 确定性 | 确定：给定输入与用户函数，第 $k$ 步结束的全局状态唯一（24.3.1 证明） | 不确定：取决于调度顺序与消息到达顺序 |
| 收敛速度 | Jacobi 迭代：信息每轮只能走一跳；轮数受直径/最慢链限制 | Gauss-Seidel 迭代：更新立即传播；通常轮数更少（实验：2.25×） |
| straggler 敏感度 | 高：(cost-1) × 超级步数 | 低：只影响该顶点自身的重算份额（实验：同步 ×4.00 vs 异步 ×1.23） |
| 容错 | 容易：检查点 + 从超级步边界重放即可（确定性重放） | 难：需要日志或一致性快照；重放结果可能与原来不同 |
| 实现复杂度 | 低（屏障由系统统一实现） | 高（锁/无锁调度、消息去重、竞态调试） |
| 内存/通信 | 消息在超级步末批量发送，便于组合与批处理 | 消息即时发送，难以批量，但可被 delta 剪枝 |
| 适合的算法 | 需要确定性与可复现性的算法（PageRank、迭代数值计算、需要可重复实验的评测） | 图直径大、幂律倾斜严重、机器异构（straggler 常见）的算法 |
24.2.8 Pregel 家族系统对比（The Pregel Family）
讲义页 2 与页 15 列出了 Pregel 启发的后续系统：Piccolo、Giraph、GraphLab、PowerGraph、LFGraph、X-Stream，并把它们分成”Pregel-like“与”more advanced“两类。下表按统一维度对比（补充说明：除 Pregel 与 Giraph 的定位来自讲义外，其余系统的机制细节来自各自论文，用于建立”从同步到异步、从 edge-cut 到 vertex-cut、从内存到外存”的演化脉络）：
| 系统 | 执行模型 | 抽象 | 分割策略 | 容错方式 | 内存/外存 | 代表应用 |
|---|---|---|---|---|---|---|
| Google Pregel | 同步 BSP（超级步 + 屏障） | 顶点为中心（Compute()） | hash 分区 / 可定制（edge-cut） | 周期性检查点 + 分区重分配 + 重算 | 内存为主，临时数据落本地盘，持久数据进 GFS/BigTable | PageRank、SSSP、连通分量 |
| Apache Giraph | 同步 BSP | 顶点为中心（Hadoop 上的开源 Pregel） | 默认 hash，可换自定义分割器 | 检查点（HDFS）+ 失败 worker 重算 | 内存 + HDFS | Facebook 的社交图分析（讲义提到 Giraph 属 Pregel-like） |
| Apache Spark GraphX | 同步 BSP（pregel() API，一轮 = 一个 RDD 迭代） | 顶点为中心 + 三元组（triplets）视图 | RDD 分区 + 用户可指定分区函数（减少 shuffle） | Lineage（血统）重算：丢失分区按依赖图重算，无失败则零成本 | 内存（RDD 缓存）+ HDFS | 与 Spark 生态统一：图算法 + SQL + 流处理（见 Lecture 26-B） |
| GraphLab / PowerGraph | 异步（也支持同步），Gather-Apply-Scatter | 顶点为中心（GAS） | PowerGraph：vertex-cut（解决幂律倾斜）；GraphLab 原版用 edge-cut | 分布式快照 / 检查点（异步下更难） | 内存 | 图机器学习（GraphLab 的定位就是 ML 友好）、幂律图 |
| GraphChi | 异步（单机） | 顶点为中心的 GAS | 分片（shard）+ 滑动窗口（sliding shard） | 单机：重启 + 中间文件 | 外存（磁盘）：让单机跑得下大图 | 单机大图分析（”用一块硬盘跑十亿边”） |
| Ligra / Galois / GAP | 同步/异步均可（共享内存多核） | 顶点或边为中心 | 共享内存，无需跨机分割 | 单机：进程重启 | 单机多核内存 | 共享内存上的高性能图算法（研究/基准测试） |
| X-Stream / Chaos | 异步/流式 | 边为中心（edge-centric） | 按边数组分区（顺序 I/O） | 中间结果落盘 | 外存友好 | 超大规模图（超过内存）的顺序扫描式处理 |
演化主线一句话：Pregel 用”同步 + edge-cut + 内存”换来了简单与确定；GraphLab/PowerGraph 用”异步 + vertex-cut”换来了速度与负载均衡；GraphChi/X-Stream 用”外存 + 顺序 I/O + 边为中心”换来了”单机也能跑大图”。
24.2.9 分布式机器学习：动机与三大并行范式（Distributed ML）
动机（必须讲清数量级）：讲义页 17-21 给出机器学习的基线——机器学习训练”模型”（计算机表示），区分训练（Training）与推理/预测/模型服务（Inference / Prediction / Model Serving）；学习分为监督/无监督/强化；核心优化器是 SGD（随机梯度下降）——最小化一个光滑可微的目标函数，常见变体有 AdaGrad（自适应梯度）、Adam（自适应矩）、RMSProp，讲义还打趣它是”很多分布式 ML 论文的稻草人（strawman）应用”。神经网络（ANN）是另一类模型：它是算子的图（graph of operators），算子可能计算量很大，图可以分成多个阶段/层（stages / layers），相邻阶段的算子之间通常是全连接通信（all to all），大多数边带权重（参数），这些权重的集合就是模型（例外：ReLU 等激活函数、dropout）。训练是前向传播 → 计算与已知答案的误差 → 反向传播更新权重（讲义页 19）；算子之间流动的数据是张量（tensors，多维向量）；默认逐条样本训练（forward + backward），扩展做法是小批量（mini-batch）；训练完成后推理只需要前向传播；新兴方向是在线训练/持续训练（online / continual training）；超参数（hyperparameter，如 batch size）不是”参数/权重”（这是考点级的区分）。ANN 的类型包括 FFNN（前馈，无环）、DNN（多层”隐藏层”）、CNN（卷积，如 Facebook 图片描述）、RNN（层内有环，处理时间/序列，如手机输入法自动纠错）、GNN（图神经网络）、Transformer；讲义页 20 还提到分布式 ML 评测中常用的网络：Inception(v3)、(G)NMT、ResNet。
规模增长带来的结论：模型参数量从百万级涨到数十亿至数万亿，训练数据从 GB 涨到 TB/PB，单机 GPU 的内存与算力都不足以承载——必须分布式训练。讲义页 21 的原话是：集中式机器学习 “Slow and often infeasible“（慢且常常不可行），分布式机器学习则通过多个 worker 并行，并用一个 “Parameter Server” 来聚合 worker 当前迭代的数据、并启动下一轮迭代。讲义同时引用了 Dean et al., “Large scale distributed deep networks”, NeurIPS 2012——这正是参数服务器（Downpour SGD）的经典论文。
三大并行范式（本讲的骨架）：
（1）数据并行（Data Parallelism）：讲义页 22 的定义是”多个 worker 运行同一个模型，每个设备拿到不同的数据，数据被切成小批量（mini-batches），worker 在每个小批量之后同步“，实现方式有两种：Parameter Server 方式（worker 把梯度推给参数服务器，服务器聚合后把新参数发回）与 All-Reduce 方式（每个 worker 把自己的权重多播给其他 worker，大家各自算出相同的结果）。变体：异步训练（Asynchronous training）。
- 优点：实现简单，与现有单机训练代码几乎一致；适合数据量大、模型能装进单机内存的场景；吞吐可以随 worker 数近似线性增长（直到通信成为瓶颈）。
- 缺点：每个 worker 都要放一份完整模型（模型太大就放不下 ⇒ 必须转向模型并行）；通信量正比于模型大小且随 worker 数增长（AllReduce 每轮要交换全部梯度）；梯度聚合点（参数服务器或 AllReduce 的同步点）成为瓶颈。
- 关键挑战：同步 vs 异步 SGD（24.2.11 详解）、通信压缩（梯度量化/稀疏化，24.2.10）、以及大批量导致的泛化性下降（学习率需随 batch 大小调整）。
（2）模型并行（Model Parallelism）：讲义页 22 的定义是”同一个模型（DNN 图）被切分到多个设备上，每次一个输入穿过这批设备“——即把不同的层放在不同机器上（层间切分），或把同一层切成多片（层内切分）。
- 优点：可以训练单机放不下的超大模型（参数量超过单卡显存）。
- 缺点：通信频繁——每一层的激活值与梯度都要跨机器传递，且往往处于串行依赖链上；流水线气泡（pipeline bubble）——层与层之间的依赖会让下游 GPU 在等上游时闲置；实现复杂、负载均衡困难（不同层的计算量不同，算子粒度不均会导致某些设备长期空闲）。
- 现代变体：流水线并行（Pipeline Parallelism）——把模型按层切成若干 stage，用 micro-batch 填充流水线以掩盖气泡（如 GPipe、PipeDream）；张量并行（Tensor Parallelism）——把单个矩阵乘切成多片、多卡协同算一个算子（如 Megatron-LM 的列并行/行并行）。补充说明：讲义页 22 只区分了”数据并行 / 模型并行”，流水线与张量并行是后续文献对”模型并行”的细分，但它们与讲义中”层放在不同设备、任意时刻一个输入穿过整批设备”的描述完全兼容。
（3）混合并行（Hybrid / 3D Parallelism）：数据并行 × 流水线并行 × 张量并行三层叠加。典型布局是：最外层按数据切分（不同数据副本），中间层按层切分（流水线 stage），最内层在单机内的多卡之间切分张量；这是现代大模型训练的事实标准（GPT-3、Megatron-Turing 等）。
机制图解（四种切分方式的对比）：
数据并行（Data Parallel）：切数据，不切模型
data shard 1 data shard 2 data shard 3
| | |
[ 完整模型副本 ] [ 完整模型副本 ] [ 完整模型副本 ]
| | |
grad_1 grad_2 grad_3
+---------+---------+---------+---------+
|
+--> AllReduce 或参数服务器：聚合成平均梯度
w = w - lr * mean(grad) -> 分发给所有 worker
* 通信量 ∝ 模型大小 x worker 数；模型必须能放进【每个】worker
模型并行-层间（Model Parallel, layer-wise）：切模型，不切数据
worker 1 : Layer 1 -> Layer 2 | --> 激活值 --> worker 2 : Layer 3 -> Layer 4
* 任意时刻只有一个输入在流动；串行依赖 -> 气泡；通信 = 层边界上的激活/梯度
流水线并行（Pipeline Parallel）：层间切分 + micro-batch 填充
time --> | mb0 | mb1 | mb2 | mb3 |
stage1 [L1 ] [L2 ] [L3 ] [L4 ]
stage2 [L1 ] [L2 ] [L3 ] [L4 ]
stage3 [L1 ] [L2 ] [L3 ] [L4 ]
* 用多个 micro-batch 填满流水线，把"气泡"摊薄
张量并行（Tensor Parallel）：把单个矩阵乘切开
Y = X @ W, W 按列切: W = [W1 | W2]
GPU0: Y1 = X @ W1 GPU1: Y2 = X @ W2
* 每层都要通信（AllReduce），因此只在【单机内 NVLink】这种高带宽场景使用
- 四种并行的全面对比：
| 维度 | 数据并行 | 模型并行（层间） | 流水线并行 | 张量并行 |
|---|---|---|---|---|
| 切分对象 | 训练数据（模型每份完整） | 模型的层 | 模型的层 + micro-batch | 单个算子内的张量 |
| 通信模式 | AllReduce / Push-Pull 梯度（一轮一次） | 层边界的激活与梯度（每层一次，串行） | stage 边界（每 micro-batch 一次） | 算子内 AllReduce（每层多次） |
| 通信频率 | 低（每个 iteration 一次） | 高 | 中 | 极高 |
| 每卡内存需求 | 完整模型 + 优化器状态 | 仅自己的层 | 仅自己的 stage | 仅自己那一片张量 |
| 可扩展性上限 | 受 batch 大小与通信效率限制 | 受层数限制（层数是硬上限） | 受 micro-batch 数与气泡比例限制 | 受单层宽度与卡间带宽限制 |
| 负载均衡 | 容易（数据均分） | 难（层大小不一） | 中（stage 需按计算量切） | 容易（等分张量） |
| 典型场景 | 数据大、模型能装下 | 模型巨大、层数多、机器少 | 超大模型 + 大数据（跨机） | 单机多卡高带宽（NVLink） |
| 代表系统 | TensorFlow 参数服务器、PyTorch DDP、Horovod | TensorFlow 初版 model parallel、Mesh-TensorFlow | GPipe、PipeDream、Megatron 的 pipeline | Megatron-LM、DeepSpeed |
- 讲义的框架层视角：讲义页 23-27 依次介绍 TensorFlow（Google 出品，大规模训练与推理框架，隐藏分布细节，用数据流图（dataflow graph）表示计算、共享状态与改变状态的操作，运行时有 200+ 标准算子，支持 CPU/GPU/TPU，支持多种通信协议；页 24 给了图像分类器的示例：先构造图
x -> W_1 -> relu -> W_2 -> softmax，再加优化算子节点（AdagradOptimizer(0.01).minimize(loss)），最后用sess.run(train_op, ...)在数据上迭代执行）；PyTorch（Meta 出品，动态计算图，自动求导，Linear模块内部就是input * weight + bias，生态包含 TorchText/TorchVision/TorchAudio；页 26 给出了DistributedDataParallel（DDP）的用法：dist.init_process_group("gloo", rank, world_size)建立进程组、DDP(model, device_ids=[rank])包装模型、之后loss.backward()+optimizer.step()就是标准的数据并行——DDP 在反向传播时自动做梯度的 AllReduce）；JAX（Google 出品，NumPy 风格 API，多后端 CPU/GPU/TPU，四个关键变换：grad自动微分、jit编译、vmap自动向量化、pmap自动并行——其中pmap就是数据并行的函数式表达）。
24.2.10 参数服务器架构（Parameter Server）
定义与目的：参数服务器（Parameter Server, PS）把”模型参数”从计算节点中独立出来，成为专门的一组服务器：server 持有全局模型参数的分片，worker 负责计算。worker 通过 Pull（拉取参数）→ 计算梯度 → Push（推送梯度） 与 server 交互，server 负责聚合梯度并更新参数。讲义页 21 正是这个结构的最简描述：”用一个 Parameter Server 聚合 worker 当前迭代的数据，并在 worker 端开始下一轮迭代”。
直观解释（”它是什么？”）：参数服务器就像图书馆 + 读者：读者（worker）到图书馆借阅（Pull）最新的书（参数），回去读完写一份批注（梯度），再把批注交回（Push）；图书馆统一把所有人的批注整理进新版书里。好处是读者之间不需要互相认识（worker 之间不通信），而且图书馆可以有好几间分馆（参数分片到多台 server），于是单机内存不再是上限。
机制图解（多 server 分片 × 多 worker 的 Push/Pull）：
+-----------+ +-----------+ +-----------+
| Server 0 | | Server 1 | | Server 2 |
| w[0:k] | | w[k:2k] | | w[2k:3k] |
| shard+rep | | shard+rep | | shard+rep |
+-----+-----+ +-----+-----+ +-----+-----+
^ ^ ^
| Push 梯度 / Pull 参数（每个 worker 只传自己负责的那一片）
+--------+---------+--------+---------+--------+---------+
| | | |
+--+-----+ +---+----+ +---+----+ +-----+--+
|Worker 0| |Worker 1| |Worker 2| |Worker 3|
+--------+ +--------+ +--------+ +--------+
数据分片 数据分片 数据分片 数据分片
完整模型副本 完整模型副本 完整模型副本 完整模型副本
注：worker 之间不直接通信；每个 server 负责一片参数（shard），
副本（rep）用于容错；worker 可以只拉自己这一步需要的那几片。
一次 worker 迭代：pull(w_shard) -> 前向/反向算梯度 g -> push(g_shard)
同步模式：server 等齐所有 worker 的 g 再更新；异步模式：收到谁的 g 就用谁的
优点：解耦计算与通信（worker 只与 server 交互，worker 数可弹性伸缩）；参数可分片到多台 server，突破单机内存限制；天然支持异步更新（server 收到一个梯度就更新，不等别人）；容错（server 可以有副本；worker 挂了不影响参数本身）。讲义页 21 的表述把 PS 的核心价值总结为”聚合当前迭代数据 + 启动下一轮”。
- 通信压缩与稀疏化（降低通信瓶颈的关键技术）：模型越大，Push/Pull 的字节数越大，通信时间会超过计算时间。工程上的三类手段：
- 梯度量化（quantization）：把 32 位浮点梯度压成更少比特——1-bit SGD（每个梯度只保留符号位）、TernGrad（三值化）。
- 梯度稀疏化（sparsification）：只发送绝对值大的梯度（例如 top-1%），其余当作 0（很多梯度本来就很接近 0）。
- 误差补偿（error feedback）：被丢弃/被量化掉的那部分误差累积在本地，下一轮补上——这是保证”压缩不破坏收敛”的关键技巧。 补充说明：1-bit SGD 报告过约 30× 的压缩比，这一数字来自相关论文而不是课程讲义；使用时请记住它以额外误差补偿与收敛速度的调整为代价。
- 关键假设与系统模型：server 是有状态的（持有参数），因此需要考虑 server 故障（副本 + 一致性，回指 Lecture 10/17 的复制与共识）；worker 是无状态的（数据在存储上，梯度随算随推），因此 worker 故障容易处理（换一台重跑）。
24.2.11 同步 vs 异步 SGD：分布式 ML 的核心权衡
定义与目的：同步 SGD（Synchronous SGD，BSP 风格）：所有 worker 每轮算完梯度后等待最慢的 worker（屏障），把梯度聚合成平均梯度后统一更新参数。异步 SGD（Asynchronous SGD）：worker 不等任何人，读当前参数 → 算梯度 → 推送并立即生效。有界延迟异步（Stale Synchronous Parallel, SSP）：允许 worker 之间的进度差不超过 $s$ 个迭代（staleness bound $s$），若某 worker 领先超过 $s$，则必须等待。
- 直观解释（”它是什么？”）（小组做作业的类比）：
- 同步 SGD = 每做一道题，所有人做完后一起对答案，然后再做下一道。好处是每个人的下一道题都基于最新且一致的答案；代价是最慢的人决定全组进度（他卡住，全组卡住）。
- 异步 SGD = 谁做完谁就把答案交到讲台上，讲台上的答案随时更新；别人可能参考到几分钟前的旧答案（stale gradient）。速度快，但大家可能基于互相矛盾的中间答案推进，步子迈大了就会来回震荡甚至发散。
- SSP = “最多允许落后 $s$ 道题”：谁做完都可以交，但如果某人已经领先别人超过 $s$ 道题，他就必须停下来等。这样既不会像同步那样被无限拖累，也不会像异步那样参考到太旧的答案。
Staleness 的定义与影响（必须讲清）：设 worker 读到参数时的版本号为 $v_{\text{read}}$，它推送梯度并被应用时的版本号为 $v_{\text{apply}}$，则这次更新的陈旧度（staleness）为 $\tau = v_{\text{apply}} - v_{\text{read}}$。worker 实际计算的是 $\nabla f(w_{v_{\text{read}}})$，却被用在 $w_{v_{\text{apply}}}$ 上，于是产生梯度误差： \(\nabla f(w_{\text{old}}) \approx \nabla f(w_{\text{new}}) - H\,(w_{\text{new}} - w_{\text{old}}) = \nabla f(w_{\text{new}}) - H\,\Delta_\tau\) 其中 $H$ 是 Hessian。误差量级正比于 $\tau$（陈旧度）× 这期间参数移动的距离 $\Delta_\tau$。这解释了工程上的三条经验：(1) 异步必须用更小的学习率（学习率越大，$\Delta_\tau$ 越大，误差越大）；(2) 陈旧度必须有界（否则 $\Delta_\tau$ 无界，误差无界）；(3) 用动量/自适应优化器时要额外小心（动量会把陈旧梯度的方向放大）。
- 机制图解（同步 / 异步 / SSP 的时间线对比——本章第二重要的图）：
时间轴 -------------->
(A) 同步 SGD：每轮一次屏障，等待最慢者
W0: [--g--] [--g--] [--g--]
W1: [--g--] [--g--] [--g--]
W2: [-----g-----](慢) [-----g-----](慢) [-----g-----](慢)
B-----------------B-----------------B B = barrier
每轮 wall time = max(t0,t1,t2)；straggler 的代价被【乘以轮数】
(B) 异步 SGD：不等待，但梯度可能陈旧
W0: [g][g][g][g][g][g]
W1: [g][g][g][g]
W2: [---g---][---g---][---g---]
版本 v: 1 2 3 4 5 6 7 8 9 ...
W2 的一次 push 可能把"基于 v=3 的梯度"应用到 v=7 上 => staleness = 4
(C) SSP（s=2）：允许领先 2 个迭代，超过就等
W0: [g][g][g] (等) [g][g][g]
W1: [g][g][g][g][g][g][g]
W2: [---g---][---g---][---g---]
W1 的 clock 领先 min(clock) 超过 2 -> W1 必须等待，直到 W2 追上
s=0 <=> 退化为同步（每轮必须全员对齐）；s=inf <=> 退化为完全异步
- SSP 的伪代码与 $s$ 的作用（详见 24.3.7）：每个 worker 维护本地时钟 $c_i$（已完成的迭代数），全局维护 $c_{\min}=\min_j c_j$。worker $i$ 在开始第 $k$ 次迭代前检查 $c_i - c_{\min} \le s$：满足则继续，否则阻塞等待直到最慢的 worker 推进（等待期间不消耗算力，但也不产出）。
- $s=0$：任何 worker 都不能领先——退化为同步 SGD 的轮次结构；
- $s=\infty$：永远满足条件——退化为完全异步；
- $s$ 越大：等待越少（wall-clock 更快），但 staleness 越大（收敛性越差）。第 24.4 节的实测：$s=0$ 需 62.4 时间单位、平均陈旧度 0；$s=3$ 需 52.0、平均陈旧度 0.60；$s=\infty$ 只需 20.8、最大陈旧度 9。
- 其他变体：
- 弹性平均 SGD（EASGD, Elastic Averaging SGD）：worker 各自保存本地参数 $w_i$，并额外被一个中心参数 $\bar{w}$ 通过弹簧力拉回（$w_i \leftarrow w_i - \eta(\nabla f_i(w_i) + \rho(w_i-\bar{w}))$），$\rho$ 控制”探索 vs 一致”的权衡——它不强求每轮同步，但对偏离中心的 worker 施加惩罚。
- 局部 SGD / 联邦平均（Local SGD / Federated Averaging, FedAvg）：worker 各自本地更新 $H$ 步，然后才同步一次（平均各 worker 的模型）。它把”通信频率”直接除以 $H$：实验中 $H=1$ 时与同步 SGD 完全等价（192 次梯度、384 条消息、最终损失 0.01976 逐位相同），$H=50$ 时每条梯度的消息数从 2.0 降到 0.04（50× 减少）。Local SGD 是联邦学习的基础（见 24.2.13）。注意：$H$ 越大，同步间隔内各 worker 的参数漂移越大，因此必须配合更小的学习率（实验中使用同步 SGD 的学习率跑 $H=20$ 会发散）。
- 全面对比表：
| 维度 | 同步 SGD | 异步 SGD | SSP | Local SGD / FedAvg |
|---|---|---|---|---|
| 收敛性 | 最好：等价于大 batch 的 SGD，步长可控、理论清晰 | 最差：staleness 引入梯度误差，学习率必须调小，可能震荡/发散 | 接近同步：陈旧度有界 $\Rightarrow$ 误差有界（实验中 $s\le10$ 仍收敛，$s=\infty$ 发散） | 与同步接近（$H$ 大时需减小学习率） |
| straggler 敏感度 | 高：最慢者决定每轮时间 | 低：慢 worker 只影响自己的产出 | 中：领先者要等最慢者，但等待有上界 | 中：每 $H$ 步同步一次，等待频率降为 $1/H$ |
| 通信频率 | 每迭代一次 AllReduce/Push-Pull | 每迭代一次（但无同步等待） | 每迭代一次（可能因等待而降低有效速率） | 每 $H$ 迭代一次 ⇒ 通信量降 $H$ 倍 |
| 容错 | 差：一个 worker 挂了整轮停摆（要么重启，要么回滚） | 好：server 继续接受其他 worker 的梯度 | 中：慢/坏 worker 会通过 $c_{\min}$ 拖住领先者（可用”踢掉超时 worker”处理） | 中：$H$ 步本地计算可容忍短暂断连（联邦学习的关键属性） |
| 实现复杂度 | 低 | 中（版本管理、无锁更新、收敛调参） | 中高（时钟同步、阻塞与唤醒、死锁避免） | 低-中（本地多步 + 周期性聚合） |
| 确定性 | 确定（同一 batch 划分下可复现） | 不确定（更新顺序影响结果） | 不确定（但陈旧度有界） | 相对确定（同步点固定） |
| 适合的算法/场景 | 需要稳定收敛与可复现实验；同构集群 | 异构/共享集群、容错优先、能容忍收敛性下降 | 想要”接近同步的收敛性 + 接近异步的速度” | 通信昂贵（跨数据中心、移动端）、联邦学习 |
24.2.12 分布式 ML 的容错与 straggler 处理
检查点（Checkpointing）：训练过程要周期性保存模型参数（以及优化器状态，如 Adam 的一阶/二阶矩）。大模型动辄数百 GB，因此检查点本身就是重活：需要分片写（每个 worker 写自己那一片）、异步写（不阻塞训练）、以及只保留最近 K 份的策略。与 Pregel 的检查点不同，这里的状态就是参数本身，而”重算”代价极高（重算若干个 iteration 远比重放超级步昂贵）。
worker 故障：同步 SGD 下，一个 worker 失败会拖住整轮——要么重启该 worker 并等待（其他 worker 空转），要么回滚到上一个检查点（浪费整轮计算）；实践中常用弹性训练（elastic training）：把失败的 worker 从进程组中剔除、按剩余 worker 数重建通信组并继续（代价是 batch 大小与学习率需要动态调整）。异步 SGD 下容错天然更好：一个 worker 挂了，server 继续接受其他 worker 的梯度，训练不中断，只是吞吐下降。
server 故障：参数服务器是有状态的，因此必须处理——实践中用参数分片 + 每个分片多副本（如 3 副本）+ 定期检查点；分片之间彼此独立（参数之间没有强一致要求），因此一致性需求比 Paxos 场景弱得多（回指 Lecture 10 的一致性模型与 Lecture 20 的复制控制：这里通常只需要”最终一致 + 定期快照”）。补充说明：如果把 server 也做成强一致的复制状态机（Lecture 17 的 Paxos/Raft），共识的开销会直接落在每个梯度更新上——实践中的做法是避免让共识进入数据面。
静默数据损坏（silent data corruption）：机器不崩溃但算错（内存位翻转、GPU 计算出错、网络丢包未被察觉）。对训练的危害是隐蔽且累积的：错误梯度被”悄悄吸收”进参数里，训练看起来还在跑，但模型质量下降且难以归因。缓解手段包括校验和、端到端 loss 异常检测、以及周期性用干净的检查点重算对比（这也再次说明”重算”这一工具的普适性——与 Lecture 5 的 MapReduce 用重算代替恢复状态是同一哲学）。
straggler 的处理（三种手段，一条重要的跨章连接）：
- 忽略慢 worker（异步）：直接不等——最简单，但换来 staleness。
- 备份 worker（backup worker）：与 Lecture 5 中 MapReduce 的 speculative execution / backup task 完全对应——”记录每个任务的进度（% done），对慢任务主动启动一个备份副本，第一个完成的副本算数，其他副本被杀死”。分布式 ML 中的对应做法是冗余计算（redundant computation）：把同一份 mini-batch 分给两个 worker，谁先算完用谁的。
- 梯度编码（gradient coding）：用纠删码（erasure coding）的思想，把 $N$ 个 worker 的梯度编码成 $N+k$ 份并分发；只要有任意 $N$ 份返回（包括最快的 $N$ 份），就能恢复出完整的平均梯度——于是”最慢的 $k$ 个 worker 可以被直接忽略”，把 straggler 问题转化成”选择等待谁”的问题。这是”用冗余计算换时间”的典型设计，与 Lecture 5 的备份任务、以及存储层的纠删码（Hadoop 3.x 用 erasure coding 替代 3 副本，见 Lecture 5）共享同一个数学内核。
24.2.13 联邦学习（Federated Learning）简述
定义与目的：讲义页 28 给出了清晰对照：分布式 ML 假设各 worker 的数据是同构的（homogenous data across workers）；联邦机器学习允许各 worker 的数据是异构的（heterogeneous），worker 可以是数据中心，也可以是移动设备，并且存在故障、可能还有隐私问题。讲义给出的例子是 Google 用联邦学习在移动设备上预测按键（keystroke prediction）。
与数据中心内分布式训练的四点差异：(1) 数据不出本地——原始数据留在用户设备上，只上传模型更新（梯度或权重），这是隐私需求的直接结果；(2) 客户端不可靠且异构——设备可能断网、关机、电量不足、算力差异巨大，”worker 数”每一轮都在变（可能只有 1% 的设备在线参与）；(3) 通信极其昂贵——移动网络的上下行带宽远低于数据中心内部网络，因此 Local SGD / FedAvg（本地多步 + 少量同步）成为基础算法；(4) 参与方可能作恶或数据敏感——因此引入安全聚合（Secure Aggregation，服务器只能看到各方更新的总和，看不到单个客户端的更新）与差分隐私（Differential Privacy，在更新里加噪声，使单个样本的影响不可区分）。
| 维度 | 数据中心内分布式训练 | 联邦学习 |
|---|---|---|
| 数据分布 | 同构（随机切分，IID） | 异构、非 IID（每个设备只有自己的使用习惯） |
| 节点数与可靠性 | 数十到数万 GPU，相对稳定 | 百万级设备，每轮只有少量在线，随时掉线 |
| 网络 | 高带宽（NVLink/InfiniBand） | 移动网络，通信是最贵的资源 |
| 主要瓶颈 | 同步/通信/内存 | 通信轮数与客户端漂移（client drift） |
| 核心算法 | 数据/模型/流水线/张量并行 + AllReduce | FedAvg / Local SGD、安全聚合、差分隐私 |
| 容错策略 | 检查点 + 弹性训练 + backup worker | 天然容忍掉线（采样参与）、忽略超时设备 |
| 典型系统 | TensorFlow、PyTorch DDP、Megatron、DeepSpeed | Google 移动端键盘预测、跨机构的医疗联邦 |
24.2.14 图处理与机器学习的交汇：GNN 的一层就是一个超级步
定义与目的：图神经网络（GNN, Graph Neural Network）把 Pregel 式的邻域聚合（neighborhood aggregation / message passing）与神经网络的可学习参数结合起来：每一层做”聚合邻居的特征 → 用可学习权重变换 → 更新自身特征“。而 Pregel 的一个超级步做的正是”接收邻居消息（Gather）→ 用用户函数计算（Apply）→ 发给邻居（Scatter）“。两者的结构完全相同——区别只在于 Pregel 的
Compute()是用户写死的确定性函数，而 GNN 的聚合/变换函数里含有可训练参数（并且需要反向传播来更新它们）。机制图解（GNN 的一层 = Pregel 的一个超级步）：
GNN 第 l 层（对应 Pregel 的第 l 个超级步）：
顶点 v 的输入 = 第 l-1 层所有邻居的隐藏向量
h_u^(l-1) h_w^(l-1) h_x^(l-1) <-- 邻居在上一层的表示
\ | /
\ | / <-- Gather：消息就是邻居的隐藏向量
+-------+---------+
|
AGGREGATE(.) <-- 与 Pregel 的 combine() 同理：sum / mean / max
|
W^(l) (可学习) <-- Pregel 里这里是用户写死的函数；GNN 里这里是【参数】
|
h_v^(l) = sigma( W^(l) . AGG(h_neighbors) ) <-- Apply：更新自身
Pregel: Compute(): reads inbox (neighbors' messages) -> new_value -> SendMessage()
GNN 一层: h_v^(l) = sigma( W^(l) AGG({h_u^(l-1) : u in N(v)}) )
L 层 GNN = L 个超级步 => 顶点 v 的感受野(receptive field) = 它的 L 跳邻域
- GNN 训练的两个特有挑战：
- 邻域爆炸（neighborhood explosion）：$L$ 层 GNN 需要 $L$ 跳邻居，感受野大小随层数指数增长（度数 $d$ 的图，$L$ 跳邻居可达 $d^L$ 个）。解决办法是采样——GraphSAGE 每层只采样固定数量的邻居，把”指数增长”变成”每层常数”；这也让 GNN 可以按 mini-batch 训练（每次只需要采样出的子图）。
- 分布式 GNN 训练是两难的交集：它同时承担图分割的困难（幂律图、跨机边＝跨机消息，即 24.2.4 的问题）与模型训练的困难（梯度同步、staleness、通信压缩，即 24.2.9-24.2.12 的问题）。因此同一套基础设施里既有”按顶点划分”（edge-cut，但 hub 顶点会爆炸）又有”按层划分”（把不同 GNN 层放到不同机器，本质是模型并行）。这正是本讲把图处理与分布式 ML 放在一起讲的现实理由。
- 图算法在 ML 中的应用（讲义页 4 的应用清单在 ML 侧的延伸）：PageRank 用于推荐系统的物品重要性排序；图嵌入（graph embedding，如 node2vec）把顶点映射成向量作为下游模型的特征；社区发现（community detection，即连通分量/标签传播的推广）用于用户分群与异常检测；最短路用于路径规划与知识图谱多跳推理。
24.3 算法伪代码与正确性分析
算法 24.3.1：Pregel 的超级步执行框架（Master + Worker）
假设与系统模型
- 进程数：1 个 leader/master + $N$ 个 worker；master 只做协调（分配分区、驱动超级步、检测故障、聚合全局值），不持有顶点。
- 同步模型：BSP。每个超级步内所有 worker 并行执行；超级步结束处有全局屏障。
- 通道：消息可靠、不丢失、不重复；不要求 FIFO（因为同一超级步内的消息在步末统一投递，顺序无关紧要）。
- 故障模型：crash-stop / crash-recovery，非拜占庭；故障在超级步边界被系统统一处理（见 24.3.2）。
- 用户函数：
Compute()确定性；combine()满足交换律与结合律；聚合器的归约函数满足交换律与结合律。 - 图：静态拓扑（本节先不考虑拓扑变更；动态拓扑见文末”补充说明”）。
伪代码
=== MASTER ===
初始：把输入图切成 N 个分区 P_1..P_N，分别交给 worker_1..worker_N
对每个 worker w：发送指令 Load(P_w)；每个顶点初始标记为 active
aggregated := {} # 全局聚合值的当前快照（供下一步读取）
superstep := 0
while true:
# ---------- 全局终止判定：没有活跃顶点且没有在途消息 ----------
if (所有顶点的 active 标志均为 false) and (所有 worker 的 outbox 均为空):
break
# ---------- 周期性检查点（在迭代开始时） ----------
if superstep mod N_ckpt == 0:
send Checkpoint(superstep) to all workers # 各自保存分区状态（见 24.3.2）
# ---------- 广播本超级步的全局聚合值（上一步归约结果） ----------
send StartSuperstep(superstep, aggregated) to all workers
# ---------- 收集屏障：等待所有 worker 完成本超级步 ----------
for each worker w (in any order):
receive SuperstepDone(w, active_count_w, aggregator_values_w, outbox_w)
# ---------- 屏障已到：聚合器归约（k 步的值在 k+1 步可读） ----------
aggregated := reduce_over_workers(aggregator_values_w)
# ---------- 消息投递：把 outbox 送到目标顶点的 inbox（唤醒停机顶点） ----------
for each message (target, m, from_w) in 所有 worker 的 outbox:
if target 的 active == false: # *** 唤醒机制 ***
target.active := true ; target.pending_halt := false
log("wake-up: vertex", target, "woken by", m)
target.inbox.append(m)
# ---------- 应用"投票停机" ----------
for each vertex v:
if v.pending_halt == true: v.active := false ; v.pending_halt := false
superstep := superstep + 1
send SaveOutput() to all workers # 把结果写回分布式存储
=== WORKER w（每个超级步执行一次） ===
upon StartSuperstep(superstep, aggregated):
outbox := [] ; aggregators := {}
for each vertex v in P_w with v.active == true: # 只跑活跃顶点
msgs := v.inbox # ① Receive：读上一步投递的消息
v.inbox := []
Compute(v, msgs, superstep, aggregated) # ② Compute：用户函数
# 用户可做三件事：
# (a) 改 v.value
# (b) SendMessage(target, m) -> outbox.append((target, m))
# (c) VoteToHalt() -> v.pending_halt := true
# (d) Aggregate(name, val) -> aggregators[name].append(val)
# ③ Send 在超级步【结束时】发生：先本地 combine，再交给 master 投递
for each target t in group_by_target(outbox):
m_list := outbox[t]
if combine is defined and |m_list| > 1:
m := fold(combine, m_list) # 要求：交换律 + 结合律
send_to_master(t, m)
else:
for each m in m_list: send_to_master(t, m)
send SuperstepDone(w, count_active(P_w), aggregators) to master
算法逻辑解说（配合 24.3.3 / 24.3.4 的数值例子看）
- 启动（讲义页 12 的步骤 1-2）：多份程序副本在集群上开始执行；leader 给每个 worker 分配一个顶点分区；worker 把顶点加载进来并全部标记为活跃。
- 驱动（讲义页 12 步骤 3）：leader 指示每个 worker 执行一次迭代；worker 遍历自己的活跃顶点并对每个顶点执行
Compute()；消息可以随时发送，但必须在迭代结束（屏障）前投递完成；worker 完成一轮后通知 leader，并把下一轮的活跃顶点数与聚合器值一起上报；当所有 worker 都到达迭代屏障，leader 才启动下一轮。 - 终止（讲义页 12 步骤 4）：当某一轮中没有任何顶点活跃、且没有消息在途时，计算停止。注意两个条件的联合判定发生在投递之后：先投递（可能唤醒一批顶点），再看活跃集合是否为空。
- 收尾（讲义页 12 步骤 5）：leader 指示每个 worker 保存自己那部分图（作为输出写回分布式存储）。
- 关键时序：第 $k$ 步发出的消息在第 $k+1$ 步被接收——这一条使”顶点在超级步 $k$ 的输入”完全由”超级步 $k-1$ 结束时的全局状态”决定。
正确性论证
- 确定性（Determinism）：命题：给定输入图、初始值、确定性用户函数与确定性 combine/aggregator，”超级步 $k$ 结束时的全局状态 $S_k$”是唯一确定的，与 worker 内部的遍历顺序、消息到达顺序、线程调度都无关。
- 证明（对 $k$ 归纳）：基础：$S_0$ 由初始图与初始值唯一确定。归纳：假设 $S_k$ 唯一确定。超级步 $k+1$ 中，顶点 $v$ 的输入包含两部分：(i) 它在 $S_k$ 中的自身状态（值、出边、是否活跃）——由归纳假设唯一；(ii) 它在超级步 $k+1$ 收到的消息集合 $M_{k+1}(v)$。而 $M_{k+1}(v)$ 完全来自超级步 $k$ 结束时各 worker 的 outbox，其内容由 $S_k$ 与用户函数决定（用户函数确定性 ⇒ 产生相同的一组消息）；因为投递发生在超级步 $k$ 的末尾（在 $k+1$ 的任何计算开始之前），所以 $M_{k+1}(v)$ 与消息在网络中的到达先后无关，是一个确定的集合（甚至是一个确定的多重集）。若使用了 combine，则因为
combine满足交换律与结合律，任何折叠顺序都得到同一个值 ⇒ $M_{k+1}(v)$ 的归并结果也唯一。于是 $v$ 的Compute()输入唯一 ⇒ 输出唯一 ⇒ $S_{k+1}$ 唯一。又因为屏障保证了”$S_{k+1}$ 只在所有 worker 都完成 $k+1$ 步后才会被读取”，不存在”读到半成品状态”的可能。$\square$ - 依赖的假设：用户函数确定性（若用户在
Compute()里调用随机数或读本地时间，确定性立刻失效）；combine 满足交换律与结合律；消息不丢失不重复（否则输入集合会变）。 - 意义：确定性是容错与调试的基础——只有”重放必然得到同一结果”，”回滚到检查点重算”才是正确的恢复策略。
- 证明（对 $k$ 归纳）：基础：$S_0$ 由初始图与初始值唯一确定。归纳：假设 $S_k$ 唯一确定。超级步 $k+1$ 中，顶点 $v$ 的输入包含两部分：(i) 它在 $S_k$ 中的自身状态（值、出边、是否活跃）——由归纳假设唯一；(ii) 它在超级步 $k+1$ 收到的消息集合 $M_{k+1}(v)$。而 $M_{k+1}(v)$ 完全来自超级步 $k$ 结束时各 worker 的 outbox，其内容由 $S_k$ 与用户函数决定（用户函数确定性 ⇒ 产生相同的一组消息）；因为投递发生在超级步 $k$ 的末尾（在 $k+1$ 的任何计算开始之前），所以 $M_{k+1}(v)$ 与消息在网络中的到达先后无关，是一个确定的集合（甚至是一个确定的多重集）。若使用了 combine，则因为
- 终止性（Termination）：命题：若用户的
Compute()满足”只在顶点值发生变化时才发送消息“，且顶点值的取值空间在算法意义下单调收敛（例如距离只减不增、标签只取更小的 ID、PageRank 残差按压缩因子衰减），则算法在有限个超级步内终止。- 证明要点：(a) 单调性：每类算法都有一个单调下降的有界量（SSSP 中是 $\sum_v d(v)$，每次严格下降至少 $\varepsilon$；连通分量中是各顶点的标签值，取 $\min$ 所以单调不增且有下界 $\min_v \text{id}$；PageRank 中是最大残差 $\max_v\vert \text{PR}^{(k)}(v)-\text{PR}^*(v)\vert $，按 24.3.3 的压缩映射以 $d^k$ 衰减）。(b) 有限状态：这些量只能取有限多个”有意义的”取值（距离是两个顶点间的某条路径长度，只有有限多条简单路径；标签只有 $\vert V\vert $ 种；残差降到 $\varepsilon$ 以下）。(c) 活跃顶点有限：每个顶点只有在值变化时才会发消息；不发消息 ⇒ 邻居不会被唤醒 ⇒ 活跃集合在有限步内清空。(d) 无消息在途：投递发生在超级步末尾，因此”活跃集合为空”的那一刻，outbox 也已经投递完毕 ⇒ master 在下一步的开头判定终止。$\square$
- 必须明确指出的一点：Pregel 系统本身不保证算法终止，这是用户的责任。 系统只提供”活跃/停机/唤醒”这套机制与”无活跃顶点且无在途消息则结束”的判定；如果用户算法不保证单调收敛（例如把顶点值写成”每轮加 1”、或让其震荡不衰减的值一直变化），那么顶点会永远活跃、作业永不终止（用户只能靠
max_supersteps之类的上限兜底）。第 24.4 节的引擎同样只提供这一判定，实验中所有算法都靠”残差/距离单调性 + 阈值”自行终止。
安全性（Safety）：任何时刻，”顶点的值”只被它自己的
Compute()修改（本地性），因此不存在两个 worker 并发写同一个顶点值的可能——这一点由分区互斥（每个顶点只属于一个分区）保证，是 BSP 模型”不需要锁”的根源。- 活性（Liveness）：只要用户函数在有限步内让活跃集合清空，系统就会在有限超级步内终止（上面的终止性论证）；若用户函数不满足该性质，系统保证不会错误终止（不会在还有活跃顶点时宣布结束），但可能永不终止。
复杂度
- 轮次复杂度：超级步数 $K$ = 算法的迭代轮数。对 BFS/SSSP 这类”每轮推进一跳”的算法，$K = O(\text{直径 } D)$；对 PageRank，$K = O\!\left(\frac{\log(1/\varepsilon)}{1-d}\right)$（见 24.3.3）。注意：$K$ 是串行时间下界，因为屏障把每一轮变成一次同步点。
- 消息复杂度：$\sum_{k=0}^{K-1} \sum_{v \in \text{active}(k)} \deg^{out}(v)$，即每轮活跃顶点的出度之和。对稠密活跃的 PageRank，$O(K\vert E\vert )$；对只有前沿活跃的 BFS/SSSP，$O(\vert E\vert )$ 到 $O(D\vert E\vert )$。
- 空间复杂度：每个 worker 保存自己分区的顶点值 + 出边表 + 收件箱；跨分区的边需要额外保存目标 worker 的 ID（讲义页 11 的”每个 worker 只保存自己的分区”带来的内存开销）。总体 $O(\vert V\vert +\vert E\vert +\vert M\vert _{\text{in-flight}})$。
- 每轮通信下限：$\Omega(\text{跨机边的消息数})$，这就是分区质量（24.2.4）与 combine（24.2.5）要优化的量。
补充说明（讲义未展开但值得知道的两点）：(1) 拓扑变更（mutating topology）：Pregel 允许在计算过程中增删边（因为边的增删是”局部”操作，只涉及相关顶点）；但增删顶点需要协调——Pregel 采用的约定是”先删除该顶点的所有边，再重新添加”，因为顶点的创建/销毁涉及分区归属与 ID 空间的全局一致性。(2) 内存中的远程边：每个 worker 只持有自己分区的顶点，但它的出边可能指向别的分区，因此每条远程边都要额外记录目标 worker 的 ID，这在幂律图上会带来可观的内存开销，也是”按度数感知的分区”（把高 degree 顶点与其边放在一起）能省内存的原因。
算法 24.3.2：Pregel 的容错与检查点恢复
假设与系统模型
- 故障模型：worker crash-stop（崩溃后不再发送任何消息），由 master 通过周期性 ping 超时检测；master 本身不故障（补充说明：论文口径下 master 也做检查点）。
- 持久存储：可靠、可寻址（GFS/BigTable/HDFS）；检查点写在那里（讲义页 13）。
- 关键前提：BSP 屏障（24.3.1 的确定性）——它保证”所有 worker 停留在同一个超级步编号上”，因此”回到检查点”有唯一确定的含义。
伪代码
=== MASTER：检查点 ===
每 N_ckpt 个超级步，在超级步开始时：
for each worker w (parallel):
send Checkpoint(superstep) to w
for each w: receive CheckpointDone(w) # 全部分区状态已落持久存储
记录 checkpoint_meta = { superstep, 每个顶点属于哪个 worker, aggregated }
=== MASTER：故障检测 ===
对每个 worker w 维护 last_ping[w]：
周期性地 ping w；若超时（超过 T_ping 未回应）：
标记 w 为 failed，停止向它发送指令
=== MASTER：恢复 ===
upon 检测到 worker w 失败：
# 1) 重分配分区：把 w 的分区 P_w 均分给存活的 worker（负载均衡）
survivors := {u : u 未失败}
把 P_w 切成 |survivors| 份，分配给 survivors，更新 owner 映射
# 2) 让所有 worker 回滚到最近一次检查点
for each 存活 worker u:
send RestoreFrom(checkpoint_meta) to u
加载 checkpoint 中的分区归属（含刚重分配的映射）
加载该检查点里保存的【顶点值、边值、收件箱消息、活跃标志】
superstep := checkpoint_meta.superstep # 全局回滚到该超级步
# 3) 从该超级步继续执行（重算若干个超级步）
wasted := 失败发生的超级步 - checkpoint_meta.superstep
记录 wasted（用于衡量恢复代价）
=== WORKER w：检查点内容 ===
对分区 P_w 中的每个顶点 v，持久化：
(v.id, v.value, v.edges 的完整边表, v.active, v.pending_halt, v.inbox 中尚未处理的消息)
以及本 worker 当前的聚合器贡献值（供跨检查点的全局判定使用）
算法逻辑解说
- 为什么会泄漏状态就恢复不对：如果把”收件箱里的消息”漏掉不存，那么重放时一个本该被唤醒的顶点会保持停机，于是它的邻居永远收不到它的消息 ⇒ 最终结果与无故障运行不同。第 24.4 节的引擎把
value / edges / active / pending_halt / inbox五样一起存档，并让整个系统（不只失败的 worker）一起回滚，从而得到与无故障运行逐位相同的结果（实验输出fault-free = crashed，断言通过）。 - 为什么”整个系统回滚”而不是”只重算丢失的分区”：因为一个顶点的输入依赖其他分区发来的消息，只重算丢失分区会缺少”上游消息的历史”。Pregel 的默认做法是全系统回滚到检查点（简单、正确，代价是重算）；confined recovery（补充说明，论文口径）则通过只重算”被丢失分区所需要的那部分消息链”来减少重算量。
- 重算量与检查点间隔的关系：若每 $N_{ckpt}$ 步做一次检查点，则故障时最多重算 $N_{ckpt}$ 个超级步——实验中
checkpoint_every=2、故障发生在超级步 5、最近检查点在超级步 4 ⇒ 重算 1 个超级步（输出中的recomputed_supersteps=1）。
正确性论证
- 安全性（恢复后结果正确）：命题：从检查点 $C$ 恢复并重新执行，最终结果与无故障执行结果相同。
- 证明：由 24.3.1 的确定性定理，全局状态序列 $S_0, S_1, S_2,\dots$ 是唯一确定的。检查点 $C$ 保存了 $S_{k}$ 的完整信息（顶点值、边值、收件箱、活跃与停机标志、分区映射、聚合值）。执行从 $S_k$ 继续时，每个超级步的输入与无故障执行时逐项相同 ⇒ 由归纳，之后的每个 $S_{k+t}$ 都与无故障执行相同 ⇒ 最终状态相同。分区重分配只改变了”谁算”，不改变”算什么”，因此不影响状态序列。$\square$
- 依赖假设：消息不丢失（否则 $M_{k+1}(v)$ 不同）；用户函数确定性（否则重算得到不同结果）；检查点的内容完整（漏消息就会破坏”输入相同”这一条）。
- 活性（恢复必然完成）：只要存活 worker 数 $\ge 1$，重分配总能完成（把失败者的分区分给存活者）；只要故障总数小于系统能承受的规模（分片与副本设计决定），系统就能继续推进。极端情况下只剩 1 台 worker 时系统仍能完成作业，只是退化为单机执行（graceful degradation）。
- 代价的正确性（检查点频率 vs 重算代价）：设单个超级步成本为 $c$，检查点一次成本为 $C_{ckpt}$，故障率为 $\lambda$（每步失败概率），则期望总成本约为 \(T \approx K c + \frac{K}{N_{ckpt}}C_{ckpt} + \lambda K \cdot \frac{N_{ckpt}}{2} c\) 对 $N_{ckpt}$ 求导得最优间隔 $N_{ckpt}^* = \sqrt{\dfrac{2 C_{ckpt}}{\lambda c}}$——检查点越贵、故障越稀，间隔应越大。这就是 24.2.6 表格里”$N$ 中等（如 5-10）是常见工程折中”的定量依据。
复杂度
- 检查点开销：每 $N_{ckpt}$ 步一次 $O(\vert V\vert +\vert E\vert +\vert \text{inbox}\vert )$ 的持久化写，均摊到每步为 $O((\vert V\vert +\vert E\vert )/N_{ckpt})$。
- 恢复开销：重算 $\le N_{ckpt}$ 个超级步 + 分区迁移（最多 $O(\vert P_w\vert )$ 个顶点的状态迁移）。
- 容错能力：可容忍任意多台 worker 崩溃（只要还剩存活机器），但每台崩溃都会引入重算；连续崩溃可能让重算不断叠加（实践中通过”恢复期间不再做检查点、恢复完成后再打检查点”来避免抖动）。
算法 24.3.3：PageRank 的 Pregel 实现
假设与系统模型
- 有向图 $G=(V,E)$，$N=\vert V\vert $；无悬挂顶点（每个顶点的出度 $\ge 1$；若有悬挂顶点，需要额外把它们的权重按 $1/N$ 重新分配，否则权重会”漏出”图外——这是 PageRank 的经典细节）。
- 阻尼系数 $d=0.85$；同步 BSP；消息可靠；用户函数确定性。
伪代码
# 每个顶点 v 的状态：value = PR(v)（初始 1/N）；出边表 Out(v)
# 顶点约定：把"我对每个出邻居的贡献" value/|Out(v)| 发出去
Compute(v, msgs, superstep, aggregated):
if superstep == 0:
v.value := 1.0 / N # 初始均匀分布
else:
total := sum(msgs) # 收集所有入邻居的贡献
new := (1 - d) / N + d * total # PageRank 公式
Aggregate("max_delta", |new - v.value|) # 上报本顶点的残差
v.value := new
# ---- 【关键】用上一步的【全局】最大残差决定是否停机（而不是只看自己）----
if aggregated["max_delta"] < eps: # 全局残差已足够小
VoteToHalt() # 全体同时停机
return
share := v.value / |Out(v)|
for each u in Out(v):
SendMessage(u, share)
算法逻辑解说（含 4 顶点小图的第一轮、第二轮完整数值推演）
- 公式：$\text{PR}(v) = \dfrac{1-d}{N} + d \sum_{u \in \text{InNeighbors}(v)} \dfrac{\text{PR}(u)}{\vert \text{OutNeighbors}(u)\vert }$。第一项 $(1-d)/N$ 是”随机跳转”的底噪（保证每个顶点至少分到一点权重，也保证迭代是压缩的），第二项是”沿链接传递”的权重。
- 消息形式：顶点 $u$ 把自己的一份 $u.\text{value}/\vert \text{Out}(u)\vert $ 发给每个出邻居（所以一个顶点的总输出正好等于自己的 PR 值，权重守恒）。目标顶点把收到的所有消息求和，乘以 $d$，加上 $(1-d)/N$。
- 收敛判定（Aggregator 的典型用法）：每个顶点在
Compute()里把自己的变化量 $\vert \text{new}-\text{old}\vert $ 上报给max_delta聚合器；系统把它全局归约成最大值，并在下一个超级步发给所有顶点。所有顶点读到同一个值，于是在同一个超级步集体VoteToHalt()⇒ 全局终止。这样既避免了”每个顶点各自停机”造成的权重泄漏（见 24.2.5 的实测对照），也让停机决策全局一致。 - 小图推演（4 个顶点，$N=4$，$d=0.85$，$(1-d)/N = 0.0375$）：边集 $1\to2,\;1\to4,\;2\to1,\;2\to3,\;3\to1,\;4\to1$，因此 $\vert \text{Out}(1)\vert =\vert \text{Out}(2)\vert =2$，$\vert \text{Out}(3)\vert =\vert \text{Out}(4)\vert =1$；入邻居为 $\text{In}(1)=\{2,3,4\},\ \text{In}(2)=\{1\},\ \text{In}(3)=\{2\},\ \text{In}(4)=\{1\}$。
超级步 0（初始化 + 发出贡献）：
所有顶点 value = 1/4 = 0.25
贡献：v1 -> 0.25/2 = 0.125 给 v2, v4 ；v2 -> 0.125 给 v1, v3
v3 -> 0.25/1 = 0.25 给 v1 ；v4 -> 0.25 给 v1
超级步 1（收消息 -> 计算 -> 再发出）：
v1 = 0.0375 + 0.85*(0.125 + 0.25 + 0.25) = 0.0375 + 0.85*0.625 = 0.568750
v2 = 0.0375 + 0.85*(0.125) = 0.0375 + 0.106250 = 0.143750
v3 = 0.0375 + 0.85*(0.125) = 0.0375 + 0.106250 = 0.143750
v4 = 0.0375 + 0.85*(0.125) = 0.0375 + 0.106250 = 0.143750
校验：0.568750 + 3*0.143750 = 1.000000 （权重守恒！）
新贡献：v1 -> 0.284375 给 v2,v4 ；v2 -> 0.071875 给 v1,v3
v3 -> 0.143750 给 v1 ；v4 -> 0.143750 给 v1
超级步 2：
v1 = 0.0375 + 0.85*(0.071875 + 0.143750 + 0.143750) = 0.342969
v2 = 0.0375 + 0.85*(0.284375) = 0.279219
v3 = 0.0375 + 0.85*(0.071875) = 0.098594
v4 = 0.0375 + 0.85*(0.284375) = 0.279219
校验：0.342969 + 0.279219 + 0.098594 + 0.279219 = 1.000001 ✓
超级步 3： v = (0.477309, 0.183262, 0.156168, 0.183262)
超级步 4： v = (0.403901, 0.240356, 0.115386, 0.240356)
超级步 5： v = (0.442032, 0.209158, 0.139651, 0.209158) <-- 注意它是【震荡收敛】的
要点：$v_1$ 在前几轮里 0.5688 → 0.3430 → 0.4773 → 0.4039 → 0.4420，上下摆动但摆幅按 $d$ 缩小——这正是”压缩映射”的可视化。第 24.4 节的 7 顶点实验最终收敛到 $\text{sum}=1.000000$，与 200 轮 Jacobi 精确解的最大误差 $1.24\times10^{-4}$。
正确性论证
- 收敛性（PageRank 迭代是压缩映射）：把迭代写成矩阵形式 $\mathbf{x}^{(k+1)} = \frac{1-d}{N}\mathbf{1} + d\,P^{\top}\mathbf{x}^{(k)}$，其中 $P$ 是行随机矩阵（$P_{uv}=1/\vert \text{Out}(u)\vert $ 当 $u\to v$）。定义映射 $T(\mathbf{x}) = \frac{1-d}{N}\mathbf{1}+dP^{\top}\mathbf{x}$。对任意两个向量 $\mathbf{x},\mathbf{y}$：$\vert T(\mathbf{x})-T(\mathbf{y})\vert _1 = d\vert P^{\top}(\mathbf{x}-\mathbf{y})\vert _1 \le d\vert \mathbf{x}-\mathbf{y}\vert _1$（因为 $P^{\top}$ 的每一列之和为 1，$L_1$ 范数不被放大）。由于 $d=0.85<1$，$T$ 是 $L_1$ 范数下的压缩映射，压缩因子为 $d$。由 Banach 不动点定理，$T$ 有唯一不动点 $\mathbf{x}^$（即 PageRank 向量），且从任意初始点出发，$\vert \mathbf{x}^{(k)}-\mathbf{x}^\vert _1 \le d^k\vert \mathbf{x}^{(0)}-\mathbf{x}^*\vert _1$ ⇒ 线性收敛，且每轮误差至少乘以 $0.85$。若要求 $\varepsilon$ 精度，需要 $K = O\!\left(\frac{\log(1/\varepsilon)}{\log(1/d)}\right) = O\!\left(\frac{\log(1/\varepsilon)}{1-d}\right)$ 轮（$d=0.85$ 时 $K=O(6.15\ln(1/\varepsilon))$，即「每约 6 轮把误差缩小为 $e$ 分之一」，缩小一个数量级约需 14 轮——这是最坏情况的界）。实测比这个界更快：$L_1$ 范数下的界 $d$ 是保守估计，实际收敛率由 $d\cdot\vert \lambda_2(P)\vert $ 决定；本实验的实测衰减因子约 $0.63$（残差 $2.4\times10^{-1}\to7.5\times10^{-4}$ 只用了 9 轮，约 2.5 个数量级）。$\square$
- 权重守恒（安全性）：命题：若没有悬挂顶点且所有顶点在同一个超级步同时停机，则每个超级步结束时 $\sum_v \text{PR}(v)=1$。
- 证明：$\sum_v \text{PR}^{(k+1)}(v) = \sum_v\left[\frac{1-d}{N} + d\sum_{u\to v}\frac{\text{PR}^{(k)}(u)}{\vert \text{Out}(u)\vert }\right] = (1-d) + d\sum_u \text{PR}^{(k)}(u)\cdot\frac{\vert \text{Out}(u)\vert }{\vert \text{Out}(u)\vert } = (1-d)+d\cdot 1 = 1$。归纳即得。$\square$
- 反例（这个证明依赖的假设被破坏时会怎样）：若某个顶点单独停机而不再发送贡献，则上式中的 $\sum_u$ 会少掉一项，守恒被破坏——这正是 24.2.5 与 24.4 中”局部停机版 PageRank 的 $\sum\text{PR}=0.8136$”的原因。这个对比是本章最值得记住的”安全性与全局判据”案例。
- 终止性：因为残差按 $d$ 衰减到 $\varepsilon$ 以下（收敛性论证），且判据是全局的（所有顶点同时满足 $\text{max\_delta}<\varepsilon$），所以终止条件是”在有限轮之后必然成立”，而不是”可能永远不成立”。$\square$
复杂度
- 轮次：$K=O\!\left(\frac{\log(1/\varepsilon)}{1-d}\right)$；$d=0.85,\varepsilon=10^{-3}$ ⇒ 约 12 轮（实测 12 个超级步）。
- 消息：每轮每个活跃顶点发出 $\vert \text{Out}(v)\vert $ 条消息 ⇒ 每轮 $O(\vert E\vert )$ 条；$K$ 轮总计 $O(K\vert E\vert )$（实测 7 顶点 12 轮 = 132 条原始消息；开启
combine=sum后实际投递 110 条，见 24.3.5）。 - 每轮每顶点的计算：$O(\text{入度})$（把入消息求和），总计算量 $O(K\vert E\vert )$。
- 空间：每个顶点 $O(\vert \text{Out}(v)\vert )$ 存边 + $O(1)$ 存值。
算法 24.3.4：单源最短路（SSSP）的 Pregel 实现
假设与系统模型
- 有向带权图，边权全为正（或至少无负环，从而最短路良定义）；单源 $s$；所有顶点初值 $+\infty$，源点为 0；同步 BSP。
- 算法本质：Bellman-Ford 的并行版本——第 $k$ 轮结束时，所有”最短路含 $\le k$ 条边”的顶点已经拿到正确距离。
伪代码
Compute(v, msgs, superstep, aggregated):
if superstep == 0:
v.value := (v.id == source) ? 0 : +inf
if v.value < +inf: # 源点在第 0 轮就广播自己的距离
for each (u, w) in Out(v): SendMessage(u, v.value + w)
else:
VoteToHalt() # 其他顶点先停机，等消息唤醒
return
best := min(msgs) # 收到若干"候选距离"，取最小
if best < v.value: # 【只有严格变小时才传播】
v.value := best
for each (u, w) in Out(v):
SendMessage(u, v.value + w) # 把自己的新距离 + 边权发出去
else:
VoteToHalt() # 没有改进 -> 本轮停机
Aggregate("reachable", (v.value < +inf) ? 1 : 0) # 统计可达顶点数（全局统计）
算法逻辑解说（含 6 顶点小图的完整多轮松弛推演）
为什么需要”投票停机 + 被唤醒”：一个顶点可能很早就收到一条”能到达”的路径（例如 $1\to5$ 权重 10），于是它算出距离 10 并停机（因为本轮它变了，但下一轮没有消息时它会停机）；几轮之后，一条更短的路径（$1\to2\to3\to4\to6\to5$，总权重 $1+1+2+1+1=6$）才沿着链传播过来——此时必须把它唤醒，否则它永远停在 10 这个错误答案上。这就是 Pregel 终止条件里”收到消息就重新变活跃”不可省略的原因。
实测推演（边集：$1\to2(1), 1\to3(5), 1\to5(10), 2\to3(1), 3\to4(2), 4\to6(1), 6\to5(1)$；源点 1；下表是每个超级步结束时的距离，
inf表示尚不可达）：
超级步 k | d(1) d(2) d(3) d(4) d(5) d(6) | 说明
---------+-------------------------------------+--------------------------------
0 | 0.0 inf inf inf inf inf | 源点广播: 1->2:1, 1->3:5, 1->5:10
1 | 0.0 1.0 5.0 inf 10.0 inf | v2,v3,v5 拿到第一版距离（v4,v6 仍停机）
2 | 0.0 1.0 2.0 7.0 10.0 inf | v3 被 v2 改进为 2；v4 从不 inf 变为 7
3 | 0.0 1.0 2.0 4.0 10.0 8.0 | v4 被改进为 4；v6 被唤醒 -> 8
4 | 0.0 1.0 2.0 4.0 9.0 5.0 | v5 被唤醒 10 -> 9；v6 改进 8 -> 5
5 | 0.0 1.0 2.0 4.0 6.0 5.0 | v5 再次被唤醒 9 -> 6 = 真最短距离 ✓
6 | 0.0 1.0 2.0 4.0 6.0 5.0 | 没有顶点活跃、没有消息在途 -> 全局终止
- 引擎实际打印的唤醒日志（第 24.4 节代码的真实输出）：
[wake-up] superstep 1: vertex 4 had halted, woken by message 7.0
[wake-up] superstep 2: vertex 6 had halted, woken by message 8.0
[wake-up] superstep 3: vertex 5 had halted, woken by message 9.0
final distances = {1: 0, 2: 1, 3: 2, 4: 4, 5: 6, 6: 5} == Bellman-Ford ground truth
supersteps=7 wakeups=3 raw_msgs=10
注意 v5 的完整生命周期：第 1 轮拿到 10 → 第 2、3 轮没有任何消息（连续两轮不变）→ 在超级步 3 的投递阶段被一条 9 的消息唤醒 → 第 4 轮更新为 9 → 再被 6 唤醒 → 第 5 轮得到正确的 6。“停机 → 被唤醒 → 再次停机 → 再次被唤醒” 是 Pregel 终止机制最精髓的表现。
- 消息量为什么这么小（只有 10 条）：因为”只有距离严格变小时才发送”这一剪枝让绝大多数顶点在绝大多数轮里都是停机的。这正对应 24.2.7 里”delta-based 计算 + 剪枝”两条加速思想。
正确性论证
- 安全性（停机时每个顶点都持有正确的最短距离）：
- 命题：终止时，对每个可达顶点 $v$，$v.\text{value} = \delta(s,v)$（真实最短距离）；对不可达顶点，值为 $+\infty$。
- 证明（两个方向）： (i) 值不会小于真实距离（下界性）：用归纳证明”任何时刻 $v.\text{value} \ge \delta(s,v)$”。初始时源点 $0=\delta(s,s)$、其他顶点 $+\infty\ge\delta$。若某轮 $v$ 更新为 $\text{best}=\min_u(u.\text{value}+w(u,v))$，由归纳 $u.\text{value}\ge\delta(s,u)$，故 $\text{best} \ge \min_u(\delta(s,u)+w(u,v)) \ge \delta(s,v)$（最短路的最优子结构：任何经由 $u$ 的路径长度 $\ge$ 最短距离）。$\square$ (ii) 值最终等于真实距离（可达性）：设 $s=v_0\to v_1\to\dots\to v_k=v$ 是一条最短路径（$k$ 条边，$k\le \vert V\vert -1$）。对 $k$ 归纳证明”在第 $k$ 个超级步结束时 $v_k.\text{value}=\delta(s,v_k)$”：$k=0$ 显然。若第 $k-1$ 步末 $v_{k-1}$ 已是正确值，那么在下一步它必然发送（因为它是新更新的，或者按我们的算法”距离严格变小就发送”），消息在第 $k$ 步被 $v_k$ 收到并被采用（因为它 $\le$ 当时的候选值，而由 (i) 它又是最小可能的），故第 $k$ 步末 $v_k$ 一定取到 $\delta(s,v_k)$。由于最短路最多 $\vert V\vert -1$ 条边，$\vert V\vert -1$ 轮内所有可达顶点都拿到正确答案。$\square$
- 依赖假设：正权（用 Bellman-Ford 而非 Dijkstra 的语义）；消息不丢失（否则某条改进消息永远到不了，顶点会停在旧值并停机——这正是”消息可靠”假设的重要性）。
- 活性（最终必然终止）：定义势函数 $\Phi=\sum_{v:\,v.\text{value}<\infty} v.\text{value}$。每次更新都让 $\Phi$ 严格减小（更新条件是 $\text{best}<v.\text{value}$，且 $\text{best}\ge 0$）；而 $\Phi$ 的取值来自”有限条简单路径的长度之和”，只有有限多个可能值，且每次至少减少最小的正边权（假设为正）⇒ 更新次数有限 ⇒ 有限轮之后没有顶点再变化 ⇒ 每轮结束时活跃集合清空 ⇒ 终止。$\square$（这正是 24.3.1 终止性论证的一个实例。）
- 与 Dijkstra 的对比（为什么会考）：Pregel 的 SSSP 是 Bellman-Ford 式的（$O(\vert V\vert )$ 轮、每轮 $O(\vert E\vert )$ 消息），而 Dijkstra 需要全局优先队列（每次取全局最小距离的顶点），这在分布式环境下是全局同步/串行的瓶颈。异步系统（如 GraphLab）可以用”优先级调度”逼近 Dijkstra 的效率（24.2.7 的收敛加速思想第 1 条）。
复杂度
- 轮次：$O(D)$，$D$ 为源点到最远顶点的最短路径边数（$D\le\vert V\vert -1$）；实测 6 顶点图为 6 轮（含终止判定轮）。
- 消息：最坏 $O(\vert V\vert \cdot\vert E\vert )$（每轮所有边都可能发消息），但有剪枝时通常接近 $O(\vert E\vert )$（实测 10 条消息）。
- 空间：$O(\vert V\vert +\vert E\vert )$。
- 讲义给出的真实性能数据（这是讲义原文的数字，务必记住）：在10 亿顶点的树上跑 SSSP，50 个 worker 用 180 秒，800 个 worker 用 20 秒；500 亿顶点的图在 800 个 worker 上用 700 秒（约 12 分钟）——讲义评价 “Pretty Fast!”。注意 50→800 worker 只加速 9 倍（16 倍机器），这是扩展效率次线性的典型例子（见 24.5）。
算法 24.3.5：Combine 与 Aggregator 的正确性
假设与系统模型
combine: V×V -> V是消息值集合上的二元运算；agg_reduce是聚合值上的二元运算。- 二者都要求满足交换律（$a\oplus b=b\oplus a$）与结合律（$(a\oplus b)\oplus c=a\oplus(b\oplus c)$）——即构成一个交换半群（commutative semigroup）。
- 消息可靠、BSP 屏障、用户函数确定性（同 24.3.1），但不要求 FIFO。
伪代码
=== Combiner（本地、每个 worker、每个超级步结束时）===
upon 超级步 k 结束（worker w 的 outbox 中已有本步产生的全部消息）:
groups := 按目标顶点把 outbox 分组 # groups[t] = [m1, m2, ..., mp]
for each target t:
if combine 已定义 and |groups[t]| > 1:
acc := groups[t][0]
for i in 2..p: acc := combine(acc, groups[t][i]) # 折叠
send_to_master(t, acc) # 只发 1 条
else:
for each m in groups[t]: send_to_master(t, m)
# 统计：raw_messages += p ；delivered_messages += (combine ? 1 : p)
=== Aggregator（全局、跨 worker，由 master 在屏障处归约）===
# 提供阶段（超级步 k，各 worker 内部）
upon Aggregate(name, value): w.aggregators[name].append(value)
# 归约阶段（超级步 k 的屏障之后，由 master 执行）
aggregated := {}
for each worker w:
for each (name, vals) in w.aggregators:
local := agg_reduce(vals) # 同一 worker 内部先归约
aggregated[name] := (name in aggregated) ? agg_reduce(aggregated[name], local) : local
# 读取阶段（超级步 k+1）：任何顶点调用 GetAggregatedValue(name) 得到同一个全局值
算法逻辑解说
- Combiner 的收益取决于”同一目标的消息是否落在同一台机器上”：实验（12 个叶子指向同一个 hub 的 fan-in 图）显示：1 个 worker 时，hub 收到的消息从 12 条降到 1 条（整体 24 → 13，1.85×）；4 个 worker 时只降到 4 条（整体 24 → 16，1.50×），因为每个 worker 只能合并”自己 outbox 里”的消息。结论：combine 是本地预聚合，它的收益随”同目标消息的局部性”与”分区质量”变化——这与 Lecture 5 里 MapReduce combiner 只能减少同一 Map 任务内的 shuffle 数据量是同一道理。
- Aggregator 的收益是”全局协调”：PageRank 用
max_delta做全局停机判据（24.3.3）；SSSP 用sum(reachable)统计可达顶点数；本讲的实验还演示了一个只用聚合器的作业（先全局求 $\sum\text{PR}$，下一步每个顶点用它算出全局均值 $\bar{\text{PR}}$，再统计有多少顶点高于均值——聚合器上报 2，与逐顶点离线计算的结果完全一致）。 - 时序约定：$k$ 步提供的聚合值，在 $k+1$ 步可读（与消息的时序完全一致），因此不需要额外的同步原语。
正确性论证
- 定理（Combiner 不改变结果）：若 $\oplus$ 满足交换律与结合律，则把同一超级步内发往同一目标顶点的消息集合 $\{m_1,\dots,m_p\}$ 折叠成单条消息 $\bigoplus_{i=1}^p m_i$ 后再投递，接收方看到的值（以及后续所有超级步的状态）与不折叠时完全相同。
- 证明：接收方对消息的处理只有两种可能：(a) 它只关心消息的折叠结果（例如 PageRank 的
sum(msgs)、SSSP 的min(msgs)、连通分量的min）——此时由结合律，任何折叠次序得到同一个值，故结果相同；(b) 它逐条处理消息。对 (b) 需要额外条件：用户函数的逐条处理结果必须与顺序无关（等价于要求”逐条处理的累积效果”也构成交换半群）。因此 Pregel 的正确用法是”要么保证用户函数对消息顺序不敏感，要么使用 combine“；若用户函数对顺序敏感（例如”只处理第一条消息”）而又使用了 combine，结果就可能改变——这就是 combine 必须满足交换律/结合律的工程含义。$\square$ - 附带结论（确定性）：由 24.3.1 的确定性论证，消息集合是确定的多重集；只要折叠结果唯一，”投递到接收方的多重集的归并结果”就唯一，因此确定性与 combine 相容。
- 证明：接收方对消息的处理只有两种可能：(a) 它只关心消息的折叠结果（例如 PageRank 的
- 定理（Aggregator 的全局一致性）：在超级步 $k+1$ 中，所有顶点读到的同名聚合值相同。
- 证明：归约在屏障之后由 master 一次性完成（所有 worker 的贡献都已到达），结果作为一个标量广播给所有 worker 的
StartSuperstep消息；因此所有顶点读到的是同一个标量，与它所在的分区、执行顺序无关。$\square$（这正是 24.3.3 中”全体同时停机”得以成立的原因。）
- 证明：归约在屏障之后由 master 一次性完成（所有 worker 的贡献都已到达），结果作为一个标量广播给所有 worker 的
- 一个必须注意的陷阱：
combine归约掉的信息无法被用户函数恢复。例如若用户想在Compute()里遍历”每个入邻居分别发来的具体值”，就不能用 combine（此时应改为把值编码进一条复合消息，或者干脆不用 combine）。
复杂度
- Combiner：不改变本地计算量（仍是 $p$ 次折叠，$O(p)$），但把跨机消息数从 $p$ 降到 1（最好情况）；代价是折叠本身消耗少量 CPU，且必须满足可折叠性。
- Aggregator：每个超级步一次 $O(N_{\text{workers}})$ 的归约（与屏障同一时机，边际成本很小），空间 $O(\vert \text{aggregators}\vert )$。
算法 24.3.6：参数服务器的 Push/Pull 协议与三种更新模式
假设与系统模型
- $S$ 台 server（参数按 $1/S$ 分片，每片可有副本），$W$ 个 worker（各自持有一部分训练数据与完整模型副本）。
- 通道：worker 与 server 之间是可靠消息通道；server 对每个参数分片的更新是原子的（分片内可以用锁或原子操作）。
- 故障模型：worker 可崩溃（无状态，重启后重新拉取参数即可）；server 可能崩溃（需要副本，本节假设其容错由复制层负责）。
- 模型：目标函数 $F(w)=\frac{1}{N}\sum_{i=1}^{N} f_i(w)$（例如岭回归或逻辑回归），mini-batch SGD。
伪代码
=== SERVER s（持有参数分片 w_s）===
维护：w_s（参数向量分片）、version（单调递增的版本号）、lock_s（分片锁）
upon Pull(worker_id): # 读
reply (w_s 的副本, version)
upon Push(worker_id, grad_s, v_read): # 写
# 模式 A：同步 SGD（server 侧等待 —— 通常由 worker 侧的 barrier 实现）
buffer[worker_id] := (grad_s, v_read)
if |buffer| == W: # 收齐所有 worker
g_avg := mean(buffer 的所有 grad_s) # 【平均梯度】
with lock_s: w_s := w_s - lr * g_avg ; version := version + 1
buffer := {} ; reply Done to all W workers
# 模式 B：异步 SGD（来者不拒）
with lock_s: w_s := w_s - lr * grad_s ; version := version + 1
reply Done to worker_id
# 模式 C：SSP（有界延迟，见 24.3.7 的时钟机制）
if (本 worker 的时钟) - (全局最小时钟) <= s: 按模式 B 更新
else: 把该 worker 标记为"必须等待"（见 24.3.7）
=== WORKER i（模式 A：同步 SGD）===
for k = 0,1,2,...:
(w_i, v) := Pull() # 从所有 server 分片拉全参数，组成完整模型
grad := ∇f_{batch_i(k)}(w_i) # 前向 + 反向（本地计算）
Push(grad, v) # 推梯度
WaitForRoundDone() # 【屏障】等所有 worker 都推完
# 此时 server 已经把平均梯度应用完毕，参数已更新到 version v+1
=== WORKER i（模式 B：异步 SGD）===
for k = 0,1,2,...:
(w_i, v_read) := Pull() # 读的是"当前"参数（可能已被别人改过很多次）
grad := ∇f_{batch_i(k)}(w_i)
Push(grad, v_read) # 【不等待】，立即生效
# staleness = (被应用时的 version) - v_read
算法逻辑解说
- Push/Pull 的三步节奏：worker
pull（拿参数）→ 本地算梯度 →push（交梯度）；server 聚合后更新参数。整个过程中 worker 之间完全不通信（这是参数服务器与 AllReduce 的结构性差别：AllReduce 要求所有 worker 参与一次全局归约，参数服务器则把”通信”集中到 server 层）。 - 参数分片：模型被切成 $S$ 片，worker 的
pull只从每个 server 拉自己需要的那片；因此单机内存上限被突破（$S$ 越大，单 server 内存压力越小），但 server 可能成为带宽瓶颈（所有 worker 都要拉同一片参数 ⇒ 需要多副本 + 缓存 + 压缩，见 24.5）。 - 三种模式的差别只有一个：
Push之后要不要等别人。 同步等所有人（屏障），异步谁先到谁先生效，SSP 等”落后太多就等”。 - 第 24.4 节的实测（4 个 worker，worker 3 慢 4 倍，目标损失 $<0.02$）：
mode 时间 梯度数 通信数 最终损失 平均陈旧度 达标
sync 249.6 192 384 0.01976 0.00 是
async 20.8 52 104 0.01971 0.44 是
ssp(s=3) 52.0 50 100 0.01930 0.60 是
local(H=20) 416.0 320 32 0.01674 0.00 是
注意两点：(a) 同步 SGD 的梯度数更多（192 vs 52），因为同步模式下每轮把 $W$ 个梯度平均后只用一次更新（每个梯度只贡献 $lr/W$ 的位移），而异步模式下每个梯度都直接以完整学习率生效——这就是”异步的有效步长更大、因此必须用更小学习率”的实测体现。(b) Local SGD 的时间最长为 416，但它只用 32 条消息（同步 SGD 用 384 条）：用时间换通信。
正确性论证（同步 vs 异步的收敛性与 staleness 的影响）
- 同步 SGD 的收敛性（等价于大 batch 的 SGD）：设第 $k$ 轮的平均梯度 $\bar g_k=\frac{1}{W}\sum_{i=1}^W \nabla f_{B_i}(w_k)$，其中 $B_i$ 是第 $i$ 个 worker 的 mini-batch。因为所有 worker 用的是同一个 $w_k$（屏障保证），所以 $\bar g_k$ 恰好是”把 $W$ 个 mini-batch 拼成一个大 batch（大小 $W\cdot B$）”的梯度。于是同步 SGD 就是大 batch SGD，其收敛性由经典随机优化理论保证：在 $L$-光滑、梯度有界方差 $\sigma^2$ 的假设下，取步长 $\eta \le 1/L$，有 $\min_{k\le K}\mathbb{E}\vert \nabla F(w_k)\vert ^2 \le \dfrac{2(F(w_0)-F^*)}{\eta K} + \eta L\sigma^2_{\text{big}}$，其中大 batch 的梯度方差 $\sigma^2_{\text{big}} = \sigma^2/(WB)$ 比单 worker 小 $W$ 倍——这就是”同步 SGD 收敛性好、噪声小”的定量来源。代价写在另一项里：每轮的时间 = $\max_i t_i$（最慢 worker），实测同步模式在 straggler 存在时耗时是异步的 12 倍（249.6 vs 20.8），且移除 straggler 后同步耗时立刻从 249.6 降到 62.4（×4.00 的减速比，与 straggler 的 4 倍慢完全吻合——因为屏障把它的代价乘上了轮数）。
- 异步 SGD 的 staleness 与梯度误差：worker $i$ 在第 $t$ 次更新时用的是 $\nabla f_{B_i}(w_{\tau})$，其中 $\tau \le t$ 是它读参数时的版本；而这次更新作用在 $w_t$ 上。定义陈旧度 $\tau_s = t-\tau$。由中值定理/Lipschitz 光滑性： \(\nabla f(w_\tau) = \nabla f(w_t) + H_i\,(w_\tau - w_t),\qquad \vert w_\tau-w_t\vert \le \sum_{j=\tau}^{t-1}\eta\vert g_j\vert \approx \eta\,\tau_s\,\bar G\) 于是梯度误差 $\approx \vert H\vert \cdot\eta\cdot\tau_s\cdot\bar G$，即正比于陈旧度 $\tau_s$ 与学习率 $\eta$。把它代进 SGD 的收敛分析，等价于把噪声方差放大了 $O(1+\eta^2\tau_s^2)$ 倍；要让”噪声项”不超过误差项，需要 $\eta = O\!\left(\frac{1}{\tau_s}\right)$（学习率必须随最大陈旧度反比缩小），否则迭代可能不收敛。 实测证据：把学习率升到 0.22，同步 SGD 正常收敛到 0.0181，而异步 SGD 发散到 $1.5\times10^{7}$，此时它的最大陈旧度达到 13（”梯度是在 13 个版本之前的参数上算出来的”）。而 SSP（$\tau_s\le 10$）在同一学习率下依然收敛（0.0182）——这就是”有界陈旧度保收敛”的实验证明。$\square$
- 注意区分两种”异步的不确定性”：(1) 更新顺序的不确定性（谁先 push 谁先生效）——它让结果不可复现；(2) staleness 引入的偏差——它让收敛变差。前者是”工程可复现性”问题，后者是”数学收敛性”问题，二者常同时出现但根因不同。
复杂度
- 通信量：同步与异步都是每 worker 每迭代 $O(\vert w\vert )$ 的 pull + push（$\vert w\vert $ 为模型大小）；AllReduce 方式则为每轮 $O(\vert w\vert \cdot W)$（或 Ring AllReduce 的 $2\vert w\vert (W-1)/W$，见 24.5）。
- 同步的每轮墙钟时间：$\max_i (t_i^{\text{comp}} + t_i^{\text{comm}})$ ⇒ 受最慢 worker 支配。
- 异步的吞吐：$\approx \sum_i \frac{1}{t_i^{\text{comp}}+t_i^{\text{comm}}}$（每台机器独立产出），因此吞吐近似与机器数成正比，但收敛所需迭代数变多、且可能不收敛。
- 空间：worker 侧 $O(\vert w\vert )$；server 侧 $O(\vert w\vert /S)$ × 副本数。
算法 24.3.7：SSP 的时钟机制与 staleness bound
假设与系统模型
- $W$ 个 worker，每个维护本地时钟 $c_i$（已完成并生效的迭代次数）；系统维护全局最小时钟 $c_{\min}=\min_j c_j$。
- 参数 $s$（staleness bound，非负整数或 $+\infty$）。
- 不变量（SSP 的核心）：任何 worker $i$ 在执行第 $k$ 次迭代时，满足 $c_i - c_{\min} \le s$，即任何一次被应用的梯度，其陈旧度不超过 $s$。
- 时钟只在迭代完成（push 生效）时递增；读取参数不改变时钟。
伪代码
=== 每个 worker i 的本地状态 ===
c_i : 本地时钟（已完成迭代数），初始 0
blocked : 是否正在等待
cand_update : 待提交的 (grad, v_read)
=== WORKER i 主循环 ===
loop:
# ---- ① 进入"可执行"状态前先检查速度限制 ----
WaitUntil( c_i - min_j c_j <= s ) # 超过 s 就阻塞，直到最慢者推进
(w, v_read) := Pull() # 读参数（记录读取时的版本）
g := ∇f_{batch}(w)
Push(g, v_read) # 提交给 server，立即生效（单条更新）
c_i := c_i + 1 # 时钟前进
# 注：c_min 由所有 worker 的时钟最小值决定，需要一个轻量的全局视图
# （实践中由 server 附带返回 min_j c_j，或由每个 worker 定期广播自己的 c_i）
=== SERVER 侧（维护全局最小时钟）===
维护 c[1..W]（每个 worker 最近上报的时钟）
upon 收到 worker i 的时钟上报: c[i] := 新值
返回 min(c) 给所有正在等待的 worker（或周期性广播）
算法逻辑解说（含 $s=0$ 与 $s=\infty$ 的退化分析）
- $s=0$：条件 $c_i-c_{\min}\le 0$ 意味着所有 worker 的时钟必须相等才能继续——任何 worker 都必须等到所有 worker 都完成本轮才能进入下一轮 ⇒ 轮次结构退化为 BSP/同步 SGD（实测：$s=0$ 的平均与最大陈旧度都是 0，即没有任何梯度被应用到比它读取时更新的参数上；时间 62.4，明显慢于 $s=10$ 的 26.0）。注意一个实现细节：如果 server 对每次 push 立即单独应用（本实现如此），那么即使 $s=0$，它的更新规则也不是”平均梯度”，而是”每条梯度用完整学习率依次生效”——这与 24.3.6 中”屏障 + 平均梯度”的同步 SGD 在更新规则上仍不同。所以严格的说法是：$s=0$ 退化到同步的调度结构（无陈旧度），而不是逐位等价于”平均梯度”式同步 SGD。
- $s=\infty$：条件恒成立 ⇒ 退化为完全异步 SGD（实测：最大陈旧度 9，时间 20.8，最快）。
- $s$ 的实测权衡（同一学习率 0.05、同一 straggler 配置）：
bound s 时间 梯度数 平均陈旧度 最大陈旧度 最终损失
s=0 62.4 50 0.00 0 0.01879 <- 结构上等价于同步
s=1 57.2 50 0.22 1 0.01848
s=3 52.0 50 0.60 3 0.01930 <- 常用折中
s=10 26.0 52 0.58 8 0.01970
s=inf 20.8 52 0.44 9 0.01971 <- 完全异步
读法：时间从 62.4 单调降到 20.8（3 倍提升），而最大陈旧度从 0 升到 9。$s$ 就是”速度 ↔ 陈旧度”这一个旋钮（这也是本章黄金法则的具体化）。
- 实现要点（工程上最容易踩坑的地方）：
- 等待必须是”阻塞而不消耗算力”：被阻塞的 worker 不应该继续拉参数或空转（否则会把 server 带宽吃满）。
min_j c_j必须是全局一致的：如果每个 worker 各自维护一份可能过期的 $c_{\min}$，就会出现”两个 worker 都以为自己没超限”的情况——实践中由 server 权威维护并随Pull回复捎带（与 Lecture 6 的心跳”捎带信息”技巧一致）。- 死锁不可能发生：时钟最小的那个 worker 永远满足 $c_i-c_{\min}=0\le s$（$s\ge0$），因此总有一个 worker 能推进，$c_{\min}$ 会不断增加 ⇒ 系统不会活锁。这是 SSP 设计中必须验证的不变量（本讲的实现里也显式利用了这条性质：卡住时只需唤醒时钟等于 $\min$ 的那些 worker）。
- 掉队/崩溃的 worker 会拖住所有人：因为 $c_{\min}$ 由最慢者决定，一个永久卡住的 worker 会让所有领先者停下 ⇒ 实践中必须配”踢掉超时 worker“（把它从时钟集合里移除，类似弹性训练）。
正确性论证
- 不变量（安全性）：命题：SSP 下任何被应用的梯度 $\nabla f(w_\tau)$ 满足 $t-\tau\le s$。
- 证明：worker $i$ 读参数时的版本为 $\tau$，读完参数后它的本地时钟至少为 $\tau$（因为它必须完成 $\tau$ 次迭代才会读到版本 $\tau$ 的参数——在”每次迭代恰好使版本 +1”的模型下 $c_i \ge \tau$）。由 SSP 的门禁条件 $c_i-c_{\min}\le s$，且被应用的时刻 $t$ 满足 $t$ 等于当时某个合法的全局进度（所有 $c_j \le c_{\min}+s$，且实际生效的版本不超过任何 worker 的”允许区间”）⇒ $t \le c_{\min}+s \le \tau+s$。故 $\tau_s=t-\tau\le s$。$\square$
- 依赖假设：时钟与实际生效版本保持一致（即”一次 push 恰好推进一个版本”，不能出现”一次 push 被应用到多个版本”）；worker 上报的时钟真实。
- 收敛性（有界陈旧度 ⇒ 有界梯度误差 ⇒ 收敛可保证）：由 24.3.6 的误差公式，梯度误差 $\vert \nabla f(w_\tau)-\nabla f(w_t)\vert \le L\eta\,\tau_s\,\bar G \le L\eta\,s\,\bar G$。把它代入 SGD 的收敛递推，得到”有效噪声方差” $\sigma^2{\text{eff}} = \sigma^2 + O(L^2\eta^2s^2\bar G^2)$。取 $\eta = O\!\left(\frac{1}{\sqrt{K}}\right)$ 且要求 $\eta\,s = O(1)$（即 $s$ 与 $1/\eta$ 同阶），则 $\min{k\le K}\mathbb{E}\vert \nabla F(w_k)\vert ^2 \to 0$。反之，若 $s=\infty$，$\tau_s$ 无界，误差项无界，收敛性无法保证——这正是”$s=\infty$ 会发散、$s$ 有限则收敛”的实验现象（学习率 0.22 时：$s=1,3,10$ 都收敛，$s=\infty$ 发散）。$\square$
- 活性（无饥饿、无死锁）：由上面第 3 条实现要点：时钟最小者永远可推进 ⇒ 系统持续推进；又由于 $s$ 有限时每个 worker 的”领先额度”有限，每个 worker 在等待有限时间后必然被放行（因为 $c_{\min}$ 单调不减且最终会追上）。$\square$
复杂度
- 通信：除参数传输外，额外需要传播 $c_{\min}$（每个超级步/迭代 $O(W)$ 的小消息，或由 server 捎带）。
- 时间：在理想情况下（无 straggler、$s$ 足够大）达到异步的吞吐 $\approx\sum_i 1/t_i$；在最坏情况下（有 straggler 且 $s$ 很小）退化为同步的 $\max_i t_i$ 每轮。SSP 的价值就是”在这两个极端之间有连续的旋钮”。
- 空间：每个 worker $O(1)$ 的时钟状态，server 侧 $O(W)$。
24.4 代码示例与分布式实现
关于代码长度：撰写规范要求单个代码块 60-160 行，但本节的三份程序都是完整可运行的教学系统（引擎 + 四个算法 + 故障注入 + 实验驱动），因此第一份约 600 行、第二三份各约 270-340 行。三份代码都只用 Python 标准库、固定随机种子、直接
python3 文件名.py即可运行并打印全部实验结果；把代码原样保存成.py文件即可复现本节引用的每一个数字。
24.4.1 完整的 Pregel 引擎（BSP + 唤醒 + Combine + Aggregator + 检查点容错）
"""
pregel.py -- a tiny, deterministic, single-process Pregel (BSP) engine.
Everything runs in ONE Python process: a "worker" is an object, the "network"
is a Python dict, and the bulk-synchronous barrier is an explicit line in
PregelMaster.run(). Sequential simulation (no threads) is deliberate: it makes
the BSP semantics exactly visible:
compute (all workers) --> BARRIER --> combine+deliver --> aggregators --> halt
Rules enforced by the engine (all of them are Pregel rules):
* messages sent during superstep k are received during superstep k+1;
* a vertex that votes to halt stops running, but is WOKEN UP if a message
arrives for it;
* a superstep ends only when every worker has finished (barrier);
* the job ends when no vertex is active and no message is in transit.
Algorithms on top of the engine: PageRank, SSSP, connected components, an
aggregator-only global statistic, plus checkpoint/crash/recovery simulation.
Run: python3 pregel.py
"""
import random
import time
from collections import defaultdict
INF = float("inf")
# --------------------------------------------------------------------------
# Graph data structures
# --------------------------------------------------------------------------
class Vertex(object):
def __init__(self, vid, value=0.0, edges=None):
self.id = vid
self.value = value
self.edges = dict(edges) if edges else {} # target_id -> edge value
self.active = True # in the active set?
self.pending_halt = False # voted to halt this superstep
self.inbox = [] # messages for THIS superstep
class ComputeContext(object):
"""The vertex API handed to the user's compute() function."""
def __init__(self, worker, vertex, step):
self.worker = worker
self.vertex = vertex
self.step = step
self.messages = vertex.inbox
def send(self, target, msg):
"""Buffer a message in the worker's outbox; delivered at superstep end."""
self.worker.outbox.append((target, msg))
def vote_to_halt(self):
self.vertex.pending_halt = True
def aggregate(self, name, value):
self.worker.aggregators[name].append(value)
def get_aggregated(self, name, default=None):
return self.worker.master.aggregated.get(name, default)
def superstep(self):
return self.step
class PregelWorker(object):
def __init__(self, wid):
self.wid = wid
self.master = None
self.vertices = {} # vid -> Vertex (this worker's partition)
self.outbox = [] # (target_vid, msg) buffered this superstep
self.aggregators = defaultdict(list)
def begin_superstep(self):
self.outbox = []
self.aggregators = defaultdict(list)
def compute_phase(self, compute):
"""Run user compute() on every ACTIVE vertex owned by this worker."""
for v in list(self.vertices.values()):
if not v.active:
continue
compute(v, ComputeContext(self, v, self.master.step))
v.inbox = [] # the messages were consumed
self.master.vertex_ops += 1
def flush(self, combine):
"""END OF SUPERSTEP: combine locally, then deliver across the 'network'."""
self.master.raw_messages += len(self.outbox)
groups = defaultdict(list)
for target, msg in self.outbox:
groups[target].append(msg)
for target, msgs in groups.items():
if combine is not None and len(msgs) > 1:
acc = msgs[0]
for m in msgs[1:]:
acc = combine(acc, m)
self.master.deliver(target, acc)
else:
for m in msgs:
self.master.deliver(target, m)
self.outbox = []
# --------------------------------------------------------------------------
# Master
# --------------------------------------------------------------------------
class PregelMaster(object):
def __init__(self, graph, num_workers=3, combine=None, agg_reducers=None,
checkpoint_every=2, crash_at=None, crash_worker=2):
self.graph = graph # vid -> Vertex (canonical objects)
self.num_workers = num_workers
self.combine = combine
self.agg_reducers = agg_reducers or {}
self.checkpoint_every = checkpoint_every
self.crash_at = crash_at # superstep at which a worker dies
self.crash_worker = crash_worker
self.workers = [PregelWorker(i) for i in range(num_workers)]
for w in self.workers:
w.master = self
self.owner = {}
for vid in graph: # default: hash(vid) mod N
wid = vid % num_workers
self.workers[wid].vertices[vid] = graph[vid]
self.owner[vid] = wid
self.step = 0
self.vertex_ops = 0
self.aggregated = {}
self.raw_messages = 0
self.delivered_messages = 0
self.inbound = defaultdict(int) # delivered messages per target
self.wakeups = 0
self.log = []
self.trace = [] # per-superstep statistics
self.value_history = [] # per-superstep snapshot of all values
self.ckpt = None
self.wasted_supersteps = 0
self.recovered = False
# ---------------- message delivery (happens at the end of a superstep) ---
def deliver(self, target, msg):
v = self.graph[target]
v.inbox.append(msg)
self.delivered_messages += 1
self.inbound[target] += 1
if not v.active: # *** wake up a halted vertex ***
v.active = True
self.wakeups += 1
self.log.append(" [wake-up] superstep %d: vertex %s had halted, "
"woken by message %s" % (self.step, target, msg))
v.pending_halt = False # a message always revives a vertex
# ---------------- checkpointing -----------------------------------------
def checkpoint(self):
state = {}
for w in self.workers:
state[w.wid] = {vid: (v.value, dict(v.edges), v.active,
v.pending_halt, list(v.inbox))
for vid, v in w.vertices.items()}
self.ckpt = {"step": self.step, "state": state,
"partition": dict(self.owner),
"aggregated": dict(self.aggregated),
"raw": self.raw_messages,
"delivered": self.delivered_messages,
"wakeups": self.wakeups}
def restore(self, dead_worker=None):
"""Roll the WHOLE system back to the last checkpoint (Pregel semantics)."""
c = self.ckpt
part = dict(c["partition"])
if dead_worker is not None:
survivors = [w for w in range(self.num_workers) if w != dead_worker]
moved = 0
for vid, wid in list(part.items()):
if wid == dead_worker:
part[vid] = survivors[moved % len(survivors)]
moved += 1
print(" [recovery] worker %d died -> %d vertices reassigned to %s"
% (dead_worker, moved, survivors))
for w in self.workers:
w.vertices = {}
for vid, wid in part.items():
self.workers[wid].vertices[vid] = self.graph[vid]
self.owner[vid] = wid
for wid, vs in c["state"].items():
for vid, (val, edges, active, ph, inbox) in vs.items():
v = self.graph[vid]
v.value, v.edges, v.active = val, dict(edges), active
v.pending_halt, v.inbox = ph, list(inbox)
self.aggregated = dict(c["aggregated"])
self.raw_messages, self.delivered_messages = c["raw"], c["delivered"]
self.wakeups = c["wakeups"]
self.step = c["step"]
# ---------------- the superstep loop ------------------------------------
def _reduce_aggregators(self):
new_agg = {}
for w in self.workers:
for name, vals in w.aggregators.items():
red = self.agg_reducers.get(name, max)
local = red(vals)
if name in new_agg:
new_agg[name] = red([new_agg[name], local])
else:
new_agg[name] = local
self.aggregated = new_agg
def run(self, compute, max_supersteps=60):
t0 = time.perf_counter()
crashed = False
while self.step < max_supersteps:
if not any(v.active for v in self.graph.values()):
break # system-wide termination
if self.step % self.checkpoint_every == 0:
self.checkpoint()
self.log.append(" [ckpt] checkpoint at start of superstep %d"
% self.step)
for w in self.workers:
w.begin_superstep()
active_now = sum(1 for v in self.graph.values() if v.active)
sent_before = self.raw_messages
# ---- compute phase: every worker runs its active vertices ----
for w in self.workers:
if (not crashed and self.crash_at is not None
and self.step == self.crash_at
and w.wid == self.crash_worker):
print(" [fault] worker %d crashes in superstep %d (mid-superstep)"
% (w.wid, self.step))
crashed = True
self.wasted_supersteps += self.step - self.ckpt["step"]
self.restore(dead_worker=w.wid)
self.recovered = True
break
w.compute_phase(compute)
else:
# ---- BARRIER: all workers are done with this superstep ----
for w in self.workers:
w.flush(self.combine) # combine + deliver
self._reduce_aggregators() # k-values read at k+1
for v in self.graph.values(): # apply the votes to halt
if v.pending_halt:
v.active = False
v.pending_halt = False
self.value_history.append({vid: self.graph[vid].value
for vid in sorted(self.graph)})
self.trace.append({"step": self.step, "active": active_now,
"sent": self.raw_messages - sent_before,
"delivered": self.delivered_messages,
"agg": dict(self.aggregated)})
self.step += 1
return {"steps": self.step, "raw": self.raw_messages,
"delivered": self.delivered_messages, "wakeups": self.wakeups,
"seconds": time.perf_counter() - t0,
"agg": self.aggregated, "recovered": self.recovered,
"wasted": self.wasted_supersteps,
"vertex_ops": self.vertex_ops}
def print_trace(self, title="superstep trace"):
print(" --- %s ---" % title)
for row in self.trace:
agg = ", ".join("%s=%.6f" % (k, v) for k, v in sorted(row["agg"].items()))
print(" k=%2d active=%2d sent=%4d delivered=%4d agg[%s]"
% (row["step"], row["active"], row["sent"],
row["delivered"], agg))
# --------------------------------------------------------------------------
# Algorithm 1: PageRank
# --------------------------------------------------------------------------
def build_pagerank_graph():
edges = {1: {2: 1, 3: 1, 4: 1, 5: 1, 6: 1, 7: 1}, # vertex 1: out-degree 6
2: {3: 1}, 3: {1: 1}, 4: {1: 1}, 5: {6: 1}, 6: {1: 1}, 7: {5: 1}}
return {vid: Vertex(vid, 0.0, edges[vid]) for vid in edges}
def make_pagerank(n, d=0.85, eps=1e-3):
"""PageRank that stops on a GLOBAL criterion, read from the aggregator.
Every vertex keeps sending its contribution until the whole graph has
converged -- that is what keeps the rank mass conserved (sum(PR) == 1).
"""
def compute(v, ctx):
if ctx.superstep() == 0:
v.value = 1.0 / n
else:
total = sum(ctx.messages) if ctx.messages else 0.0
new = (1.0 - d) / n + d * total
ctx.aggregate("max_delta", abs(new - v.value))
v.value = new
if ctx.get_aggregated("max_delta", 1.0) < eps: # global residual
ctx.vote_to_halt()
return
share = v.value / len(v.edges)
for t in v.edges:
ctx.send(t, share)
return compute
def make_pagerank_local_halt(n, d=0.85, eps=1e-3):
"""The naive variant: each vertex stops as soon as ITS OWN delta is small.
A halted vertex stops contributing, so its neighbours lose rank mass.
Kept here on purpose: it is the classic Pregel PageRank pitfall.
"""
def compute(v, ctx):
if ctx.superstep() == 0:
v.value = 1.0 / n
share = v.value / len(v.edges)
for t in v.edges:
ctx.send(t, share)
return
total = sum(ctx.messages) if ctx.messages else 0.0
new = (1.0 - d) / n + d * total
delta = abs(new - v.value)
v.value = new
ctx.aggregate("max_delta", delta)
if delta > eps:
share = new / len(v.edges)
for t in v.edges:
ctx.send(t, share)
else:
ctx.vote_to_halt()
return compute
def exact_pagerank(graph, d=0.85, iters=200):
"""Reference Jacobi iteration: what a fixed-iteration job would produce."""
n = len(graph)
pr = {vid: 1.0 / n for vid in graph}
for _ in range(iters):
new = {}
for vid in graph:
s = sum(pr[u] / len(graph[u].edges) for u in graph
if vid in graph[u].edges)
new[vid] = (1 - d) / n + d * s
pr = new
return pr
# --------------------------------------------------------------------------
# Algorithm 2: single-source shortest path (Bellman-Ford style)
# --------------------------------------------------------------------------
def build_sssp_graph():
"""Vertex 5 is reachable the expensive way (1->5, w=10) early, then it sits
idle, HALTS, and three supersteps later the cheap path (1->2->3->4->6->5)
wakes it up with a shorter distance -- the point of the demo."""
edges = {1: {2: 1, 3: 5, 5: 10}, 2: {3: 1}, 3: {4: 2},
4: {6: 1}, 5: {}, 6: {5: 1}}
return {vid: Vertex(vid, INF, edges[vid]) for vid in edges}
def make_sssp(source):
def compute(v, ctx):
if ctx.superstep() == 0:
v.value = 0.0 if v.id == source else INF
if v.value < INF:
for t, w in v.edges.items():
ctx.send(t, v.value + w)
else:
ctx.vote_to_halt()
return
best = min(ctx.messages) if ctx.messages else INF
if best < v.value:
v.value = best
for t, w in v.edges.items():
ctx.send(t, v.value + w)
else:
ctx.vote_to_halt()
ctx.aggregate("reachable", 1.0 if v.value < INF else 0.0)
return compute
# --------------------------------------------------------------------------
# Algorithm 3: connected components
# --------------------------------------------------------------------------
def build_cc_graph():
"""Two components plus one isolated vertex: {1,2,3,4}, {5,6}, {7}."""
undirected = [(1, 2), (2, 3), (3, 4), (5, 6)]
ids = [1, 2, 3, 4, 5, 6, 7]
edges = {i: {} for i in ids}
for a, b in undirected:
edges[a][b] = 1
edges[b][a] = 1
return {vid: Vertex(vid, vid, edges[vid]) for vid in ids}
def cc_compute(v, ctx):
best = min([v.value] + list(ctx.messages))
changed = best < v.value
if changed:
v.value = best
if ctx.superstep() == 0 or changed:
for t in v.edges:
ctx.send(t, v.value) # broadcast my component id
else:
ctx.vote_to_halt()
# --------------------------------------------------------------------------
# Algorithm 4: a global statistic computed only with an aggregator
# --------------------------------------------------------------------------
def make_pr_stat(n):
"""Aggregator-only job: stage 1 publishes sum(PR) globally, stage 2 counts
the vertices whose rank is above the (globally known) mean."""
def compute(v, ctx):
if ctx.superstep() == 0:
ctx.aggregate("pr_sum", v.value) # every vertex contributes
return
mean = ctx.get_aggregated("pr_sum", 0.0) / n # global view, next superstep
if v.value > mean:
ctx.aggregate("above_mean", 1.0)
ctx.vote_to_halt()
return compute
# --------------------------------------------------------------------------
# Experiments
# --------------------------------------------------------------------------
def print_history(hist, fmt="%7.5f", head="k "):
print(" %s| %s" % (head, " | ".join("v%-6d" % vid for vid in sorted(hist[0]))))
for k, row in enumerate(hist):
print(" %-2d | " % k + " | ".join(fmt % row[vid] for vid in sorted(row)))
def exp_i_pagerank():
print("=" * 78)
print("(i) PageRank, 7 vertices, d=0.85, eps=1e-3, combine=sum")
print("=" * 78)
g = build_pagerank_graph()
m = PregelMaster(g, num_workers=3, combine=lambda a, b: a + b,
agg_reducers={"max_delta": max}, checkpoint_every=10 ** 9)
res = m.run(make_pagerank(len(g)), max_supersteps=200)
print_history(m.value_history, head="PR ")
ref = exact_pagerank(g)
err = max(abs(ref[vid] - g[vid].value) for vid in g)
print(" supersteps=%d raw_msgs=%d delivered=%d wakeups=%d vertex_ops=%d"
% (res["steps"], res["raw"], res["delivered"], res["wakeups"],
res["vertex_ops"]))
print(" max|PR_pregel - PR_jacobi(200)| = %.2e sum(PR) = %.6f (must be 1)"
% (err, sum(g[vid].value for vid in g)))
print(" active-set size per superstep: %s"
% [row["active"] for row in m.trace])
m.print_trace(title="aggregator trace: global max residual -> global halt")
assert err < 1e-3 and abs(sum(g[vid].value for vid in g) - 1.0) < 1e-9
g2 = build_pagerank_graph()
m2 = PregelMaster(g2, num_workers=3, combine=lambda a, b: a + b,
agg_reducers={"max_delta": max}, checkpoint_every=10 ** 9)
r2 = m2.run(make_pagerank_local_halt(len(g2)), max_supersteps=200)
print(" for contrast, the NAIVE 'halt when my own delta is small' version:")
print(" sum(PR) after %d supersteps = %.6f (mass leaked: %.4f) "
"vertices still active = %d"
% (r2["steps"], sum(g2[vid].value for vid in g2),
1.0 - sum(g2[vid].value for vid in g2),
sum(1 for v in g2.values() if v.active)))
return res
def exp_ii_sssp():
print()
print("=" * 78)
print("(ii) SSSP from vertex 1 -- 'vote to halt' + 'wake up on new message'")
print("=" * 78)
g = build_sssp_graph()
m = PregelMaster(g, num_workers=3, combine=None,
agg_reducers={"reachable": sum}, checkpoint_every=10 ** 9)
res = m.run(make_sssp(1), max_supersteps=40)
print(" distances after every superstep ('inf' = not reached yet):")
print_history(m.value_history, fmt="%7.1f", head="k ")
for line in m.log:
if "[wake-up]" in line:
print(line)
final = {vid: g[vid].value for vid in sorted(g)}
truth = {1: 0.0, 2: 1.0, 3: 2.0, 4: 4.0, 5: 6.0, 6: 5.0}
assert final == truth, (final, truth)
print(" final distances = %s == Bellman-Ford ground truth"
% {k: int(v) for k, v in final.items()})
print(" supersteps=%d wakeups=%d raw_msgs=%d"
% (res["steps"], res["wakeups"], res["raw"]))
return res
def exp_iii_combine():
print()
print("=" * 78)
print("(iii) the combiner: same answer, fewer messages on the wire")
print("=" * 78)
nleaves = 12
def build_fanin():
edges = {0: {}}
for i in range(1, nleaves + 1):
edges[0][i] = i
edges[i] = {99: 1}
edges[99] = {}
return {vid: Vertex(vid, INF, edges[vid]) for vid in edges}
print(" SSSP fan-in: source 0 -> %d leaves -> hub 99" % nleaves)
for nw in (1, 2, 4):
for label, comb in (("no combine", None), ("combine=min", min)):
g = build_fanin()
m = PregelMaster(g, num_workers=nw, combine=comb,
agg_reducers={"reachable": sum},
checkpoint_every=10 ** 9)
r = m.run(make_sssp(0), max_supersteps=40)
print(" workers=%d %-12s raw=%-4d delivered=%-4d overall=%.2fx"
" msgs arriving at hub=%d"
% (nw, label, r["raw"], r["delivered"],
r["raw"] / max(1.0, r["delivered"]), m.inbound[99]))
for label, comb in (("no combine", None), ("combine=sum", lambda a, b: a + b)):
g = build_pagerank_graph()
m = PregelMaster(g, num_workers=3, combine=comb,
agg_reducers={"max_delta": max},
checkpoint_every=10 ** 9)
r = m.run(make_pagerank(len(g)), max_supersteps=200)
print(" PageRank 7-vertex graph: %-12s raw=%-4d delivered=%-4d "
"overall=%.2fx msgs arriving at vertex 1=%d"
% (label, r["raw"], r["delivered"],
r["raw"] / max(1.0, r["delivered"]), m.inbound[1]))
def exp_iv_cc_and_agg():
print()
print("=" * 78)
print("(iv) connected components; active set shrinks; aggregator global stat")
print("=" * 78)
g = build_cc_graph()
m = PregelMaster(g, num_workers=3, combine=None, checkpoint_every=10 ** 9)
r = m.run(cc_compute, max_supersteps=30)
comp = {vid: int(g[vid].value) for vid in sorted(g)}
print_history(m.value_history, fmt="%7.0f", head="cid")
print(" component ids = %s" % comp)
print(" supersteps=%d raw_msgs=%d active per superstep=%s wall=%.6fs"
% (r["steps"], r["raw"], [row["active"] for row in m.trace],
r["seconds"]))
assert comp == {1: 1, 2: 1, 3: 1, 4: 1, 5: 5, 6: 5, 7: 7}
g = build_pagerank_graph()
m = PregelMaster(g, num_workers=3, combine=lambda a, b: a + b,
agg_reducers={"max_delta": max}, checkpoint_every=10 ** 9)
m.run(make_pagerank(len(g)), max_supersteps=200)
g2 = build_pagerank_graph()
m2 = PregelMaster(g2, num_workers=3, agg_reducers={"pr_sum": sum,
"above_mean": sum},
checkpoint_every=10 ** 9)
for vid in g2:
g2[vid].value = g[vid].value
r2 = m2.run(make_pr_stat(len(g2)), max_supersteps=4)
total = m2.trace[0]["agg"]["pr_sum"] # aggregator value of superstep 0
mean = total / len(g2)
above = [vid for vid in sorted(g2) if g2[vid].value > mean]
print(" aggregator-only job: global sum(PR)=%.6f -> mean=%.5f ; "
"above-mean vertices counted by the aggregator = %d %s"
% (total, mean, int(r2["agg"].get("above_mean", 0)), above))
assert int(r2["agg"]["above_mean"]) == len(above)
def exp_v_fault_tolerance():
print()
print("=" * 78)
print("(v) fault tolerance: worker 2 crashes at superstep 5; checkpoint_every=2")
print("=" * 78)
g = build_sssp_graph()
m = PregelMaster(g, num_workers=3, combine=None, checkpoint_every=2,
crash_at=5, crash_worker=2,
agg_reducers={"reachable": sum})
r = m.run(make_sssp(1), max_supersteps=40)
fault = {vid: g[vid].value for vid in sorted(g)}
for line in m.log:
if "[ckpt]" in line or "[recovery]" in line:
print(line)
g2 = build_sssp_graph()
m2 = PregelMaster(g2, num_workers=3, combine=None, checkpoint_every=2,
agg_reducers={"reachable": sum})
r2 = m2.run(make_sssp(1), max_supersteps=40)
clean = {vid: g2[vid].value for vid in sorted(g2)}
print(" fault-free run: distances=%s supersteps=%d"
% ({k: int(v) for k, v in clean.items()}, r2["steps"]))
print(" crashed run : distances=%s supersteps=%d recovered=%s "
"recomputed_supersteps=%d"
% ({k: int(v) for k, v in fault.items()}, r["steps"], r["recovered"],
r["wasted"]))
assert fault == clean
print(" VERIFIED: identical results -> checkpoint + BSP replay is "
"deterministic recovery")
if __name__ == "__main__":
random.seed(425)
exp_i_pagerank()
exp_ii_sssp()
exp_iii_combine()
exp_iv_cc_and_agg()
exp_v_fault_tolerance()
【代码做什么？】
- 数据结构的建模：
Vertex保存id / value / edges（出边邻接表）/ active / pending_halt / inbox；ComputeContext是交给用户compute()的顶点 API，提供send()、vote_to_halt()、aggregate()、get_aggregated()、superstep()——这就是 24.3.1 伪代码里那些操作的可执行版本。 - 把机器变成对象：
PregelWorker持有自己的顶点分区、一个outbox（本超级步产生的消息）与aggregators（本超级步提供的聚合值）；PregelMaster持有全部 worker、owner（顶点 → worker 的路由表），并在run()里用一段显式循环实现屏障：先让所有 worker 跑完compute_phase()，再统一flush()（combine + 投递），再归约聚合器，最后应用”投票停机”。 - 三条 Pregel 语义被显式实现：(a) 消息在超级步末投递（
flush()只在所有 worker 的 compute 阶段结束后才执行）；(b) 投票停机（pending_halt在每个超级步末统一生效）；(c) 收到消息即唤醒（deliver()里若目标顶点active == False，就把它置回活跃、记一次wakeups并打印日志）。 - 四个算法：PageRank（全局残差判据 + 一个刻意写错的”局部停机”对照版）、SSSP（Bellman-Ford 式，含唤醒）、连通分量（标签传播）、以及一个只用聚合器的全局统计作业。
- 五个实验：(i) 7 顶点 PageRank 的逐超级步 PR 值 + 残差轨迹；(ii) 6 顶点 SSSP 的逐超级步距离 + 唤醒日志；(iii) 开关 Combine 的消息量对比（1/2/4 个 worker）；(iv) 连通分量 + 聚合器统计 + 活跃顶点数递减；(v) worker 2 在超级步 5 崩溃 → 分区重分配 → 回滚到超级步 4 的检查点 → 重算 → 与无故障结果逐位比对。
【分布式机制透视】
- 进程与网络：worker 是对象而不是进程（有意为之：顺序模拟让 BSP 语义完全可见，不会因为线程调度掩盖”消息在步末投递”这条规则）；”网络”就是 master 的路由表
owner+deliver()。第 24.4.2/24.4.3 节则改用离散事件模拟来表达”并发的墙钟时间”。 - 屏障在哪里：
for w in self.workers: w.compute_phase(compute)这四行执行完毕就是屏障；它之后的flush()才允许消息流动。这一行代码的位置，就是”第 $k$ 步发的消息只能第 $k+1$ 步收到”的实现。 - 状态如何维护：每个 worker 只保存自己分区的顶点；顶点值、边表、收件箱、活跃标志就是全部状态。跨分区的边由
owner路由（真实系统里需要额外保存目标 worker ID，见 24.3.7 的补充说明）。 - 并发与时序：顺序模拟 + 显式屏障 ⇒ 结果与”真实并行执行”在 BSP 语义下等价（这正是 24.3.1 确定性定理的实践含义：既然结果与顺序无关，顺序模拟就是合法的实现）。
- 容错：
checkpoint()用深拷贝保存{step, 每个顶点的 (value, edges, active, pending_halt, inbox), 分区映射, 聚合值, 计数器}；restore(dead_worker)把死掉 worker 的分区重新分给存活 worker，并把整个系统回滚到检查点——这正是讲义页 13 的”leader 重新分配分区 + 从最近检查点重新加载”。
【与理论的对应】
compute_phase+flush+_reduce_aggregators+ “应用停机”四步，逐行对应 24.3.1 伪代码的四个阶段。deliver()里的唤醒逻辑对应 24.3.1 的终止条件（”无活跃顶点且无在途消息”必须在投递之后判定，否则会漏掉唤醒）。_reduce_aggregators()的归约时机（屏障之后、下一步之前）对应 24.3.5 的”$k$ 步的值在 $k+1$ 步可读”。- 最后一段断言
assert fault == clean是 24.3.2 恢复正确性定理（利用 BSP 确定性重放）的直接验证。 - (i) 中”局部停机版 PageRank 的 $\sum\text{PR}\ne1$”则是 24.3.3 权重守恒证明依赖”全体同一步停机”这一条件的反例验证。
运行输出（真实运行结果，python3 pregel.py）：
==============================================================================
(i) PageRank, 7 vertices, d=0.85, eps=1e-3, combine=sum
==============================================================================
PR | v1 | v2 | v3 | v4 | v5 | v6 | v7
0 | 0.14286 | 0.14286 | 0.14286 | 0.14286 | 0.14286 | 0.14286 | 0.14286
1 | 0.38571 | 0.04167 | 0.16310 | 0.04167 | 0.16310 | 0.16310 | 0.04167
2 | 0.33411 | 0.07607 | 0.11149 | 0.07607 | 0.11149 | 0.21470 | 0.07607
3 | 0.36335 | 0.06876 | 0.13342 | 0.06876 | 0.13342 | 0.16353 | 0.06876
4 | 0.33228 | 0.07290 | 0.13135 | 0.07290 | 0.13135 | 0.18631 | 0.07290
5 | 0.35341 | 0.06850 | 0.13047 | 0.06850 | 0.13047 | 0.18015 | 0.06850
6 | 0.34368 | 0.07149 | 0.12972 | 0.07149 | 0.12972 | 0.18239 | 0.07149
7 | 0.34750 | 0.07012 | 0.13089 | 0.07012 | 0.13089 | 0.18038 | 0.07012
8 | 0.34560 | 0.07066 | 0.13026 | 0.07066 | 0.13026 | 0.18191 | 0.07066
9 | 0.34683 | 0.07039 | 0.13045 | 0.07039 | 0.13045 | 0.18111 | 0.07039
10 | 0.34608 | 0.07056 | 0.13039 | 0.07056 | 0.13039 | 0.18144 | 0.07056
11 | 0.34647 | 0.07046 | 0.13044 | 0.07046 | 0.13044 | 0.18129 | 0.07046
supersteps=12 raw_msgs=132 delivered=110 wakeups=0 vertex_ops=84
max|PR_pregel - PR_jacobi(200)| = 1.24e-04 sum(PR) = 1.000000 (must be 1)
active-set size per superstep: [7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7, 7]
--- aggregator trace: global max residual -> global halt ---
k= 0 active= 7 sent= 12 delivered= 10 agg[]
k= 1 active= 7 sent= 12 delivered= 20 agg[max_delta=0.242857]
k= 2 active= 7 sent= 12 delivered= 30 agg[max_delta=0.051607]
k= 3 active= 7 sent= 12 delivered= 40 agg[max_delta=0.051177]
k= 4 active= 7 sent= 12 delivered= 50 agg[max_delta=0.031072]
k= 5 active= 7 sent= 12 delivered= 60 agg[max_delta=0.021129]
k= 6 active= 7 sent= 12 delivered= 70 agg[max_delta=0.009728]
k= 7 active= 7 sent= 12 delivered= 80 agg[max_delta=0.003816]
k= 8 active= 7 sent= 12 delivered= 90 agg[max_delta=0.001892]
k= 9 active= 7 sent= 12 delivered= 100 agg[max_delta=0.001225]
k=10 active= 7 sent= 12 delivered= 110 agg[max_delta=0.000749]
k=11 active= 7 sent= 0 delivered= 110 agg[max_delta=0.000387]
for contrast, the NAIVE 'halt when my own delta is small' version:
sum(PR) after 200 supersteps = 0.813566 (mass leaked: 0.1864) vertices still active = 7
==============================================================================
(ii) SSSP from vertex 1 -- 'vote to halt' + 'wake up on new message'
==============================================================================
distances after every superstep ('inf' = not reached yet):
k | v1 | v2 | v3 | v4 | v5 | v6
0 | 0.0 | inf | inf | inf | inf | inf
1 | 0.0 | 1.0 | 5.0 | inf | 10.0 | inf
2 | 0.0 | 1.0 | 2.0 | 7.0 | 10.0 | inf
3 | 0.0 | 1.0 | 2.0 | 4.0 | 10.0 | 8.0
4 | 0.0 | 1.0 | 2.0 | 4.0 | 9.0 | 5.0
5 | 0.0 | 1.0 | 2.0 | 4.0 | 6.0 | 5.0
6 | 0.0 | 1.0 | 2.0 | 4.0 | 6.0 | 5.0
[wake-up] superstep 1: vertex 4 had halted, woken by message 7.0
[wake-up] superstep 2: vertex 6 had halted, woken by message 8.0
[wake-up] superstep 3: vertex 5 had halted, woken by message 9.0
final distances = {1: 0, 2: 1, 3: 2, 4: 4, 5: 6, 6: 5} == Bellman-Ford ground truth
supersteps=7 wakeups=3 raw_msgs=10
==============================================================================
(iii) the combiner: same answer, fewer messages on the wire
==============================================================================
SSSP fan-in: source 0 -> 12 leaves -> hub 99
workers=1 no combine raw=24 delivered=24 overall=1.00x msgs arriving at hub=12
workers=1 combine=min raw=24 delivered=13 overall=1.85x msgs arriving at hub=1
workers=2 no combine raw=24 delivered=24 overall=1.00x msgs arriving at hub=12
workers=2 combine=min raw=24 delivered=14 overall=1.71x msgs arriving at hub=2
workers=4 no combine raw=24 delivered=24 overall=1.00x msgs arriving at hub=12
workers=4 combine=min raw=24 delivered=16 overall=1.50x msgs arriving at hub=4
PageRank 7-vertex graph: no combine raw=132 delivered=132 overall=1.00x msgs arriving at vertex 1=33
PageRank 7-vertex graph: combine=sum raw=132 delivered=110 overall=1.20x msgs arriving at vertex 1=22
==============================================================================
(iv) connected components; active set shrinks; aggregator global stat
==============================================================================
cid| v1 | v2 | v3 | v4 | v5 | v6 | v7
0 | 1 | 2 | 3 | 4 | 5 | 6 | 7
1 | 1 | 1 | 2 | 3 | 5 | 5 | 7
2 | 1 | 1 | 1 | 2 | 5 | 5 | 7
3 | 1 | 1 | 1 | 1 | 5 | 5 | 7
4 | 1 | 1 | 1 | 1 | 5 | 5 | 7
component ids = {1: 1, 2: 1, 3: 1, 4: 1, 5: 5, 6: 5, 7: 7}
supersteps=5 raw_msgs=18 active per superstep=[7, 7, 6, 3, 2] wall=0.000118s
aggregator-only job: global sum(PR)=1.000000 -> mean=0.14286 ; above-mean vertices counted by the aggregator = 2 [1, 6]
==============================================================================
(v) fault tolerance: worker 2 crashes at superstep 5; checkpoint_every=2
==============================================================================
[fault] worker 2 crashes in superstep 5 (mid-superstep)
[recovery] worker 2 died -> 2 vertices reassigned to [0, 1]
[ckpt] checkpoint at start of superstep 0
[ckpt] checkpoint at start of superstep 2
[ckpt] checkpoint at start of superstep 4
[ckpt] checkpoint at start of superstep 4
[ckpt] checkpoint at start of superstep 6
fault-free run: distances={1: 0, 2: 1, 3: 2, 4: 4, 5: 6, 6: 5} supersteps=7
crashed run : distances={1: 0, 2: 1, 3: 2, 4: 4, 5: 6, 6: 5} supersteps=7 recovered=True recomputed_supersteps=1
VERIFIED: identical results -> checkpoint + BSP replay is deterministic recovery
输出怎么读：
- (i)：PageRank 从均匀分布 $1/7=0.14286$ 出发，12 个超级步后收敛；
sum(PR) = 1.000000（权重守恒），与 200 轮 Jacobi 精确解最大误差 $1.24\times10^{-4}$。聚合器轨迹显示max_delta从 $0.242857$ 按约 $0.85$ 的因子衰减到 $7.49\times10^{-4}$ —— 这就是 24.3.3 的压缩映射在数值上的样子。最后一段对照实验是本章最值得记住的反例：「我自己不变了就停机」的版本跑到 200 个超级步仍未收敛，$\sum\text{PR}$ 掉到 $0.813566$（权重泄漏 18.6%，且 7 个顶点仍然全部活跃）。 - (ii)：SSSP 的 6 顶点推演与 24.3.4 表格完全一致；三条
[wake-up]日志证明了”停机后被更短路径唤醒”这一机制真实发生（其中 v5 在超级步 3 被消息 9.0 唤醒，最终从 10 修正到 6）。 - (iii)：combiner 把 hub 收到的消息从 12 条压成 1/2/4 条（对应 1/2/4 个 worker），整体消息量降为 1.85×/1.71×/1.50× ——分区越细，combine 的局部性越差，这是 24.3.5 里”combine 是本地优化”的实测。
- (iv)：连通分量 5 个超级步得出 $\{1,2,3,4\}\to1,\{5,6\}\to5,\{7\}\to7$；活跃顶点数逐超级步递减
[7, 7, 6, 3, 2]（收敛特征）；纯聚合器作业统计出全局 $\sum\text{PR}=1.000000$、均值 $0.14286$、高于均值的顶点数 2（与离线计算一致）。 - (v)：故障注入后仍得到
distances={1:0, 2:1, 3:2, 4:4, 5:6, 6:5}，与无故障运行完全相同，recomputed_supersteps=1—— 检查点容错的端到端验证。
24.4.2 同步 BSP 与异步图处理的收敛速度对比（含 straggler 注入）
"""
sync_async_graph.py -- synchronous (BSP) vs asynchronous graph processing.
The same PageRank job, the same graph, the same accuracy target -- only the
execution model differs:
SYNC : Jacobi supersteps, static partition over P workers, and a BARRIER at
the end of every superstep. A superstep costs
max over workers (sum of the costs of that worker's active vertices)
so one slow vertex (a straggler) is paid once per superstep.
ASYNC : one global work queue over P cores, no barrier. A vertex is
recomputed as soon as a neighbour changed; the moment it finishes,
its new value is visible to its neighbours (Gauss-Seidel style), so a
straggler is paid once per recomputation of THAT vertex, not once per
superstep.
Printed: time to the accuracy target, recomputations, rounds, straggler
sensitivity (hot vertex vs peripheral vertex), determinism of both models, and
an ASCII convergence chart (log10 error vs simulated time).
Run: python3 sync_async_graph.py
"""
import heapq
import math
import random
from collections import deque
N_VERTICES = 60
DAMPING = 0.85
TARGET_ERR = 1e-4
STRAGGLER_COST = 6.0 # a slow/loaded machine hosting one vertex
N_WORKERS = 4 # both models get exactly 4 workers / cores
DELTA = 1e-7 # async: only propagate updates bigger than this
def build_graph(n=N_VERTICES, chords=40, seed=425):
"""A small-world ring: diameter is high enough that propagation order matters."""
random.seed(seed)
edges = {v: set() for v in range(n)}
for v in range(n):
edges[v].add((v + 1) % n) # the ring
while sum(len(e) for e in edges.values()) < n + chords:
a, b = random.randrange(n), random.randrange(n)
if a != b:
edges[a].add(b) # shortcuts
in_edges = {v: [] for v in range(n)}
for a, outs in edges.items():
for b in outs:
in_edges[b].append(a)
return edges, in_edges
def exact_pagerank(edges, in_edges, iters=500):
n = len(edges)
pr = {v: 1.0 / n for v in edges}
for _ in range(iters):
pr = {v: (1 - DAMPING) / n
+ DAMPING * sum(pr[u] / len(edges[u]) for u in in_edges[v])
for v in edges}
return pr
def run_sync(edges, in_edges, ref, costs, workers=N_WORKERS, cap=400):
"""BSP: compute -> BARRIER -> next superstep."""
n = len(edges)
value = {v: 1.0 / n for v in edges}
assign = {v: v % workers for v in edges} # static hash partition
t = 0.0
steps = 0
ops = 0
samples = []
while steps < cap:
new = {v: (1 - DAMPING) / n
+ DAMPING * sum(value[u] / len(edges[u]) for u in in_edges[v])
for v in edges}
ops += n
# ---- the barrier: every worker serves its own vertices sequentially
load = [0.0] * workers
for v in edges:
load[assign[v]] += costs[v]
t += max(load) # slowest worker decides
value = new
steps += 1
err = max(abs(value[v] - ref[v]) for v in edges)
samples.append((t, err))
if err < TARGET_ERR:
break
return {"time": t, "ops": ops, "rounds": steps, "samples": samples,
"value": dict(value), "err": err}
def run_async(edges, in_edges, ref, costs, order="fifo", cores=N_WORKERS,
delta=DELTA, cap=200000):
"""Work queue on `cores` cores, no barrier, immediate propagation."""
n = len(edges)
value = {v: 1.0 / n for v in edges}
queue = deque(sorted(edges))
queued = set(edges)
running = [] # heap of (finish_time, seq, vertex)
seq = 0
t = 0.0
ops = 0
recomp = {v: 0 for v in edges}
samples = []
def start(now):
nonlocal seq
while len(running) < cores and queue:
v = queue.popleft() if order == "fifo" else queue.pop()
queued.discard(v)
seq += 1
heapq.heappush(running, (now + costs[v], seq, v))
start(0.0)
err = float("inf")
while running and ops < cap:
finish, _, v = heapq.heappop(running)
t = finish
new = (1 - DAMPING) / n \
+ DAMPING * sum(value[u] / len(edges[u]) for u in in_edges[v])
change = abs(new - value[v])
value[v] = new
ops += 1
recomp[v] += 1
if change > delta: # delta-based pruning
for nb in edges[v]:
if nb not in queued:
queued.add(nb)
queue.append(nb)
start(t)
if ops % 5 == 0:
err = max(abs(value[u] - ref[u]) for u in edges)
samples.append((t, err))
if err < TARGET_ERR:
break
err = max(abs(value[u] - ref[u]) for u in edges)
return {"time": t, "ops": ops, "rounds": ops, "samples": samples,
"value": dict(value), "err": err, "left": len(queue),
"recomp": recomp}
def make_costs(edges, in_edges, placement):
costs = {v: 1.0 for v in edges}
if placement == "hot": # busy vertex: recomputed all the time
slow = max(in_edges, key=lambda v: (len(in_edges[v]), v))
costs[slow] = STRAGGLER_COST
return costs, slow
if placement == "peripheral": # rarely touched vertex
slow = min(in_edges, key=lambda v: (len(in_edges[v]), v))
costs[slow] = STRAGGLER_COST
return costs, slow
return costs, None
# --------------------------------------------------------------------------
# ASCII chart of log10(error) against simulated time
# --------------------------------------------------------------------------
def resample(samples, tmax, nbuckets=30):
prof = [None] * nbuckets
for t, err in samples:
b = min(nbuckets - 1, int(t / tmax * nbuckets)) if tmax > 0 else 0
prof[b] = err if prof[b] is None else min(prof[b], err)
last = None
for i in range(nbuckets):
if prof[i] is None:
prof[i] = last
last = prof[i]
return prof
def draw_chart(series, tmax, title):
print(" %s" % title)
print(" log10(err)")
profs = [(lbl, mark, resample(s, tmax)) for lbl, s, mark in series]
nb = len(profs[0][2])
for level in range(-1, -7, -1):
line = " %5d |" % level
for i in range(nb):
cell = " "
for _, mark, prof in profs:
e = prof[i]
if e and e > 0 and math.floor(math.log10(e)) == level:
cell = mark if cell == " " else "*"
line += cell
print(line)
print(" +" + "-" * nb)
print(" time 0" + " " * (nb - 13) + "%.0f units" % tmax)
print(" " + " ".join("%s=%s" % (lbl, mk) for lbl, mk, _ in profs))
def main():
edges, in_edges = build_graph()
ref = exact_pagerank(edges, in_edges)
print("graph: %d vertices, %d directed edges (ring + shortcuts), damping=%.2f"
% (len(edges), sum(len(e) for e in edges.values()), DAMPING))
print("target: max|PR - PR*| < %.0e ; both models use %d workers, "
"straggler cost %.0f vs normal cost 1"
% (TARGET_ERR, N_WORKERS, STRAGGLER_COST))
print()
print(" %-6s %-11s %9s %9s %9s %9s %s"
% ("mode", "straggler", "time", "recompute", "rounds", "slowdown",
"converged"))
results = {}
base = {}
for placement in ("none", "hot", "peripheral"):
costs, slow = make_costs(edges, in_edges, placement)
s = run_sync(edges, in_edges, ref, costs)
a = run_async(edges, in_edges, ref, costs)
results[("sync", placement)] = s
results[("async", placement)] = a
if placement == "none":
base = {"sync": s["time"], "async": a["time"]}
tag = placement if slow is None else "%s(v%d)" % (placement, slow)
for mode, r in (("sync", s), ("async", a)):
print(" %-6s %-11s %9.1f %9d %9d %8.2fx %s"
% (mode, tag, r["time"], r["ops"], r["rounds"],
r["time"] / base[mode], r["err"] < TARGET_ERR))
print()
print(" straggler cost analysis")
hot_v = make_costs(edges, in_edges, "hot")[1]
per_v = make_costs(edges, in_edges, "peripheral")[1]
print(" SYNC : the barrier pays the straggler in every superstep in which")
print(" it is active; PageRank keeps every vertex active, so the")
print(" penalty is +%.0f x %d supersteps = +%.0f time units"
% (STRAGGLER_COST - 1, results[("sync", "none")]["rounds"],
results[("sync", "hot")]["time"] - base["sync"]))
print(" ASYNC: the penalty is about (cost-1) x (#recomputations of that")
print(" vertex) / number of cores -- it never multiplies by rounds")
print(" hot vertex v%d: recomputed %d times -> total time +%.0f"
% (hot_v, results[("async", "hot")]["recomp"][hot_v],
results[("async", "hot")]["time"] - base["async"]))
print(" peripheral vertex v%d: recomputed %d times -> total time +%.0f"
% (per_v, results[("async", "peripheral")]["recomp"][per_v],
results[("async", "peripheral")]["time"] - base["async"]))
print(" => async isolates the straggler; BSP multiplies it by the number")
print(" of supersteps the job needs.")
print()
print(" --- determinism ---")
costs, _ = make_costs(edges, in_edges, "hot")
r1 = run_sync(edges, in_edges, ref, costs)
r2 = run_sync(edges, in_edges, ref, costs)
print(" SYNC twice: identical convergence trace? %s"
% (r1["samples"] == r2["samples"] and r1["time"] == r2["time"]))
a1 = run_async(edges, in_edges, ref, costs, order="fifo")
a2 = run_async(edges, in_edges, ref, costs, order="lifo")
print(" ASYNC fifo order: time=%8.1f recomputations=%6d" % (a1["time"], a1["ops"]))
print(" ASYNC lifo order: time=%8.1f recomputations=%6d <- schedule matters!"
% (a2["time"], a2["ops"]))
dif = max(abs(a1["value"][v] - a2["value"][v]) for v in edges)
print(" both schedules reach the same fixed point within the target error")
print(" (max-error %.1e and %.1e) but the values are NOT bit-identical:"
% (a1["err"], a2["err"]))
print(" max value difference = %.2e -> async execution is not deterministic"
% dif)
print()
s = results[("sync", "none")]
a = results[("async", "none")]
tmax = max(s["samples"][-1][0], a["samples"][-1][0])
draw_chart([("sync", s["samples"], "S"), ("async", a["samples"], "A")], tmax,
"max |PR(t) - PR*| vs simulated time (no straggler)")
print(" async reaches the target in %.2fx less time and %.2fx fewer "
"vertex recomputations than BSP"
% (s["time"] / a["time"], s["ops"] / a["ops"]))
if __name__ == "__main__":
main()
【代码做什么？】
- 造一张 60 个顶点的小世界环图（环 + 40 条随机捷径）——直径较大，因此”传播顺序”会真正影响收敛速度（对比 24.4.1 里直径很小的 7 顶点图）。
- 同步版（
run_sync）：Jacobi 超级步 + 静态哈希分区到 4 个 worker + 屏障；一个超级步的耗时 = 负载最重的那个 worker 的耗时之和，这正是”最慢者决定整轮”的建模。 - 异步版（
run_async）：一个全局工作队列 + 4 个核 + 离散事件模拟（heapq维护”谁在什么时刻算完”）；顶点算完立刻把新值传播给邻居，没有屏障；并加入 delta 剪枝（变化量 $>10^{-7}$ 才继续传播）。 - straggler 注入：把某个顶点的计算成本设为 6.0（正常为 $1.0+0.3$），并且分别测试三种放置位置——无 straggler、热点顶点（入度最大）、边缘顶点（入度最小），从而暴露”straggler 成本如何被放大”的机制差异。
- 确定性与调度敏感性：同步版跑两遍比对收敛轨迹；异步版分别用 FIFO 与 LIFO 两种调度顺序跑，比较所需的重算次数与最终数值。
- ASCII 收敛图：把两个运行的”最大误差 vs 模拟时间”画在同一张对数坐标图上。
【分布式机制透视】
- 两种时间模型的差别：同步版的
t += max(load)一行，就是”屏障”这个概念的全部；异步版的时间来自事件堆——每个核独立推进，t是”最后一个完成的任务的结束时刻”（makespan）。 - straggler 的放大机制：同步版里，成本为 6 的顶点在每一个超级步里都要被付一次（PageRank 中所有顶点每轮都活跃），于是它的成本被乘以超级步数；异步版里，它只在”该顶点自己被重算“时付出代价（热点顶点 13 次、边缘顶点 8 次），其余时间它只是一台慢核，不影响别人。
- 调度即策略：异步版的
order="fifo"/"lifo"就是真实的调度器策略选择，结果相差 27 倍（800 vs 21340 次重算），这是”异步系统必须把调度器当作一等公民来设计”的最好证据。 - delta 剪枝的作用：
if change > delta决定了”消息是否继续向下游传播”，它让异步版在 480 次重算内达到目标，而同步版需要 1080 次（每个超级步都要重算全部 60 个顶点）。
【与理论的对应】
- 同步版 = 24.3.1 的 BSP 框架；异步版 = 24.2.7 的异步执行；
resample()/draw_chart()画出的曲线就是”收敛速度对比”的定量证据。 - “同步确定性、异步不确定”在输出里被明确验证（同步两次轨迹完全相同；异步两种调度下最终值最大差 $5.75\times10^{-5}$）。
- = 24.2.7 表格中”straggler 敏感度：同步高 / 异步低”的实测（同步 ×4.00 vs 异步 ×1.23，就是 straggler 的 4 倍系数）。
运行输出（真实运行结果，python3 sync_async_graph.py）：
graph: 60 vertices, 100 directed edges (ring + shortcuts), damping=0.85
target: max|PR - PR*| < 1e-04 ; both models use 4 workers, straggler cost 6 vs normal cost 1
mode straggler time recompute rounds slowdown converged
sync none 270.0 1080 18 1.00x True
async none 120.0 480 480 1.00x True
sync hot(v50) 360.0 1080 18 1.33x True
async hot(v50) 217.0 800 800 1.81x True
sync peripheral(v0) 360.0 1080 18 1.33x True
async peripheral(v0) 130.0 480 480 1.08x True
straggler cost analysis
SYNC : the barrier pays the straggler in every superstep in which
it is active; PageRank keeps every vertex active, so the
penalty is +5 x 18 supersteps = +90 time units
ASYNC: the penalty is about (cost-1) x (#recomputations of that
vertex) / number of cores -- it never multiplies by rounds
hot vertex v50: recomputed 13 times -> total time +97
peripheral vertex v0: recomputed 8 times -> total time +10
=> async isolates the straggler; BSP multiplies it by the number
of supersteps the job needs.
--- determinism ---
SYNC twice: identical convergence trace? True
ASYNC fifo order: time= 217.0 recomputations= 800
ASYNC lifo order: time= 6115.0 recomputations= 21340 <- schedule matters!
both schedules reach the same fixed point within the target error
(max-error 8.6e-05 and 9.0e-05) but the values are NOT bit-identical:
max value difference = 5.75e-05 -> async execution is not deterministic
max |PR(t) - PR*| vs simulated time (no straggler)
log10(err)
-1 |
-2 |ASSSS
-3 | AAAA***SSSSSSSSSS
-4 | AAAAA SSSSSSSSSSS
-5 | AAAAAAAAAAAAAAAA*
-6 |
+------------------------------
time 0 270 units
sync=S async=A
async reaches the target in 2.25x less time and 2.25x fewer vertex recomputations than BSP
输出怎么读：
- 主表：同步 270 时间单位 / 1080 次重算，异步 120 / 480 ⇒ 异步时间与计算量都降为 2.25×（约 2.3 倍加速），与 24.2.7 中”更新立即传播（Gauss-Seidel）比 Jacobi 收敛快”的机理一致。
- straggler 行：同步 ×1.33（热点）与 ×1.33（边缘）——它对两类顶点的惩罚相同（因为同步版每轮全员活跃，放哪儿都一样）；异步对边缘顶点只 ×1.08（+10 个时间单位），对热点顶点 ×1.81（+97）——“straggler 放在哪里”在异步系统里才成为一个真正的问题。
- 确定性行：同步两次运行轨迹完全相同；异步 FIFO 800 次重算 vs LIFO 21340 次（27 倍差距），二者最终值有 $5.75\times10^{-5}$ 的差异 ⇒ 异步失去确定性。
- 图：
S（同步）的曲线先快速下降再在 $-3$ 附近拉出长尾；A（异步）的曲线在时间轴最左侧就扎到 $-5$ ——一眼看出异步在时间上的优势。
24.4.3 分布式 SGD：同步 / 异步 / SSP / Local SGD 的完整对比
"""
dist_sgd.py -- synchronous vs asynchronous vs SSP vs Local SGD on a parameter
server, with a straggler and a simulated wall clock.
Problem: ridge regression (distributed least squares) on synthetic data, solved
by mini-batch SGD. The model is sharded over 2 parameter servers; every worker
pulls the parameters, computes a gradient on its own mini-batch, and pushes the
gradient back. Only the ORDER in which those pushes are allowed to land
differs between the four modes:
sync : barrier every iteration; the round is applied as the average of all
workers' gradients; wall clock per round = SLOWEST worker.
async : a worker pushes as soon as it is done; nobody waits. Gradients are
therefore STALE: they were computed on an older version of w.
ssp : bounded staleness -- a worker may be at most s iterations ahead of
the slowest worker, then it must wait. s=0 is sync, s=inf is async.
local : Local SGD / FedAvg -- each worker takes H local steps before the
models are averaged (all-reduce). Fewer communications.
Printed: wall time and iterations to the target loss, staleness statistics,
straggler impact, behaviour at a large learning rate (async diverges), an ASCII
loss chart, and the communication-count comparison.
Run: python3 dist_sgd.py
"""
import heapq
import math
import random
N_FEATURES = 20
N_TRAIN = 4000
N_EVAL = 256
BATCH = 32
N_WORKERS = 4
STRAGGLER_WORKER = 3
STRAGGLER_FACTOR = 4.0 # that worker's iteration takes 4x longer
RIDGE = 1e-3
ITER_COST = 1.0 # normal compute cost of one mini-batch
COMM_COST = 0.30 # extra cost of one pull+push per iteration
# --------------------------------------------------------------------------
# data + model
# --------------------------------------------------------------------------
def make_data(seed=425):
random.seed(seed)
w_star = [random.gauss(0, 0.5) for _ in range(N_FEATURES)]
X, y = [], []
for _ in range(N_TRAIN + N_EVAL):
row = [random.gauss(0, 1) for _ in range(N_FEATURES)]
X.append(row)
y.append(sum(a * b for a, b in zip(row, w_star)) + random.gauss(0, 0.1))
return X[:N_TRAIN], y[:N_TRAIN], X[N_TRAIN:], y[N_TRAIN:]
def loss(w, X, y):
tot = 0.0
try:
for row, t in zip(X, y):
p = sum(a * b for a, b in zip(row, w))
tot += (p - t) ** 2
return tot / len(X) + RIDGE * sum(v * v for v in w)
except OverflowError: # a diverging run: the loss exploded
return 1e30
def gradient(w, X, y, idx):
g = [2 * RIDGE * v for v in w]
for i in idx:
row = X[i]
p = sum(a * b for a, b in zip(row, w))
e = 2.0 * (p - y[i]) / len(idx)
for j in range(N_FEATURES):
g[j] += e * row[j]
return g
def batches(n, batch, k, rng):
idx = [(k * i) % n for i in range(n)]
rng.shuffle(idx)
for i in range(0, n - batch + 1, batch):
yield idx[i:i + batch]
# --------------------------------------------------------------------------
# the simulated cluster
# --------------------------------------------------------------------------
def iter_cost(wid, straggler=True):
c = ITER_COST + COMM_COST
if straggler and wid == STRAGGLER_WORKER:
c *= STRAGGLER_FACTOR
return c
class Cluster(object):
"""One mini-batch is pre-assigned to each (worker, iteration) pair."""
def __init__(self, X, y, seed=425):
self.X, self.y = X, y
self.rng = random.Random(seed)
self.streams = {}
for wid in range(N_WORKERS):
self.streams[wid] = list(batches(len(X), BATCH,
wid, random.Random(seed + wid)))
def batch(self, wid, k):
s = self.streams[wid]
return s[k % len(s)]
def run(mode, X, y, Xe, ye, lr, target, s=0, H=1, max_iters=4000,
straggler=True, seed=425, eval_every=10):
"""Returns wall time, gradient computations, communications, loss history."""
cl = Cluster(X, y, seed)
w = [0.0] * N_FEATURES
hist = []
comms = 0
iters = 0
staleness = []
clock = [0] * N_WORKERS
t = 0.0
def record(now, force=False):
if force or len(hist) == 0 or iters % eval_every < N_WORKERS:
hist.append((now, loss(w, Xe, ye)))
record(0.0, True)
if mode == "sync":
while iters < max_iters:
grads = []
for wid in range(N_WORKERS):
grads.append(gradient(w, X, y, cl.batch(wid, clock[wid])))
clock[wid] += 1
iters += N_WORKERS
comms += 2 * N_WORKERS
avg = [sum(g[j] for g in grads) / N_WORKERS
for j in range(N_FEATURES)]
w = [w[j] - lr * avg[j] for j in range(N_FEATURES)]
t += max(iter_cost(wid, straggler) for wid in range(N_WORKERS))
record(t)
if hist[-1][1] < target or hist[-1][1] > 1e7:
break
elif mode in ("async", "ssp"):
bound = float("inf") if mode == "async" else s
heap = []
blocked = []
for wid in range(N_WORKERS):
heapq.heappush(heap, (iter_cost(wid, straggler), wid))
while heap and iters < max_iters:
now, wid = heapq.heappop(heap)
t = now
ver = clock[wid]
# ---- SSP: a worker more than s iterations ahead of the slowest
# worker must WAIT (the slowest worker itself never waits,
# because its clock equals min(clock), so there is no deadlock)
if ver - min(clock) > bound:
blocked.append(wid)
continue
g = gradient(w, X, y, cl.batch(wid, ver))
staleness.append(max(0, sum(clock) // N_WORKERS - ver))
w = [w[j] - lr * g[j] for j in range(N_FEATURES)]
clock[wid] += 1
iters += 1
comms += 2
heapq.heappush(heap, (t + iter_cost(wid, straggler), wid))
for b in list(blocked): # release workers that may go on
if clock[b] - min(clock) <= bound:
blocked.remove(b)
heapq.heappush(heap, (t, b))
record(t)
if hist[-1][1] < target or hist[-1][1] > 1e7:
break
elif mode == "local":
while iters < max_iters:
models = []
for wid in range(N_WORKERS):
wl = list(w)
for k in range(H):
g = gradient(wl, X, y, cl.batch(wid, clock[wid] + k))
wl = [wl[j] - lr * g[j] for j in range(N_FEATURES)]
clock[wid] += H
models.append(wl)
iters += N_WORKERS * H
comms += 2 * N_WORKERS
w = [sum(m[j] for m in models) / N_WORKERS
for j in range(N_FEATURES)]
t += max(iter_cost(wid, straggler) for wid in range(N_WORKERS)) * H
record(t, force=True)
if hist[-1][1] < target or hist[-1][1] > 1e7:
break
else:
raise ValueError(mode)
return {"mode": mode, "time": t, "iters": iters, "comms": comms,
"hist": hist, "final_loss": hist[-1][1],
"reached": hist[-1][1] < target,
"diverged": hist[-1][1] > 1e7,
"staleness": (sum(staleness) / len(staleness)) if staleness else 0.0,
"max_staleness": max(staleness) if staleness else 0}
# --------------------------------------------------------------------------
# ASCII loss chart: log10(loss) against simulated time
# --------------------------------------------------------------------------
def draw_chart(series, tmax, nbuckets=32):
print(" log10(loss) (loss at the END of each time bucket)")
profs = []
for label, hist, mark in series:
prof = [None] * nbuckets
for tt, l in hist:
b = min(nbuckets - 1, int(tt / tmax * nbuckets)) if tmax > 0 else 0
prof[b] = l if prof[b] is None else min(prof[b], l)
last = None
for i in range(nbuckets):
if prof[i] is None:
prof[i] = last
last = prof[i]
profs.append((label, mark, prof))
allv = [l for _, hist, _ in series for _, l in hist if 0 < l < 1e7]
hi = int(math.ceil(math.log10(max(allv)))) if allv else 1
lo = int(math.floor(math.log10(min(allv)))) if allv else -3
for level in range(hi, lo - 1, -1):
line = " %5d |" % level
for i in range(nbuckets):
cell = " "
for _, mark, prof in profs:
e = prof[i]
if e and e > 0 and math.floor(math.log10(e)) == level:
cell = mark if cell == " " else "*"
line += cell
print(line)
print(" +" + "-" * nbuckets)
print(" time 0" + " " * (nbuckets - 13) + "%.0f units" % tmax)
print(" " + " ".join("%s=%s" % (l, m) for l, m, _ in profs))
def main():
X, y, Xe, ye = make_data()
w0 = [0.0] * N_FEATURES
print("model: ridge regression, %d features, %d training samples, mini-batch"
" %d, %d workers (worker %d is %.0fx slower)"
% (N_FEATURES, N_TRAIN, BATCH, N_WORKERS, STRAGGLER_WORKER,
STRAGGLER_FACTOR))
print("initial loss = %.4f" % loss(w0, Xe, ye))
lr, target = 0.05, 0.02
print()
print("=== A. moderate learning rate lr=%.2f, target loss < %.2f ==="
% (lr, target))
r = {}
for mode, kw in (("sync", {}), ("async", {}), ("ssp", {"s": 3}),
("local", {"H": 20})):
r[mode] = run(mode, X, y, Xe, ye, lr, target, **kw)
print(" %-6s %10s %10s %10s %10s %12s %8s"
% ("mode", "time", "grads", "comms", "final", "avg stale", "reached"))
for mode in ("sync", "async", "ssp", "local"):
d = r[mode]
print(" %-6s %10.1f %10d %10d %10.5f %12.2f %8s"
% (mode, d["time"], d["iters"], d["comms"], d["final_loss"],
d["staleness"], d["reached"]))
print()
print(" straggler impact (same runs, straggler removed):")
for mode, kw in (("sync", {}), ("async", {}), ("ssp", {"s": 3})):
d = run(mode, X, y, Xe, ye, lr, target, straggler=False, **kw)
print(" %-6s time=%7.1f with straggler vs %7.1f without -> x%.2f"
% (mode, r[mode]["time"], d["time"], r[mode]["time"] / d["time"]))
print()
print("=== B. effect of the staleness bound s (lr=%.2f, s=0 equals a "
"bulk-synchronous schedule, s=inf equals async) ===" % lr)
print(" %-10s %9s %9s %10s %10s %12s" %
("bound s", "time", "grads", "avg stale", "max stale", "final loss"))
for sv, name in ((0, "s=0"), (1, "s=1"), (3, "s=3"), (10, "s=10"),
(float("inf"), "s=inf")):
d = run("ssp", X, y, Xe, ye, lr, target, s=sv)
print(" %-10s %9.1f %9d %10.2f %10s %12.5f"
% (name, d["time"], d["iters"], d["staleness"],
d["max_staleness"], d["final_loss"]))
print()
print("=== C. large learning rate lr=0.22: staleness breaks convergence ===")
print(" %-10s %9s %11s %14s %s" %
("mode", "grads", "comms", "final loss", "verdict"))
for mode, kw in (("sync", {}), ("async", {}), ("ssp", {"s": 1}),
("ssp", {"s": 3}), ("ssp", {"s": 10})):
d = run(mode, X, y, Xe, ye, 0.22, target, max_iters=900, **kw)
tag = mode if mode != "ssp" else "ssp(s=%d)" % kw["s"]
verdict = "DIVERGED" if d["diverged"] else ("converged" if d["reached"]
else "not reached")
print(" %-10s %9d %11d %14.4g %s (avg stale %.2f, max %d)"
% (tag, d["iters"], d["comms"], d["final_loss"], verdict,
d["staleness"], d["max_staleness"]))
print(" async diverges: its gradients were computed on parameters up to "
"%d versions old" % big_max(X, y, Xe, ye, target=target))
print()
print("=== D. loss vs simulated time (lr=%.2f, with straggler) ===" % lr)
tmax = max(r[m]["time"] for m in ("sync", "async", "ssp"))
draw_chart([("sync", r["sync"]["hist"], "S"),
("async", r["async"]["hist"], "A"),
("ssp", r["ssp"]["hist"], "P")], tmax)
print()
print("=== E. communication volume (Local SGD / FedAvg) ===")
print(" sync SGD : pull+push every iteration = 2 messages per gradient")
print(" Local SGD : pull+push every H local steps = 2/H messages per gradient")
print(" %-12s %9s %12s %12s %10s %12s" %
("mode", "grads", "comm msgs", "msgs/grad", "time", "final loss"))
for H in (1, 5, 20, 50):
d = run("local", X, y, Xe, ye, lr, target, H=H)
tag = "local H=%d" % H
print(" %-12s %9d %12d %12.4f %10.1f %12.5f"
% (tag, d["iters"], d["comms"], d["comms"] / d["iters"],
d["time"], d["final_loss"]))
d = r["sync"]
print(" %-12s %9d %12d %12.4f %10.1f %12.5f"
% ("sync SGD", d["iters"], d["comms"], d["comms"] / d["iters"],
d["time"], d["final_loss"]))
h50 = run("local", X, y, Xe, ye, lr, target, H=50)
print(" Local SGD with H=50 sends %.4f messages per gradient vs %.1f for "
"sync SGD -> %.0fx fewer messages"
% (h50["comms"] / h50["iters"], r["sync"]["comms"] / r["sync"]["iters"],
(r["sync"]["comms"] / r["sync"]["iters"])
/ (h50["comms"] / h50["iters"])))
def big_max(X, y, Xe, ye, target):
"""max staleness observed by a diverging async run (for the report text)."""
d = run("async", X, y, Xe, ye, 0.22, target, max_iters=900)
return d["max_staleness"]
if __name__ == "__main__":
main()
【代码做什么？】
- 造分布式最小二乘（岭回归）问题：20 维特征、4000 个训练样本、256 个独立评测样本；mini-batch 32；4 个 worker，其中 worker 3 的每轮耗时是别人的 4 倍（straggler）。
- 用离散事件模拟一个参数服务器集群：worker 每轮”pull 参数 → 算梯度 → push”，push 成本 0.30 时间单位（模拟通信），计算成本 1.0；四种模式的区别只在”push 之后能不能继续跑”。
- A 段：中等学习率（0.05）下四种模式的时间、梯度数、通信数、最终损失、平均陈旧度；并做”有/无 straggler”的对照。
- B 段：把 staleness bound $s$ 从 0 扫到 $\infty$，看”速度 ↔ 陈旧度”这条曲线。
- C 段：把学习率提高到 0.22 —— 同步收敛、异步发散、SSP 仍然收敛。
- D 段：三种模式的损失曲线对照图（对数纵轴 vs 模拟时间）。
- E 段：Local SGD / FedAvg 把本地步数 $H$ 从 1 扫到 50，统计”每条梯度需要多少条通信消息”，验证 $H=1$ 与同步 SGD 完全等价、$H=50$ 时通信量降为 1/50。
【分布式机制透视】
- 墙钟时间是怎么算出来的：同步模式的”一轮 = 最慢 worker 的成本”（
t += max(cost_i)）就是屏障；异步/SSP 用事件堆推进（heapq），每个 worker 独立前进；SSP 额外维护clock[]并在超限时把 worker 放进blocked列表，等最慢者推进后再释放——这段blocked逻辑就是 24.3.7 伪代码的可执行版本。 - staleness 是被”测量”出来的：代码在每次 push 时记录
sum(clock)//N_WORKERS - ver，于是输出里的”平均/最大陈旧度”不是估计值而是实测值（异步最大 13、SSP($s$=3) 最大 3）。 - 通信计数：
comms += 2（pull + push）是”通信次数”的直接计数；Local SGD 把同步周期拉长为 $H$，于是每条梯度的通信次数 = 2/H。 - “发散”如何被检测：
loss()捕获OverflowError并返回 $10^{30}$，训练循环在损失超过 $10^7$ 时提前判发散并停止——这模拟了真实训练中”loss 变 NaN/爆炸后必须重启”的场景。
【与理论的对应】
- A 段印证 24.3.6 的表格：”同步收敛好但慢（且梯度数更多，因为平均梯度 = 更大 batch 的等效步长更小）、异步快但陈旧、SSP 折中”。
- B 段是 24.3.7 的 $s=0$/$s=\infty$ 退化分析的实测（$s=0$：陈旧度恒为 0，等价于同步调度结构；$s=\infty$：陈旧度最大 9，最快）。
- C 段是 24.3.6 的 staleness 误差公式 $\vert \nabla f(w_\tau)-\nabla f(w_t)\vert \propto \eta\,\tau_s$ 的直接验证：学习率放大 4.4 倍后，无界延迟（$\tau_s$ 最大 13）的异步发散，而有界延迟（$\tau_s\le10$）的 SSP 仍然收敛。
- E 段验证 Local SGD 的通信-收敛权衡，也是 24.2.13 中联邦学习为什么必须用 FedAvg 的定量理由。
运行输出（真实运行结果，python3 dist_sgd.py）：
model: ridge regression, 20 features, 4000 training samples, mini-batch 32, 4 workers (worker 3 is 4x slower)
initial loss = 3.7480
=== A. moderate learning rate lr=0.05, target loss < 0.02 ===
mode time grads comms final avg stale reached
sync 249.6 192 384 0.01976 0.00 True
async 20.8 52 104 0.01971 0.44 True
ssp 52.0 50 100 0.01930 0.60 True
local 416.0 320 32 0.01674 0.00 True
straggler impact (same runs, straggler removed):
sync time= 249.6 with straggler vs 62.4 without -> x4.00
async time= 20.8 with straggler vs 16.9 without -> x1.23
ssp time= 52.0 with straggler vs 16.9 without -> x3.08
=== B. effect of the staleness bound s (lr=0.05, s=0 equals a bulk-synchronous schedule, s=inf equals async) ===
bound s time grads avg stale max stale final loss
s=0 62.4 50 0.00 0 0.01879
s=1 57.2 50 0.22 1 0.01848
s=3 52.0 50 0.60 3 0.01930
s=10 26.0 52 0.58 8 0.01970
s=inf 20.8 52 0.44 9 0.01971
=== C. large learning rate lr=0.22: staleness breaks convergence ===
mode grads comms final loss verdict
sync 40 80 0.01813 converged (avg stale 0.00, max 0)
async 90 180 1.536e+07 DIVERGED (avg stale 0.52, max 13)
ssp(s=1) 80 160 0.01767 converged (avg stale 0.24, max 1)
ssp(s=3) 100 200 0.01648 converged (avg stale 0.66, max 3)
ssp(s=10) 333 666 0.01822 converged (avg stale 1.77, max 8)
async diverges: its gradients were computed on parameters up to 13 versions old
=== D. loss vs simulated time (lr=0.05, with straggler) ===
log10(loss) (loss at the END of each time bucket)
1 |
0 |SSSSSS
-1 |*PP SSSSSSSSSS
-2 | AA*****************************
+--------------------------------
time 0 250 units
sync=S async=A ssp=P
=== E. communication volume (Local SGD / FedAvg) ===
sync SGD : pull+push every iteration = 2 messages per gradient
Local SGD : pull+push every H local steps = 2/H messages per gradient
mode grads comm msgs msgs/grad time final loss
local H=1 192 384 2.0000 249.6 0.01976
local H=5 220 88 0.4000 286.0 0.01779
local H=20 320 32 0.1000 416.0 0.01674
local H=50 600 24 0.0400 780.0 0.01721
sync SGD 192 384 2.0000 249.6 0.01976
Local SGD with H=50 sends 0.0400 messages per gradient vs 2.0 for sync SGD -> 50x fewer messages
输出怎么读（本节最有价值的实验）：
- A 段：
sync 249.6 / async 20.8 / ssp(3) 52.0 / local(H=20) 416.0（时间单位）——异步在同一目标损失上比同步快约 12 倍，而 SSP 落在两者之间；local最慢却是唯一把通信量降到 32 条（同步 384 条）的模式。 - straggler 段：同步 ×4.00（正好等于 straggler 的 4 倍慢——因为屏障把它的代价乘上了轮数），异步 ×1.23，SSP ×3.08 ⇒ 同步训练最大的敌人就是 straggler，这与讲义页 23（MapReduce 的 slow server）中”最慢的任务拖慢整个作业”完全同源。
- B 段：
s=0 → 62.4 / s=1 → 57.2 / s=3 → 52.0 / s=10 → 26.0 / s=inf → 20.8，最大陈旧度0/1/3/8/9—— $s$ 这一个旋钮同时控制速度与陈旧度。 - C 段：
lr=0.22时同步最终损失 0.0181（收敛），异步 $1.536\times10^{7}$（发散，最大陈旧度 13），而ssp(s=1/3/10)分别收敛到 0.0177 / 0.0165 / 0.0182 ⇒ “有界延迟保证收敛”。 - D 段：
S曲线平滑但耗时最长；A曲线最陡（很快落到目标）；P（SSP）居中——三种模式的性格一眼可见。 - E 段：
local H=1与sync SGD的三项指标完全相同（192 梯度 / 384 消息 / 损失 0.01976，逐位相同），验证”$H=1$ 的 Local SGD 就是同步 SGD”；H=50时每条梯度的消息数从 2.0 降到 0.04，即 50× 的通信削减——这正是联邦学习的基础。
24.5 性能与可扩展性分析
24.5.1 Pregel 的性能模型
总时间 ≈ 超级步数 × 每步的（计算 + 通信 + 同步）开销 + 检查点开销 + 恢复开销。
| 成本项 | 表达式 | 主要影响因素 | 降低手段 |
|---|---|---|---|
| 计算 | $K \cdot \max_w(\text{本步顶点计算量})$ | 顶点度数分布（幂律 ⇒ 倾斜） | 按度数感知的分区、vertex-cut、把 hub 的邻居与其放同一机器 |
| 通信 | $K \cdot \sum_{v\in\text{active}} \deg^{out}(v)$ 条消息 | 跨机边数量、活跃集合大小 | combine、delta 剪枝、局部性分区（edge-cut 最小化）、把消息批量发送 |
| 屏障/同步 | $K \cdot (\text{最慢 worker 的耗时} - \text{平均耗时})$ | straggler（慢盘/网络/CPU/别的作业干扰） | backup worker（与 MapReduce 的 speculative execution 同源）、异步执行 |
| 检查点 | $\frac{K}{N_{ckpt}} \cdot (\vert V\vert +\vert E\vert )$ 的持久化 | 状态大小、存储带宽 | 增大 $N_{ckpt}$、只存必要状态、异步写 |
| 恢复 | $\le N_{ckpt}\cdot c$ 的重算 | 检查点间隔、故障率 | 调优 $N_{ckpt}$（24.3.2 的 $N^*=\sqrt{2C_{ckpt}/(\lambda c)}$）、confined recovery |
三个必须记住的定量事实：
- 消息数与图的结构强相关：PageRank 每轮 $O(\vert E\vert )$；SSSP 在有剪枝时接近 $O(\vert E\vert )$（实测 6 顶点图只发了 10 条消息），最坏 $O(D\vert E\vert )$。轮数与图直径/最慢收敛链相关——这就是”异步在直径大的图上赢得多”的原因（24.4.2 实测 2.25×）。
- 幂律图导致双重倾斜：计算倾斜（一个 hub 的一轮计算是普通顶点的 $10^5$ 倍）与通信倾斜（它的出边每轮都要发消息）。缓解顺序是：combine（把”多对一”的消息合并）→ vertex-cut（把 hub 的边摊开）→ 优先级调度（先处理重要顶点）。
- 讲义给出的实测数据（务必记住的具体数字）：10 亿顶点树的 SSSP，50 worker = 180 秒、800 worker = 20 秒（16 倍机器只换来 9 倍加速 ⇒ 扩展效率约 56%）；500 亿顶点、800 worker = 700 秒。这组数字说明两件事：(a) Pregel 能把”单机不可能”的图变成”分钟级”的作业；(b) 即便在最理想的树形图上，扩展效率也明显次线性——通信与屏障的开销随机器数增长。
24.5.2 同步 vs 异步：收敛速度与确定性的定量权衡
| 指标 | 同步 BSP | 异步 | 实测来源（本讲的实验） |
|---|---|---|---|
| 达到同一精度的模拟时间 | 270 | 120（2.25× 更快） | 24.4.2（60 顶点小世界图，PageRank） |
| 顶点重算次数 | 1080 | 480（2.25× 更少） | 同上 |
| straggler（4 倍慢）的代价 | ×4.00（同步 SGD）/ ×1.33（BSP 图） | ×1.23（SGD）/ ×1.08~1.81（BSP 图，取决于放置） | 24.4.2 / 24.4.3 |
| 确定性 | 完全确定（两次轨迹逐位相同） | 不确定（FIFO 800 vs LIFO 21340 次重算；最终值差 $5.75\times10^{-5}$） | 24.4.2 |
| 收敛性（ML 场景） | 学习率 0.22 仍收敛 | 同一学习率发散（$1.5\times10^7$） | 24.4.3 C 段 |
研究文献的定量参考：GraphLab 系列报告异步相对同步可达 10-100× 加速（补充说明：该数字不在课程讲义中，来自论文；本讲实验在直径较大的图上得到 2.25×，在 ML 场景得到约 12× 的墙钟优势——加速比强烈依赖图的直径、机器异构程度与算法的收敛特性，不应把”10-100×”当成普适常数）。
24.5.3 分布式 ML 的通信瓶颈
| 通信方案 | 每轮每 worker 的通信量 | 总通信量（$N$ 个 worker） | 特点 |
|---|---|---|---|
| 参数服务器（朴素） | pull + push 全部参数 | $\approx 2\vert w\vert N$（但集中在 server 侧，server 带宽 $\propto N$） | server 是瓶颈 ⇒ 必须分片 + 多副本 + 压缩 |
| AllReduce（朴素，每人对所有人多播） | $(N-1)\vert w\vert $ | $N(N-1)\vert w\vert $ | 通信量随 $N^2$ 增长，不可扩展 |
| Ring AllReduce | $2\vert w\vert (N-1)/N$ | $2\vert w\vert (N-1)$ | 带宽最优：每轮的通信量与 $N$ 几乎无关（$\to 2\vert w\vert $），因此成为数据并行的主流实现（Horovod/NCCL） |
| 梯度压缩（1-bit SGD / TernGrad） | $\vert w\vert /32$（1 bit vs 32 bit） | 同比例下降（可达约 30× 压缩比，论文口径） | 需要误差补偿来维持收敛；稀疏化（top-k）在梯度天然稀疏时更有效 |
| Local SGD / FedAvg | $2\vert w\vert /H$ | $2\vert w\vert N/H$ | 用”本地多步”换通信（实测 $H=50$ ⇒ 每条梯度的消息数降 50×） |
带宽/计算比（communication-to-computation ratio）与扩展效率：设单轮计算时间 $T_{\text{comp}}$、通信时间 $T_{\text{comm}}$，则”$N$ 个 worker 的理想加速”为 \(S(N) = \frac{N\,T_{\text{comp}}}{T_{\text{comp}} + T_{\text{comm}}(N)}\) 随着 $N$ 增大，$T_{\text{comm}}$ 通常增长（AllReduce 的 $\log N$ 或线性项、参数服务器的 server 带宽争用），于是扩展效率 $S(N)/N$ 单调下降。要延缓下降只有三条路：减少通信频率（Local SGD / 更大的 batch）、减少通信数据量（压缩、量化、稀疏化）、减少通信次数与提高带宽利用（Ring AllReduce、NVLink/InfiniBand、梯度累积）。这也解释了讲义页 22-27 为什么把 TensorFlow/PyTorch/JAX 的”分布式”能力放在框架层：框架的职责就是把通信与并行的复杂性藏起来（讲义页 23 的原话是”隐藏分布细节”）。
24.5.4 straggler：同步训练最大的敌人
- 为什么在现代集群里尤其严重：大集群中 GPU 降频、ECC 错误、网络抖动、邻居作业抢占、检查点导致的 I/O 尖峰都不可避免；同步 SGD 下最慢的 worker 决定每轮时间（实测 ×4.00 的减速正好等于 straggler 的慢速倍数，因为每轮都要付一次）。而且规模越大，”至少有一台机器慢”的概率越高（$1-(1-p)^N$）。
- 四种应对手段（按侵入性排序）：(1) 异步/SSP——不再等（代价是陈旧度）；(2) backup worker / 冗余计算——与 Lecture 5 的 backup task 同源，用算力换时间；(3) 梯度编码（gradient coding）——用纠删码把”取平均”变成”取任意足够多的份数即可恢复”，从而合法地忽略慢 worker；(4) 弹性训练——把慢/坏节点踢出通信组并按剩余节点数调整 batch 与学习率。
- 一个常被忽略的代价：追求”消除 straggler”的手段往往带来更大的 batch（冗余计算/编码会放大有效 batch），而大 batch 会降低模型泛化能力，需要配合学习率 warmup 与更长的训练——这是”系统优化”与”统计学习”之间的经典冲突。
24.5.5 现代实践清单（把本章所有机制落到工程上）
| 技术 | 解决的问题 | 与本章概念的对应 |
|---|---|---|
| 数据并行 + Ring AllReduce | 通信量随 $N$ 爆炸 | 24.5.3；AllReduce 是”数据并行”的实现之一（讲义页 22） |
| 梯度压缩（量化/稀疏化 + 误差补偿） | 通信带宽 | 24.2.10；参数服务器的 Push/Pull 数据量 |
| 混合并行（3D：数据 × 流水线 × 张量） | 超大模型 + 超大数据 | 24.2.9 的范式组合 |
| 梯度检查点（gradient checkpointing） | 激活值显存不够 | 用计算换内存——与 Pregel/MapReduce 的”重算代替存储”同一哲学 |
| ZeRO（优化器状态/梯度/参数分片） | 每卡都要存完整优化器状态（冗余内存） | 24.2.10 的”参数分片”思想推广到训练状态上 |
| 混合精度训练（FP16/BF16 + FP32 主权重） | 算力与带宽 | 24.5.3 的”压缩”思想在数制层面的体现 |
| 异步/SSP 训练与”踢掉慢节点” | straggler 与陈旧梯度 | 24.2.11、24.3.7 |
| 弹性的重配置（elastic training） | worker 故障 | 24.2.12；对应 Pregel 的”分区重分配” |
24.6 关键要点
- 图处理的根本困难是”迭代 + 不规则 + 动态收敛 + 分割困难”，MapReduce 四条都不匹配：每次迭代都要落盘并重启 job、按块切分导致幂律倾斜、无法表达活跃集合、图分割本身 NP-hard。Pregel 的答案是”图的拓扑常驻内存 + 顶点为中心的 BSP 超级步“。
- “Think like a vertex” 是 Pregel 最大的贡献：程序员只写”一个顶点收到消息后做什么”，系统负责并行、通信、同步与容错。抽象层级的降低，换来的是编程效率与系统优化的解耦。
- BSP 的三条语义必须记牢：(a) 第 $k$ 步发的消息只能在第 $k+1$ 步收到；(b) 投票停机的顶点会被新消息唤醒；(c) 只有”没有活跃顶点且没有在途消息”才终止。由 (a)+(b) 推出 Pregel 的确定性，由确定性推出”检查点 + 重放“这一容错方案的简洁性。
- 全局判据要用 Aggregator，本地判据会破坏不变式：PageRank 用局部停机判据会让 $\sum\text{PR}$ 从 1 泄漏到 0.8136；用全局
max_delta判据才能守恒并全员同时停机。“谁能看到全局状态”决定了算法正确性。 - 图处理与分布式 ML 共享同一个核心矛盾（本章黄金法则）：计算必须频繁交互（图的邻居 / ML 的梯度），因此通信与同步成为瓶颈。BSP 屏障给出确定性与简单性，代价是 straggler 与收敛慢；异步给出速度，代价是确定性与收敛性的损失。SSP 与 Local SGD 都是在这两端之间找折中——SSP 用”有界陈旧度 $s$”换取”接近异步的速度 + 接近同步的收敛性”，Local SGD 用”$H$ 步本地计算”换取”通信量降 $H$ 倍”。实验中这两条折中路线的效果分别是 62.4 → 20.8 的时间提升与 50× 的通信削减。
- “重算”是贯穿全课的容错哲学：MapReduce 重跑失败的 task（Lecture 5），Spark 用 lineage 重算丢失的分区（Lecture 26-B），Pregel 从检查点重放超级步（本章），大模型训练用 gradient checkpointing 省显存（用计算换内存）。它们的共同前提都是”执行是确定性的”——一旦异步执行破坏了确定性，这条最省事的容错路径也就消失了。
- 扩展效率永远次线性：讲义自己的数据就是最好的例子（50→800 worker 只得到 9 倍加速）。所有优化——combine、delta 剪枝、Ring AllReduce、梯度压缩、Local SGD——本质上都在做同一件事：降低”通信与同步”在总时间中的占比。
24.7 常见陷阱与注意事项
- 以为”发送消息”后立刻能被自己或别人看到。错在：Pregel 的消息在超级步结束时才投递，本步发、本步读是读不到的。正确做法：把”发送”当成”写给下一个超级步的自己/邻居”，需要本步内用到邻居的新值时必须显式建模为”下一步再算”（这也是 BSP 确定性的来源）。
- 顶点投票停机后就认为它永远不再参与计算。错在：停机只是”本轮不再主动运行”，只要收到消息就会被唤醒。SSSP 中 v5 正是因为被唤醒才从错误的 10 修正到 6。正确做法：把
VoteToHalt()理解为”等待输入”，而不是”结束”。 - 用”我自己不变了”当作全局收敛判据。错在：顶点停机后不再向邻居发送自己那份贡献，破坏了 PageRank 的权重守恒（实测 $\sum\text{PR}=0.8136$）；连通分量、SSSP 里也会出现”局部看似稳定、全局仍未收敛”。正确做法：用 Aggregator（全局最大残差/全局计数）做停机判据，让所有顶点在同一步一起停。
- 把 Combine 当成”万能的免费优化”。错在：combine 只能合并同一个 worker 的 outbox 里、同一超级步、同一目标的消息；分区越细收益越小（实验：1 个 worker 把 hub 的 12 条消息合成 1 条，4 个 worker 只能合成 4 条）。而且
combine()必须满足交换律与结合律，并且不能丢弃用户真正需要的信息（若用户想逐条看到消息，就不能用 combine）。正确做法：只在”聚合类语义”（sum/min/max）上使用，且先想清楚分区对收益的影响。 - 在异步系统里做”回滚到检查点再重放”的容错。错在：异步执行没有全局一致的超级步边界，两次执行的中间状态不同，重放会得到不同结果（实验里 FIFO 与 LIFO 的最终值差 $5.75\times10^{-5}$，重算次数差 27 倍）。正确做法：异步系统要用日志（logging）、一致性快照或lineage 重算，并且明确”我接受结果不可复现”。
- 异步 SGD 沿用同步 SGD 的学习率。错在：异步的梯度是在 $s$ 个版本之前的参数上算出来的，其误差 $\propto \eta\cdot s$；同一学习率下同步收敛而异步发散（实验：lr=0.22 时同步 0.0181 vs 异步 $1.5\times10^7$）。正确做法：学习率按”最大陈旧度”反比缩小（$\eta = O(1/s)$），或者直接用 SSP 给陈旧度加上界。
- 把”模型并行”当成分片就能线性加速。错在：层间切分把串行依赖引入系统（每层都要跨机通信 + 流水线气泡），实践中模型并行常常比数据并行更慢，它存在的理由是”数据并行放不下模型”，不是”更快”。正确做法：先在数据并行维度扩展；模型放不下时，单机内用张量并行（高带宽 NVLink）、跨机用流水线并行（micro-batch 填气泡），并把数据并行放在最外层。
- 忽略”检查点本身”的成本与”恢复后重复故障”。错在：大模型参数上百 GB，频繁检查点会吃掉大量带宽；而恢复期间又发生故障会让重算不断叠加。正确做法：用 $N^*=\sqrt{2C_{ckpt}/(\lambda c)}$ 估算最优间隔、分片异步写、恢复完成后再打新的检查点、并保留最近 K 份以支持回滚。
- 认为”异步一定比同步快”。错在：异步的每一次计算都更便宜地推进，但总工作量可能更大（LIFO 调度下重算 21340 次 vs FIFO 800 次）；在小直径、低倾斜的图上，同步 BSP 反而更优（同步版每轮全员并行、无调度开销）。正确做法：按图的直径、倾斜程度、机器异构程度选模型，而不是教条地选”异步”。
24.8 思考题（带答案）
Q1（计算推演题）：4 个顶点，$N=4$，$d=0.85$，边为 $1\to2,\ 1\to4,\ 2\to1,\ 2\to3,\ 3\to1,\ 4\to1$。所有顶点初始 PR 为 $0.25$。请用 Pregel 的超级步模型算出超级步 1 与超级步 2 结束时每个顶点的 PR 值，并验证权重守恒。（提示：$(1-d)/N=0.0375$。）
答：先算每个顶点的出度：$\vert \text{Out}(1)\vert =2,\ \vert \text{Out}(2)\vert =2,\ \vert \text{Out}(3)\vert =1,\ \vert \text{Out}(4)\vert =1$；入邻居：$\text{In}(1)=\{2,3,4\},\text{In}(2)=\{1\},\text{In}(3)=\{2\},\text{In}(4)=\{1\}$。 超级步 0：每个顶点发出 $0.25/\vert \text{Out}\vert $：$v_1\to v_2,v_4$ 各 $0.125$；$v_2\to v_1,v_3$ 各 $0.125$；$v_3\to v_1$ 给 $0.25$；$v_4\to v_1$ 给 $0.25$。 超级步 1（这些消息在步末投递，本步被读取）： $v_1 = 0.0375+0.85\times(0.125+0.25+0.25)=0.0375+0.53125=\mathbf{0.568750}$； $v_2 = 0.0375+0.85\times0.125=\mathbf{0.143750}$； $v_3 = 0.0375+0.85\times0.125=\mathbf{0.143750}$； $v_4 = 0.0375+0.85\times0.125=\mathbf{0.143750}$。 校验：$0.56875+3\times0.14375=1.000000$ ✓（权重守恒）。 超级步 2：新的贡献是 $v_1\to 0.284375$（给 $v_2,v_4$）、$v_2\to 0.071875$（给 $v_1,v_3$）、$v_3\to 0.143750$（给 $v_1$）、$v_4\to 0.143750$（给 $v_1$）： $v_1=0.0375+0.85\times(0.071875+0.14375+0.14375)=\mathbf{0.342969}$； $v_2=0.0375+0.85\times0.284375=\mathbf{0.279219}$； $v_3=0.0375+0.85\times0.071875=\mathbf{0.098594}$； $v_4=\mathbf{0.279219}$。校验：$0.342969+0.279219+0.098594+0.279219=1.000001$ ✓。 要点：$v_1$ 在第 1 步虚高到 0.5688、第 2 步又跌回 0.3430（后续 0.4773 → 0.4039 → 0.4420），摆动幅度按 $d=0.85$ 收缩——这正是压缩映射；而”每步 $\sum\text{PR}=1$”正是”所有顶点在同一个超级步一起计算、一起发出贡献”的结果。
Q2（”某个直观但错误的想法错在哪”题）：有位同学为了让 PageRank 更快结束，把 Compute() 改成”只要我自己这一步的变化量小于 $10^{-3}$ 就调用 VoteToHalt()“。他指出”每个顶点最终都会停机，所以系统会终止，结果也应该差不多”。请指出错误，并说明正确的做法与后果。
答：错在把局部稳定当成了全局收敛。顶点一旦停机就不再向邻居发送自己那份 $\text{PR}(u)/\vert \text{Out}(u)\vert $，于是邻居的求和 $\sum_{u\in\text{In}(v)}$ 会少一项，权重不再守恒——实验实测：这个版本跑到 200 个超级步仍不收敛，7 顶点图的 $\sum\text{PR}=0.813566$，权重泄漏 18.6%，并且各顶点值会在”被唤醒 → 重新变小 → 再次停机”之间来回震荡（因为邻居失去贡献后值下降，被下降的邻居唤醒后又把自己加回去）。$\sum\text{PR}=1$ 的归纳证明依赖”所有顶点在同一个超级步同时停机”，局部停机恰好破坏了这个前提；同时”所有顶点都停了”并不等于”结果正确”——它只说明没人再主动发言，而不是答案已经收敛。正确做法：把每个顶点的变化量 $\vert \text{new}-\text{old}\vert $ 通过 Aggregator 归约成全局最大残差 max_delta，在下一个超级步让每个顶点读到同一个全局值，只有当全局 max_delta < eps 时全体一起 VoteToHalt()；这样权重守恒（实验 $\sum\text{PR}=1.000000$）且终止条件真正对应”所有顶点都不再变化”。一般教训：终止判据必须与算法依赖的不变式（这里是权重守恒）一致，而”局部判据 + 全局不变式”的组合是分布式系统里最经典的一类 bug。
Q3（设计权衡题）：你的团队要在一个异构的、按小时租用的云集群（机器性能差异达 3-5 倍，且随时可能被抢占）上训练一个 100 亿参数的推荐模型，训练数据 50 TB。请给出并行方案（数据/模型/流水线/张量如何组合）、同步策略（同步/异步/SSP）与容错方案，并解释每一层选择与本章哪个机制对应。若换成一个”模型只有 1 亿参数、但单次迭代通信极贵（跨数据中心）”的场景，你的方案会怎么变？
答（方案一，异构云 + 百亿参数 + 海量数据）：
- 并行：数据并行（最外层）+ 流水线并行（跨机）+ 张量并行（机内） 的 3D 组合。100 亿参数在 FP16 下约 20 GB，加上优化器状态（Adam 约 3-4 倍）会超过单卡，因此不能纯数据并行；把模型按层切成若干 pipeline stage 放到不同机器（缓解跨机带宽），在每台机器内部用张量并行切分大矩阵（利用 NVLink 高带宽，对应 24.2.9 的”张量并行只在机内用”这一原则），最外层用数据并行扩展吞吐。
- 同步策略：选 SSP（有界延迟） 或”同步 + 踢掉慢节点”。异构 + 可抢占 ⇒ 同步 SGD 会被 straggler 反复拖慢（实测 ×4.00），纯异步又会让 100 亿参数模型的收敛变差（学习中率一大就发散，见 C 段实验）；SSP 用 $s$ 把陈旧度限制在可控范围（实验证明 $s\le10$ 时同一学习率仍收敛）。
- 容错：分片 + 异步的周期性检查点（模型几百 GB，必须分片写、并与训练重叠执行）+ 弹性训练（节点被抢占时缩减通信组、调整 batch 与学习率）+ 冗余/备份 worker 处理偶发 straggler（对应 Lecture 5 的 backup task）。参数服务器侧用分片 + 多副本 + 定期快照（不需要 Paxos 级别的强一致，因为参数分片之间独立）。 方案二（1 亿参数、跨数据中心通信极贵）：瓶颈从”算力/内存”变成”通信轮数“，所以：
- 并行：回到数据并行（1 亿参数约 400 MB，单卡放得下 ⇒ 不需要模型并行，避免引入层间串行通信）。
- 同步策略：用 Local SGD / FedAvg——每轮本地多算 $H$ 步再聚合，把通信量降 $H$ 倍（实验 $H=50$ ⇒ 消息数降 50×），并配合梯度压缩（量化 + 误差补偿）进一步压字节数；$H$ 增大后必须调小学习率（本地漂移）。
- 容错：跨数据中心意味着断连是常态 ⇒ 用”采样参与 + 容忍掉线”的联邦式策略，而不是”全员同步”。 一句话总结：先确定瓶颈在”算力/内存”还是在”通信/同步”，再决定并行维度与同步策略——这正是本章黄金法则的工程版本。