Agreement in Asynchronous Systems

Implementable Agreement Abstractions Despite Asynchrony and a Minority of Crashes

在本章中，我们介绍一种栈方法来实现一种 read/write-implementable agreement abstraction 在 $ CAMP_{n,t}[t < n/2] $ 模型下。我们将实现两种 abstraction，分别是 renaming 和 approximate agreement。

The Renaming Agreement Abstraction

在之前的文章中，我们默认了进程$p_i$的 identity $id_i$ 就是它的下标$i$，但是在这里，这两个将不会被看作是等价的。下标仅仅能够用来寻找进程。每个进程都有一个唯一的$id_i$，可以用一个值来唯一限定它（如 IP 地址），且我们有$id_i \in \{1,...,N\}$，$N$是命名空间的大小，通常来说$n$远小于$N$。

Renaming 抽象只有一个操作new_name()，它的具体定义如下：

• R-termination：一个 non-faulty 进程执行new_name()后会终止。
• R-validity：一个新名字是在$[1..M]$中的一个数。
• R-agreement：没有不同的进程返回了相同的名字。
• R-index independence：对于任意的两个进程，如果一个进程得到了新名字$v$，那么另一个进程也有能力得到$v$。（该性质保证了新名字与索引无关）

我们记参加到 renaming 的进程数为$p$，显然我们有$ M \geq p $，更进一步，如果$M$仅仅取决于$p$，那么我们称该算法是大小自适应(size-adaptivity)的，这样我们就有$ M = f(p), f(1) = 1, \forall p, 2 \leq p \leq n: p-1 \leq f(p-1) \leq f(p) $。对于非大小自适应的，$M=2n-1$是该算法的下界，而对于大小自适应的，$M=2p-1$是该算法的下界。

下面我们就来实现大小自适应的 renaming abstraction，正如上述所说，我们用到了栈方法，即该抽象是基于另一个抽象之上的，在这里我们基于 snapshot abstraction。我们将 snapshot object 记为$STATE$，每个原子寄存器$STATE[i]$是被两个值组成的 pair，第一个是$STATE[i].init\_id$，包含了$p_i$的初始名字，其次是$STATE[i].prop$，包含了上一次new_name()的 proposal，每个$STATE[i]$都初始化为$\langle \perp, \perp \rangle$。伪代码如下：

func new_name(id_i):
    prop_i = 1
    while true:
        STATE.write(i, <id_i, prop_i>)
        state_i = STATE.snapshot()
        if for all j != i, state_i[j].prop != prop_i:
            return prop_i
        else:
            let set1 = {state_i[j].prop | state_i[j].prop != none AND i <= j <= n}
            let free = {1, 2,..., M} - set1
            let set2 = {state_i[j].init_id | state_i[j].init_id != none AND i <= j <= n}
            let r = rank of id_i in set2
            prop_i = free[r]

伪代码很容易理解，不过我们还是来解释一下。考虑上述代码的主要谓词：
$$ \forall j \neq i: state_i[j].prop \neq prop_i $$

• 当该谓词为真时，意味着没有其他进程与其竞争该名字$prop_i$，那么自然可以返回该名字。
• 当该谓词为假时，意味着有其他进程与其竞争该名字$prop_i$，首先我们根据统一的全局状态得到$set1$，即，所有新名字的提议，和$set2$，即，所有被$p_i$看作竞争对手的进程的初始名字。在这里$set1$被用作排除已经被使用的名字，而$set2$被用作来决定一个符合要求的 slot，因此$free$是一个符合要求的名字集合，随后$p_i$计算想要获得名字的进程中的排名$r$，并决定提议$prop_i$为$free[r]$。

The Approximate Agreement Abstraction

下面我们介绍一种 agreement，称为 approximate agreement，它的具体定义如下：

• AA-termiantion：一个 non-faulty 进程执行decide()后会终止。
• AA-validity：记$vmin$为所有进程中最小的提议值，$vmax$为所有进程中最大的提议值，那么有$vmin \leq v_i \leq vmax$。
• AA-agreement：对于任意两个进程的提议值，我们有$|v_i - v_j| \leq \epsilon$。

如定义所见，每个进程都提议不同的值是被允许的，此外，每个进程 propose 的值都必须在区间$ [x,\ (x+D)] $中，其中$D$已知，$x$未知。算法构建在 snapshot object 上。具体伪代码如下：

func propose(v_i):
    est_i = v_i
    r_i = 0
    R = 1 + log_2(floor(D/epsilon))
    for r in {1, ..., R}:
        r_i = r_i + 1
        SNAP[r_i].write(i, est_i)
        val_i = SNAP[r_i].snapshot()
        est_i = (min(val_i) + max(val_i)) / 2
    return est_i

在随后我们会看到，consensus 不能在$CAMP_{n,t}[t \geq n/2]$模型下甚至是$CAMP_{n,t}[t = 1]$模型下实现，但是 approximate agreement 可以在$CAMP_{n,t}[t < n/2]$模型下实现。

Consensus: Power and Implementability Limit in Asynchronous Systems

The Total Order Broadcast Abstraction

我们介绍另一种 broadcast abstraction，称为 total order broadcast，它相较于 urb-broadcast，多了一个性质：

• TO-delivery：如果一个进程先 to-deliver 了消息 m，随后又 to-deliver 了消息 m'，那么没有进程先 to-deliver 了 m'，然后 to-deliver 了 m。

与 urb-broadcast 类似，我们可以推导出每个 non-faulty 进程的 deliver 消息序列都相同，而一个 faulty 进程的 deliver 消息序列是上述的一个序列的前缀。

实际上，TO-broadcast 和 Consensus 是等价的。我们可以通过 TO-broadcast 来实现 Consensus，而 Consensus 也可以用来实现 TO-broadcast。下面是在$CAMP_{n,t}[TO-broadcast]$下实现 Consensus 的伪代码：

func propose(v_i):
    TO_broadcast(v_i)
    wait(the first message m delivered by TO-delivery)
    return m

上述代码的正确性极其显然。

The State Machine Approach

现实中，一个系统会提供具有各种服务的客户端，一个服务通常被定义为客户端可以执行的一组命令（或者请求）。而一个命令（请求）可以导致服务状态的变化或者进行输出，而具体的输出则完全取决于初始状态和命令的序列。我们将这样的服务称为状态机。

如果这个服务在单体机器上被实现，那么机器的崩溃对于服务来说是致命的，所以一个自然的想法就是复制这个服务到各个机器上，更正规的讲，一个 state machine replication 技巧就是使得一个服务能够是 client fault-tolerant 的方法。理想化的，我们期望这个复制对于客户端来看是透明的。而上述的主要问题就在于要保持每台机器都执行相同的命令序列，因此，可以使用 TO-broadcast 来实现这一点。

一个这样的 object，被定义为一个初始状态$s_0$，一组有限的操作集合和一个任意的上述集合元素的序列。如果每个操作都可以在该 object 的任意状态下执行，那么我们称这个 object 的操作是 total 的。每个操作都可以看作$op_x(param_x,\ result_x)$，其中$param_x$是$op_m$的输入参数，而$result_x$就是该操作的返回值。我们也可以定义一个如下转化函数：$\delta(s, op_x(param_x)) = \langle s', result_x \rangle$，其中$s'$是执行了$op_x(param_x)$后的状态。具体伪代码如下：

when op(param) is locally invoked by the client:
    result_i = none
    let msg = <op(param), i>
    TO_broadcast(msg)
    wait(result_i != none)
    return result_i

background task T:
    while true:
        msg = TO_deliver()
        <state_i, res> = delta(state_i, msg.op(param))
        if msg.proc == i:
            result_i = res

Ledger Object

一个由加密货币启发的 object，称为 ledger object，它提供了两个操作，分别为read和append，执行read返回一个目前列表的状态的一个副本，而执行append则将一个新的状态添加到列表中。具体的定义如下：

• 如果一个进程执行了append或者read且没有崩溃，那么它一定会终止。
• 方法read与append被看作是被顺序执行的（Atomic Register 的相关定义）。
• read的返回列表是从第一个记录到最后一个在该read执行之前 append 到账本上的的记录的一个副本。

ledger 似乎和 read/write register 很相似，但是一个重要的区别是，由于并发，read/write register 允许一个值 v 被在没有进程读过的情况下被覆盖，就像从来没有进程执行过write(v)一样，但是账本会记录所有的状态并保存在其中。

而与状态机相比，状态机无需保存操作的执行序列，且只需保存它的最终状态。而 ledger object 的最终状态就是整个操作序列，因此，账本有能力让进程检查它的历史记录，查看是否有特定的操作被执行过。

账本可以在$CAMP_{n,t}[TO-broadcast]$模型下实现。

The Frontier Between Read/Write Registers and Consensus

首先我们需要知道的是，consensus agreement 不能够在$CAMP_{n,t}[t \geq 1]$模型下实现，这就是著名的 FLP 不可能性定理。有兴趣的读者可以参考证明原文：Impossibility of Distributed Consensus with One Faulty Process。随后，我们考虑如下问题：我们在上文介绍了的 object 有些可以在$CAMP_{n,t}[t < n/2]$中实现，而有些可以在$CAMP_{n,t}[CONS]$中实现，由于 FLP 不可能性定理，我们知道$CAMP_{n,t}[CONS]$是远远比$CAMP_{n,t}[t < n/2]$更强的模型，那么究竟两者的边界在哪里呢？或者说，我们怎么知道某种 object 可以在上述两种模型中实现呢？下面我们介绍 Consensus number。

Consensus Number

一个并发 object 的 consensus number 是一个最大的 t，使得 consensus 可以在异步系统中，由任意个 read/write register 和这样的 object 对于任意的 t 个进程实现，但是不能在 t+1 个进程中实现。我们将 consensus number 记为$CN(obj)$。对于常见 object 的 consensus number，read/write register 的 consensus number 为 1，stack 和 queue 的 consensus number 为 2，ledger 的 consensus number 为$+\infty$。我们可以观察到，由于 FLP 不可能定理，$CN(obj) \geq 2$的 object 不能在$CAMP_{n,t}[t < n/2]$中实现。

Implementing Consensus in Enriched Crash-Prone Asynchronous Systems

我们在上文已经知道 Consensus 不能在$CAMP_{n,t}[t \geq 1]$模型下实现，因此，我们需要对模型进行增强来获得实现 Consensus 的能力。实际上有三种常见的增强模型，分别是：

• 第一种方法是对消息的传递增加一个假设，即，假设有这么一个回合，进程从相同的一组 correct 进程得到消息。
• 第二种方法对 failure 的信息进行了提供，我们可以使用 eventual leader 来实现共识。
• 第三种方法我们给模型增加了随机性，每个进程都被允许抽取一个随机数。

在这里，我们只介绍第二种方法的实现，并且再介绍一个相似的、由 Paxos 启发的算法。

Enriching $CAMP_{n,t}[\emptyset]$ with a Perfect Failure Detector

我们首先介绍第一种 failure detector，称为 perfect failure detector，它的定义我们前文都已经介绍过了，在这里不再重复。伪代码如下：

func propose(v_i):
    est_i = v_i
    r_i = 1
    while r_1 <= t+1:
        if r_i == i:
            broadcast(EST(est_i))
        wait(EST(est) received from p_r_i OR r_i in suspected_i)
        if EST(est_r_i) is received:
            est_i = est_r_i
        r_i = r_i + 1
    return est_i

我们来证明一下算法的正确性：

首先，算法的 CC-validity 是显然的，我们首先证明 CC-termination。让我们观察第一轮，如果$p_1$是 non-faulty 的且它执行了广播操作，那么消息最终会被所有 non-faulty 进程接收到，如果$p_1$崩溃了，那么最终我们有$1 \in suspected_i$，因此，所有进程都不会在wait中被永远阻塞。类似的，我们可以证明第二轮、第三轮...，因此，我们有 CC-termination。

我们再证明 CC-agreement。由于 t 是可能出错进程数量的一个上界，那么在 t+1 个进程中一定有一个 correct 进程，假设第一个这样的进程是$p_x$，由于 CC-termination，我们知道它一定会进入到第$x$轮并广播它的估计值。因此，所有执行完第$x$轮的进程都会得到相同的估计值，而在随后的轮数中，没有其他的估计值会被广播，因此我们有 CC-agreement。

Enriching $CAMP_{n,t}[t < n/2]$ with an Eventual Leader

我们首先介绍一个新的 failure detector，称为 eventual leader，它的定义如下：

• validity: $\forall i: \forall \tau: leader_i^\tau$ contains a process identity.
• eventual leadership: $\exists \ell \in Correct(F), \exists \tau: \forall \tau' \geq \tau: \forall i \in Correct(F): leader_i^{\tau'} = \ell $

伪代码如下：

func propose(v_i):
    est1_i = v_1
    r_i = 0
    while true:
        r_i = r_i + 1
        # Phase 1
        my_leader_i = leader_i
        broadcast(PHASE1(r_i, est1_i, my_leader_i))
        wait(
            PHASE1(r_i, -, -) received from (n-t) processes AND
            (PHASE1(r_i, -, -) received from p_my_leader_i OR
            my_leader_i != leader_i)
        )
        if (exist l: PHASE1(r_i, -, l) received from > n/2 processes) AND ((r_i, v, -) recieved from p_l):
            est2_i = v
        else:
            est2_i = none

        # Phase 2
        broadcast(PHASE2(r_i, est2_i))
        wait(PHASE2(r_i, -) received from (n-t) processes)
        let rec_i = {est2 | PHASE2(r_i, est2) have been received}
        switch:
            case rec_i = {v}:
                broadcast(DECIDE(v))
                return v
            case rec_i = {v, none}:
                est1_i = v
            case rec_i = {none}:
                continue

when DECIDE(v) is received:
    broadcast(DECIDE(v))
    return v

我们来讲解一下，首先，PHASE1 期望所有进程在执行 PHASE2 之前都有着相同的估计值，我们可以观察到，当 eventual leader 被成功选取时，我们就可以达成这个目的。因此，PHASE1 的主要目的就变为了保证算法的 safety property(no two different values are decided)，下面我们介绍一种 property，称为 quasi-agreement，它的定义如下：
$$ ((est2_i^r \neq\ \perp)\ \land (est2_j^r \neq\ \perp)) \Rightarrow (est2_i^r = est2_j^r = v) $$
该 property 可以在保证 safety 的情况下，为 PHASE2 的判断做出保证，下面我们来具体分析算法 PHASE1 的行为：

• 首先$p_i$读取它的 leader，并广播，需要注意的是，broadcast 是一个宏因此不是 reliable 的。
• 随后$p_i$等待 n-t 个消息，注意到，由于 t 个进程可能崩溃，因此这是最大的可收到的且不会被永远阻塞的消息数量。且由于$t < n/2$，因此 n-t 个进程中包含了大多数进程，即包含了至少一个正确的进程。同时，进程$p_i$等待它所认为的 leader 的消息，或者它的 leader 发生了变化。

• 下面我们就要考虑如何保证 quasi-agreement，如果有这样一个进程$p_{\ell}$，使得：

  1. 大多数进程都认为$p_{\ell}$是 leader（由它们发送的消息得知）。
  2. 被进程$p_{\ell}$发送的消息被一个进程接收到。

  那么$p_i$就会将它的估计值设置为$p_{\ell}$发送的消息的值，否则，$p_i$就会将它的估计值设置为 none。由于对任意两个 majority 集合，它们两个集合不可能认为不同的进程是 leader，因此我们有 quasi-agreement。

至于 PHASE2 就很清晰易懂，我们在这里不再赘述。

A Paxos-Inspired Consensus Algorithm

最后我们来介绍一种由 Paxos 启发的共识算法，Paxos 算法的原文：The Part-Time Parliament参考如上。我们得以在$CAMP_{n,t}[t < n/2, \Omega]$中实现它，首先，我们来介绍一种新的通信抽象，称为 Alpha Communication Abstract，它只有一种操作alpha(r, v)，它的定义如下：

• Alpha-validity: 方法alpha(r, v)返回的值要么是$\perp$，要么是值$v'$，其中有这样一个回合$r' \leq r$，且alpha(r', v')被执行过。
• Alpha-agreement: 如果alpha(r, -)和alpha(r', -)都被执行过且返回了$v$和$v'$，那么$((v \neq \perp) \land (v' \neq \perp)) \Rightarrow (v = v')$。
• Alpha-convergence: 如果对于 I = alpha(r, -)，任意其他操作 I' = alpha(r', -)在 I 终止之前开始执行，且$r' \leq r$，那么 I 返回一个非$\perp$的值。
• Alpha-termination: 如果一个 non-faulty 进程执行了alpha(r, -)，那么它一定会终止。

因此，一个 Consensus 可以被如下构建：

func propose(v_i):
    r_i = 0
    while res_i == none:
        if leader_i == i:
            res = ALPHA.alpha(r+i, v_i)
            if res != none:
                broadcast(DECIDE(res))
                res_i = res
            else:
                r_i = r_i + n
    return res_i

when DECIDE(v) is received:
    broadcast(DECIDE(v))
    return v