Chandy-Lamport算法核心解读

骑猪看日落 2022-10-31 10:53 152阅读 0赞

参考文章：[Chandy-Lamport算法核心解读][Chandy-Lamport]

![Chandy-Lamport算法核心解读][Chandy-Lamport 1]

## 简介 ##

本文会介绍论文《Distributed Snapshots: Determining Global States of Distributed Systems》以及其中Chandy-Lamport算法的核心原理，原论文地址 [https://www.microsoft.com/en-us/research/uploads/prod/2016/12/Determining-Global-States-of-a-Distributed-System.pdf][https_www.microsoft.com_en-us_research_uploads_prod_2016_12_Determining-Global-States-of-a-Distributed-System.pdf] 。

其中作者之一Leslie Lamport就是Paxos的作者，本论文也是Flink实现分布式snapshot和exactly-once语义的基础，虽然算法意义巨大但其原理并不高深，很多人独立思考也可以想到，只是没有论文中的严格数学证明而已，作者也认为其核心思想很straighforward。

## 算法原理 ##

Chandy-Lamport算法的目标是让多个分布式节点能一起完成snapshot保存全局状态。我们知道单机要完成一次snapshot非常简单，在特定时间停止处理新数据然后dump内存状态并记录时间点即可，而分布式系统在多个节点上运行，而且互相之间都有通信，要让所有节点“同时”dump内存保存所有状态信息也是不可能的。

在计算机领域，使用锁来协调任务之间的执行顺序也是常用的方法，分布式snapshot理论上也可以通过依赖分布式锁服务来保证每个节点按顺序来做本地snapshot，但通过锁来限制读写和snapshot会导致性能非常差，lock-free是更好的方案。Chandy-Lamport算法就是通过巧妙的算法不依赖锁来实现分布式snapshot的，因此使用此算法不依赖分布式锁服务或者共享内存。但注意，Chandy-Lamport算法要求节点间通信可靠并且消息有序，在网络传输层可以使用TCP来保证，如果使用UDP消息可能乱序那么这个算法保存得到的全局状态很有可能也是有问题的。

说了这么多，这个算法基本原理就是让某个节点做global snapshot时发送一个marker，每个节点在处理数据时遇到marker执行下面的判断来做local snapshot，最后所有节点接收到marker并正确处理后就可以组成一个global snapshot，核心算法就是下面几句。

![b00782b13ad0fa4334a6fb4f4b17d1cd.png][]

由于公式需要定义多个节点p、q以及传输的状态c等，不理解的人看完应该还是不理解，这里通过算法执行的流程演示来解读下。

## 算法演示 ##

首先定义场景一，集群只有节点P和Q两个节点，并且由P来发起global snapshot，P先做完local snapshot，然后把marker信息传给了Q，如下图。

![1d4e718266e24ce45494a75e5c6038da.png][]

Q接受到marker后，由于自己还没有做local snapshot（对应论文has not recorded its state），因此按前面算法的逻辑就会先做local snapshot，把状态都保存后会把marker继续传下去，这里因为只有两个节点就会把marker传回给P，如下图。

![fb72697fa9e0bceafa56ff897d644d68.png][]

P在接收到marker后，按照前面的算法，已经做过local snapshot，那么记录一些message信息，这里除了marker没有其他信息，因此所有节点marker都处理完并且都保存了local snapshot，这样包括所有local snapshot的global snapshot就完成了。

场景一非常简单，因为节点间没有错误而且只传递了marker信息，那么我们定义更复杂的场景二，节点P在连续发送m1、m2消息，并且Q在连续发送n1、n2消息，如下图。

![85a13fc0650bdbc132c7ce013de69b0a.png][]

此时P要发起一次global snapshot，那么按照前面的算法由于P还没有做local snapshot，因此P会先做local snapshot把内存状态保存下来，并且把marker消息添加到输出信息中，这时由于已经完成local snapshot因此不能继续消费n1、n2而是记录下来，如下图。

![10e2f5af092794756ce1fda11361602a.png][]

然后Q继续消费P传送的消息，这里就用到前面提到的前提，消息是可靠并且保序的，Q可以按顺序处理m1、m2消息，然后处理marker消息。由于这是分布式系统，P和Q不会停止服务，假设P这时还发送了m3、m4消息，而Q在处理m1、m2后就可能继续给下游节点P发送n3、n4消息，如下图。

![986f08118a7ddb946f5a76aa16c23656.png][]

这时Q就准备处理marker消息了，按照前面的算法，由于Q还没有保存状态，因此也是先做local snapshot，然后要给下游节点P发送marker消息，如下图。

![d3a113ead60d958bde719e9e48c8676e.png][]

这时候Q已经接受marker并且完成local snapshot了，像Flink这类应用就会把Q的snapshot（或者称为checkpoint）上传到统一的snapshot管理服务，等所有节点都完成后整个global snapshot才算完成。

P虽然完成了local snapshot但没有收到marker消息，按照前面消息是可靠并且保序的背景，P会先处理消息n3和n4，最后才处理marker，按前面的算法描述这时已经做完local snapshot的P需要记录所有来自input channel的数据，这时Q也可以继续处理m3、m4准备下一轮的snapshot，如下图。

![ac180358cbe16fb90b1965b2b81df848.png][]

最后P也接收到marker了，这时P需要把local snapshot和所有缓存的消息n1、n2、n3、n4一起上传snapshot管理节点，待那个服务确认所有节点的snapshot都完成后就把整个global snapshot序列化保存下来。

在实际应用中，snapshot是一直在运行的，因此每个marker都会带snapshot版本信息，如果某个节点发生故障了，那么所有节点都需要放弃当前处理的数据以及当前状态，统一从上个成功的global snapshot中恢复。那么可以想像一下，在前面已经保存global snapshot的情况下，如果P和Q继续传递消息，不管是哪个节点发生故障，故障时是否有local snapshot，实际上都可以恢复到当前的状态继续服务。

在一个更复杂的场景三，每个节点的输入和输出都有可能有多个节点，对于多个输出，只需要输出marker时输出多个统一snapshot版本的marker即可。而多输入的情况会复杂一些，在强一致的场景（例如Flink的exactly-once语义），每个节点是需要等待所有input channel的marker到达后才做local snaphot和处理marker后续数据，因此节点的性能取决于最慢的input channel，服务整体吞吐和性能也会降低。而在非强一致的场景可以，允许节点一直处理input channel中到达的数据，等到所有marker到达才做local snapshot，这样部分marker后的数据也会加到local snapshot里面，如果节点发生故障从global snapshot恢复后还会继续算这部分marker以后的数据，这也是Flink中在不强制同步marker处理只能保证at-least-once语义的原因，当然这样服务的吞吐和性能会更高。多input/output channel场景如下图，S1表示Step1。

![98fcef02e7adccadf0d01dc55bf48294.png][]

## 总结 ##

> 最后可以总结下，Chandy-Lamport算法确实很straighforward，就是节点在不断发消息的同时做local snapshot和发送marker（Flink中称为barrier个人觉得更加直观），然后所有节点都处理marker前的数据，marker后的数据先缓存记录，等所有节点都处理完marker前的数据并且都导出自己的local snapshot后，这个global snapshot就完成了，而且保证所有节点都处理完marker前的数据。
> 
> 在前面的场景二中，节点P按顺序发送了消息m1, m2, marker, m3, m4，并且在发送marker的时候完成local snapshot，因此global snapshot就包含了节点P节点发送完m1、m2信息时的状态，并且保存了P待处理的消息包括n1, n2, n3, n4。而节点Q按顺序发送了消息n1, n2, n3, n4，接受了消息m1, m2, marker, m3, m4，在处理完m1、m2消息后完成local snapshot，因此global snapshot也就包含了节点Q处理完m1、m2消息时的状态，而在整个系统中Q接收到marker时已经发出n1, n2, n3, n4这些信息也在global snapshot中可用于恢复。
> 
> 对于已经理解Chandy-Lamport算法的人，这篇论文其实几句话就可以说清楚了，而不理解的人除了论文可能还需要找更多资料反复阅读才行，我们希望帮忙大家把论文中“能说清楚的几句话”找出来，在理解论文的亮点和核心的情况下更好地阅读论文，有所侧重地了解更多论文的背景与细节。

[Chandy-Lamport]: https://zhuanlan.zhihu.com/p/113612578
[Chandy-Lamport 1]: /images/20221024/ea610cd1221b4550bfff088280de33fc.png
[https_www.microsoft.com_en-us_research_uploads_prod_2016_12_Determining-Global-States-of-a-Distributed-System.pdf]: https://link.zhihu.com/?target=https%3A//www.microsoft.com/en-us/research/uploads/prod/2016/12/Determining-Global-States-of-a-Distributed-System.pdf
[b00782b13ad0fa4334a6fb4f4b17d1cd.png]: /images/20221024/2109c2bba02f4846aca39c0a2495953b.png
[1d4e718266e24ce45494a75e5c6038da.png]: /images/20221024/3d10af986bc04fa5a072e497469289ff.png
[fb72697fa9e0bceafa56ff897d644d68.png]: /images/20221024/f98bc00cc4f9497ab80d423e0afef1e1.png
[85a13fc0650bdbc132c7ce013de69b0a.png]: /images/20221024/349b10aa2057453eb19219d830ca8ec9.png
[10e2f5af092794756ce1fda11361602a.png]: /images/20221024/0c8a757aeb584703abbcc9d04c5529bc.png
[986f08118a7ddb946f5a76aa16c23656.png]: /images/20221024/5ace59afe4e745d4af11a8e89115964a.png
[d3a113ead60d958bde719e9e48c8676e.png]: /images/20221024/bc201d8e8c1845ba920ce0197884619d.png
[ac180358cbe16fb90b1965b2b81df848.png]: /images/20221024/51957a9f572344d2814af81292b36139.png
[98fcef02e7adccadf0d01dc55bf48294.png]: /images/20221024/6484d7db5def4b1492bfca197dfcbdb0.png