使用分布式一緻性算法替代2PC/3PC中的TM，能達到容錯的分布式事務送出算法。
改算法使用Paxos和2PC高度融合，達到和2PC一樣的延時。
1. 分布式事務介紹
分布式事務是在多個節點上執行一序列操作，這些操作後緊跟着一個操作，标記要麼commit，要麼abort。各個節點執行事務送出協定來決定這個事務是應該commit還是abort。事務能夠送出當且僅當所有的節點都同意送出。在分布式系統中實作這種all-or-nothgin的原子性是不實際的。
經典的事務送出協定是2PC：使用一個單一節點的coordinator（協調者）來達成一緻。一旦協調者出現故障，會阻塞整個2PC的執行。我們使用Paxos一緻性算法，通過多個協調者來實作分布式事務的送出，同時性能上達到理論上的最優值。它能夠保證隻要多數派協調者正常工作，分布式事務的送出就可以正常的執行下去。
算法模型
- Compute model：假設算法是在多個程序上執行，程序之間通過互動消息進行同步，程序通過持久化避免failure；
- Cost model：計算節點之間的消息互動，消息延遲，持久化的寫延遲。假設同一節點上程序之間的消息延遲可以忽略不計；
- Failure model：節點和程序failure；消息丢包重發亂序；
- 拜占庭：不處理比特反轉等拜占庭錯誤；
一般一個算法的正确性包括兩個方面：safety和liveness
- Safety：描述算法哪些行為是允許發生的（達成的commit之後不能丢失）；
- Liveness：描述算法哪些行為是一定要發生的（比如最終會達成commit）；
2. 事務送出
在分布式系統中，事務是有一組運作在不同節點上的resource manager執行(簡稱RM)。當某個RM發射commit或者abort請求後事務就執行結束了。當且僅當所有的RM都同意執行commit，事務才得以commit，否則事務就abort了。事務的最基本的特性是所有的RM要麼全部同意commit，要麼全部同意abort，這也是事務的一個不變式。
在工程實踐中，一個RM必須先加入事務的運作過程。假設參與事務的RM數目是固定的。
假設所有的RM開始的狀态是'working'，算法的目标是所有的RM要麼全部是'committed'，要麼全部是'aborted'狀态。
算法的Safety展現在：
- Stability： 一旦一個RM進入comitted或者aborted狀态，這個RM就一直停留在這個狀态上；
- Consistency：不存在一個RM是commited狀态，另一個RM是aborted狀态；
此外，RM還有一個prepared狀态：
- 一個RM要進入committed狀态，當且僅當所有的RM都進入過prepared狀态。
事務最終進入commit，須經過以下時序：
- 所有的RM以任意順序進入prepared狀态；
- 所有的RM以任意順序進入committed狀态；
事務最終進入abort，須經過以下時序：
- 任意RM在working狀态可以進入aborted狀态；
當一個RM發現不能完成自己的那部分事務時就進入abort狀态。
另外，FLP定理指出在一個确定的純異步的環境下是不可能達成stability和consistency，是以本算法假設消息是最終在一定時間内可達到的。事務送出協定的Liveness如下：
- Non-Triviality：如果在算法執行過程中網絡沒有故障，那麼：
1. 所有的RM都會進入prepared，然後都會進入committed；
2. 某個RM進入aborted，其他的RM最終都會進入aborted；
- Non-Blocking：任意時刻，隻要有足夠多的網絡節點不發生故障，那麼所有RM的最終要麼是committed要麼是aborted；足夠多是保證一個消息的response最終能夠達到。
用TLA+精确地定義事務送出。兩個狀态謂詞：
1. canCommit：為TRUE的條件是對于任意一個RM要麼是prepared要麼是committ；
2. notCommitted： 為TRUE的條件是沒有RM是committed；
next-state描述任意RM所有可能的狀态轉換：
1. Prepare：任意RM可以從working狀态進入prepared狀态；
-
Decide：對于任意RM，
a）如果在prepared狀态并且此時的canCommit為TRUE，那麼這個RM就可以進入committed狀态；
b）如果RM在working或者prepared并且此時的notCommitted為TRUE，那麼就進入aborted狀态；
3 Two-Phase Commit
3.1 2PC協定
2PC送出協定是一個事務送出協定，通過一個TM做為協調者來推動所有的RM在事務送出協定上的前進。TM的狀态有：
1. init；
2. preparing；
3. committed；
4. aborted；
2PC算法流程：
1. 某個RM進入prepared狀态，并給TM發送prepared消息；
2. TM進入preparing狀态并向其他RM發送prepared消息；
3. RM收到prepared消息後，如果此時是working狀态，則進入prepared狀态，并給TM回prepared消息；
4. 如果TM收到的所有RM消息都是prepared，則進入committed狀态，并給所有的RM發送committed消息；
5. 否則，TM進入aborted狀态，并給所有RM發送aborted消息，RM收到aborted消息後進入aborted狀态；
像其他的異步算法處理failure一樣，2PC執行過程中，發送任意消息前都會把目前狀态持久化。比如TM收到一個prepared消息後，先持久化prepared狀态在媒體上，然後在記憶體中進入prepared狀态。當發生failure，TM重新開機後從媒體上讀取到之前的狀态，并繼續執行2PC算法。理論上，當機和機器pause是等價的。
3.2 2PC的複雜度
事務送出協定複雜度的重要名額是算法在正常運作到committed狀态的消息數。
1. 初始RM給TM發送prepard消息 (1 message)；
2. TM給其他RM發送prepared消息 (N-1message)；
3. 每個RM給TM回複prepared的response消息 (N-1 message)；
4. TM給所有的RM發送Committed消息 (Nmessage)；
可以看到：
- ***一個RM要進入Committed狀态需要經過4個relay***；
- 總共有3N-1消息被發送；
- 如果TM和發起2PC的RM在同一個node上，可以省掉TM和這個RM的2個消息，那麼總共有3N-3個。
3.3 2PC的問題
2PC算法可以正确的處理RM的failure。比如：算法執行過程中，某個RM當機，TM必須等到所有RM的回複才能做出決定，那麼此時的TM會因為收不到RM的響應而逾時，TM向所有的RM發送aborted消息。其他的RM進入aborted狀态。如果TM發生了failure，所有的RM收不到aborted消息或者committed消息，就不知道如何進行下去了，TM阻塞了2PC的運作。
非阻塞的事務送出協定：即使TM出現failure，也不阻塞事務的運作。這些協定稱為3PC，通常在一個TM出現failure後，選擇另外一個TM繼續執行協定。然後，這些3PC的正确性都沒有得到證明。
4. Paxos Commit
4.1 Paxos一緻性算法
paxos算法是多數派算法，算法保證最多隻有一個v最終choosen出來。在節點數為2F+1的叢集中，可以容忍F個節點failure。算法也分成兩個階段：
1. Phase 1a： leader選擇一個ballot來構造prepare消息進行廣播，期望獲得其他acceptor節點授權對該ballot的執行權利；
2. Phase 1b： acceptor收到prepare消息後
a. 如果沒有把執行權承諾給更大的ballot，就承諾給目前這個ballot，并持久化ballot，承諾後續不在接受比ballot小的執行權，同時傳回此時acceptor已經接受的v；
b. 否則，就拒絕這個balotl的執行權；
3. Phase 2a：leader在收到多數acceptor的響應後，兩種可能的結果
a. Free：沒有任何acceptor發出過phase2本消息，也就是目前為止沒有choosen一個v；此時，leader可以發射出自己v，通常選擇client發過來的第一個v；
b. Forced：某些acceptor曾經發出過2b消息，那麼選出最大ballot對應的v；此時的這個v可能已經達成了一緻意見，leader能做的就是幫助其他acceptor也對這個v達成一緻意見。使用v發射請求；
4. Phase 2b：當acceptor收到<balotl, v>，如果沒有承諾過比ballot更大的執行權，就執行這個<ballot, v>；否則忽略；
5. Phase 3： 當leader收到多數派的Phase2b響應後，就可以得出結論：v已經被多數派達成一緻意見；
當ballot=0時可以跳過phase1，是以不會有比ballot=0更小的v達成一緻意見。此時直接發送自己的v。phase1的作用有兩個：
1. 得到ballot的執行權；
2. 得到此時可能已經達成一緻的v；
paxos算法可以從兩方面優化：
1. phase 2a消息隻發送到多數派上；
2. phase 2b消息直接回給client；
4.2 Paxos Commit
在2PC的算法中TM會出現單點故障。可以借助多副本的一緻性算法，TM充當client對要做出的決定（committed/aborted）發射出proposal，一緻性算法最終決議出一個值。Mohan, Strong, and Finkelstein提出了同步算法，TM等待所有的RM回複prepare消息後再發射proposal。但是，等待RM回複給TM消息要多一個消息的延遲。下面讨論Paxos Commit算法是如何消除掉這個消息延遲的。
Paxos Commit算法使用獨立的Paxos instance，即每個RM對應一個Paxos instance，都可以獨立的發出committed/aborted的proposal。最終隻有當所有instance被choosen的值都是committed，事務才會被committed。這種每個RM一個instance的想法可以應用到所有的一緻性算法上，但是如何優化時延就和具體的算法息息相關了。
Paxos Commit算法使用2F+1個acceptors和一個current leader。是以，Paxos Commit算法的角色有：
1. N個RM；
2. 2F+1個acceptor；
3. 一個leader；
我們假設RM知道所有acceptor的資訊(比如IP)。Paxos算法中ballot=0的phase2a消息可以標明任意的v。通常是由leader發送phase2a消息，顯然，可以通過事先標明任意角色發送phase2a消息，Paxos算法任然能夠正确運作。在Paxos Commit中，RM各自發送phase2a消息，標明的value為committed或者aborted。
Paxos Commit算法的執行過程如下：
1. 任意一個想要進行事務送出的RM發送BeginCommit消息給leader；
2. leader向所有其他RM發送Prepare消息；
3. 如果RM決定要參與這個事務的commit，就發送消息給所有的acceptor；
4. 否則，RM發送消息給所有的acceptor；
5. acceptor每收到phase2a消息，都向leader回複phase2b消息（Paxos Commit算法比其他算法優化的地方）；
6. leader可以知道：對于某個RM的instance，如果收到了F+1個phase2b的回複，而且這些回複的value是一緻的，那麼根據Paxos算法，就可以認為choosen了一個值；
7. 如果leader确定了所有RM的instance值，并且全部都是Prepared，則事務可以送出，leader向所有的RM發送commit消息，促使RM進入committed；
8. 如果leader确定了所有RM的instance值，并且有一個是Aborted，則事務可以放棄執行，leader向所有的RM發送abort消息，促使RM進入Aborted；
上述算法還有優化的空間：
1. acceptor可以将所有的phase2b消息打包一次性發給leader；
2. acceptor可以将phase2b消息直接發給RM，RM替代leader的角色，這樣可以減少延遲，但是增加了總的消息數目；
3. 如果acceptor收到過abort的phase2a消息，可以忽略其他消息，直接通知leader或者RM進入aborted狀态；
每個RM的instance在ballot=0的消息中可能達不到choosen狀态，leader會因為逾時，主動使用更大的ballot執行phase1a的流程。如果leader執行到phase2a階段時，發現沒有任何value（可能RM網絡分區導緻發給acceptor的<phase2a, ballot=0>的消息沒有達到，或者其他什麼原因），則嘗試讓Aborted達成choosen，一旦決定這樣做，RM曾經發給acceptor的<phase2a, ballot=0>消息就會被拒絕掉。
leader的逾時處理必須是發送aborted消息，因為，在leader逾時時可能有RM發送給某個acceptor的abort消息剛剛達到，而沒有達成choosen狀态。此時leader恰好逾時，但是沒有收到這個acceptor的phase1b消息。
4.3 The Cost of Paxos Commit
下面計算Paxos Commit算法在事務最終達成commit的消息數目：
1. 事務發起者RM發送一個BeginCommit消息 (1個)；
2. leader向其他RM發送Prepare消息(N-1個)；
3. 每個RM廣播<phase2a, ballot=0>消息，fast paxos隻需要發送F+1個 N(F+1)個；
4. acceptor打包發送phase2b給acceptor (F+1個)；
5. leader廣播最終的committed或者aborted消息（N個）；
最終，***RM要經曆5個消息延遲得知事務是被committed還是aborted***。總的消息數目是(N+1)(F+3)-2個。如果第一個RM和leader在同一個node上可以節省一個消息，另外，第一個RM可以把BeginCommit消息和phase2a消息打包，又能節省一個消息，那麼一共是N+1)(F+3)-4個。如果acceptor個數比RM多，每個acceptor和RM部署在一起，那麼每個RM廣播phase2a消息時，發到本node上的消息就可以省略了，那麼又能節省出F+1個，最終剩下N(F+3)-3個。
如果phase2b消息直接發給RM，那麼可以節省一個消息延遲，隻要4個延遲就能完成事務送出。但是會多出N(F+1)個消息。
從磁盤寫入的次數上，Paxos Commit中一個acceptor可以把多個RM的多個instance一次性打包寫盤。
5 Paxos Commit和 2PC Commit的角色對比
在2PC中TM可以單獨決定是commit還是abort，在做出決定之前把狀态存入本地的磁盤中，當TM當機會阻塞整個2PC協定的運作。Paxos算法可以對一個值進行一緻性決議，本質上Paxos Commit是把2PC中TM的存儲替換成Paxos中的acceptor的存儲，把TM替換成leader。另外，Paxos Commit做了延遲優化。但是，Paxos Commit中的leader不能單方面的決定abort，leader必須要走一遍标準的Paxos協定的prepare+accept兩階段把abort值進行一遍決議。
6 Transaction Creation and Registration
前面讨論的是固定數目的RM參與單個事務的Paxos Commit算法。在真實系統中往往是先建立一個事務，然後RM加入。下面介紹如何在參與事務RM成員數目固定的算法基礎上，修改以支援可變數目RM的事務算法。
為了支援可變RM數目的事務，引入一個registrar角色。registrar地位和RM類似，做為一個額外的RM，記錄參與事務的RM集合。不同的是registrar在Paxos協定中的輸入是一個RM集合，而不是commit/abort。同樣的，給registrar也配置設定一個Paxos執行個體。
一個事務送出的條件是：
1. regstar標明一個RM集合，并且在leader上達成一緻；
2. RM集合中每個RM對應的Paxos執行個體都choosen出Prepared；
6.1 Transaction Creation
每個節點本地都有啟動一個transaction service。通過這個service，RM可以建立和管理事務。建立事務的時候，service構造出descriptor，包含：uid，registrar，initial leader，其他候選leader，acceptors。
該事務的整個執行期間的消息都會帶上descriptor，這樣接收者可以從descriptor找到對應的事務，以及後續要給哪些程序發送消息。
6.2 Join a Transaction
一個RM要加入事務，需要先向registrar發送join消息。join消息需要包含事務的descriptor，因為registrar可能是第一次收到這個事務的descriptor。建立事務的RM向其他參與者廣播descriptor。
registrar收到join消息後，把RM加入事務的RM集合，同時回複ack。RM收到ack後表明該RM就是事務的參與者之一了。
6.3 Committing a Transaction
事務的發起者RM向registrar發送BeginCommit，而不是向leader發送，因為registrar知道所有事務的參與者。緊接着registrar向所有其他事務的參與者RM發送Prepare消息，同時，拒絕後續參與這個事務的join消息，因為此時事務已經開始了。
registrar在收到BeginCommit後，要進行參與事務RM集合的決議。descriptor中包含了RM集合和acceptors，是以registrar隻需要向acceptor發送包含RM集合的phase2a消息，前面提到過registrar也對應一個Paxos執行個體。如果registrar的執行個體choosen失敗，leader也會發送Abort消息，幫助這個registrar的Paxos執行個體運作到終點。
6.4 Learning the Outcome
acceptor出現故障是很容易了解的，因為acceptor之間運作的是Paxos協定，在2F+1個acceptor的叢集中，可以容忍F個故障。隻要有多數派的acceptor正常工作，Paxos Commit協定就可以正常運作，這也是算法的設計目标。
如果一個節點P在隻有事務descriptor的情況下是如何學習到事務最終是committed還是aborted了呢？descriptor中包含了所有可能的Leader，任意程序P可以向所有可能的leader發送該事務的descriptor詢問事務的執行結果。如果所有的leader都故障了，P必須等待至少一個leader恢複。否則，正常運作的leader會傳回事務結果給P。
然而可能出現沒有一個leader知道事務的執行結果。需要運作标準Paxos協定的Learning過程。此時任意選擇一個leader L，L使用一個新的ballot執行registrar的Paxos執行個體，直到達到choosen狀态。如果達成的狀态是abortd則終止事務，否則L就得到了一個參與事務的RM集合。L開始進入Paxos Commit流程，使用新的ballot給每個參與事務的RM執行一個Paxos執行個體，如果最終所有的RM都達到了Prepared狀态，則L就學習到了’事務可以送出‘的這一結果。L就可以把這個事務的結果通知給其他程序。
這個過程也可能出現失敗，比如由于網絡分區，選取出了兩個不同的leader，這兩個leader同時發送給acceptor，出現踩踏問題，但是Paxos算法可以保證最終隻能一個一緻的結果。隻要隻有一個Leader在Learning，最終就能保證一定學習到事務執行的結果。
2PC的Learning過程是非常簡單的，TM同時扮演者acceptor，leader，registrar角色。任意程序P想要知道事務的執行結果，隻要給TM發送消息即可。如果TM故障了，隻能等待别無他法。
如果RM或者事務的發起者Leader失敗了，事務送出協定就暫時hang住了。隻要有程序使用上述過程嘗試學習該事務的執行結果，都會推動事務的繼續前進。比如，RM在發送phase2a消息後，逾時了沒有收到後續的消息。RM嘗試向Leader學習事務的執行結果，Leader使用新的ballot運作标準paxos協定發送phase2a消息，最終Leader學習到所有RM的狀态，然後回複給這個RM。
7 總結
2PC是經典的事務送出協定，也被稱為是同步的協定，因為2PC是不能容錯的，一旦TM出現故障其他RM必須同步的等TM恢複過來。
Paxos Commit引入了多個TM，隻要多數派正常運作，就不會阻塞事務送出協定的運作。
在正常failure-free的場景下，Paxos Commit隻比2PC多一個消息時延，在經過Fast Paxos優化後，這個多出來的消息時延也可以被消除掉。理論上達到了非阻塞協定的最小時延。
8 TLA+代碼
作者附帶了TLA+的形式化描述，并沒有包含failover相關的邏輯。通過tla+也能進一步精确的了解算法。
----------------------------- MODULE PaxosCommit ----------------------------
EXTENDS Integers
Maximum(S) ==
LET Max[T \in SUBSET S] ==
IF T = {} THEN -1
ELSE LET n == CHOOSE n \in T : TRUE
rmax == Max[T \ {n}]
IN IF n \geq rmax THEN n ELSE rmax
IN Max[S]
CONSTANT RM, \* The set of resource managers.
Acceptor, \* The set of acceptors.
Majority, \* The set of majorities of acceptors
Ballot \* The set of ballot numbers
ASSUME \* We assume these properties of the declared constants.
/\ Ballot \subseteq Nat
/\ 0 \in Ballot
/\ Majority \subseteq SUBSET Acceptor
/\ \A MS1, MS2 \in Majority : MS1 \cap MS2 # {}
Message ==
[type : {"phase1a"}, ins : RM, bal : Ballot \ {0}]
\cup
[type : {"phase1b"}, ins : RM, mbal : Ballot, bal : Ballot \cup {-1},
val : {"prepared", "aborted", "none"}, acc : Acceptor]
\cup
[type : {"phase2a"}, ins : RM, bal : Ballot, val : {"prepared", "aborted"}]
\cup
[type : {"phase2b"}, acc : Acceptor, ins : RM, bal : Ballot,
val : {"prepared", "aborted"}]
\cup
[type : {"Commit", "Abort"}]
-----------------------------------------------------------------------------
VARIABLES
rmState,
aState,
msgs
PCTypeOK ==
/\ rmState \in [RM -> {"working", "prepared", "committed", "aborted"}]
/\ aState \in [RM -> [Acceptor -> [mbal : Ballot,
bal : Ballot \cup {-1},
val : {"prepared", "aborted", "none"}]]]
/\ msgs \in SUBSET Message
PCInit ==
/\ rmState = [rm \in RM |-> "working"]
/\ aState = [ins \in RM |->
[ac \in Acceptor
|-> [mbal |-> 0, bal |-> -1, val |-> "none"]]]
/\ msgs = {}
Send(m) == msgs' = msgs \cup {m}
RMPrepare(rm) ==
/\ rmState[rm] = "working"
/\ rmState' = [rmState EXCEPT ![rm] = "prepared"]
/\ Send([type |-> "phase2a", ins |-> rm, bal |-> 0, val |-> "prepared"])
/\ UNCHANGED aState
RMChooseToAbort(rm) ==
/\ rmState[rm] = "working"
/\ rmState' = [rmState EXCEPT ![rm] = "aborted"]
/\ Send([type |-> "phase2a", ins |-> rm, bal |-> 0, val |-> "aborted"])
/\ UNCHANGED aState
RMRcvCommitMsg(rm) ==
/\ [type |-> "Commit"] \in msgs
/\ rmState' = [rmState EXCEPT ![rm] = "committed"]
/\ UNCHANGED <<aState, msgs>>
RMRcvAbortMsg(rm) ==
/\ [type |-> "Abort"] \in msgs
/\ rmState' = [rmState EXCEPT ![rm] = "aborted"]
/\ UNCHANGED <<aState, msgs>>
-----------------------------------------------------------------------------
Phase1a(bal, rm) ==
/\ Send([type |-> "phase1a", ins |-> rm, bal |-> bal])
/\ UNCHANGED <<rmState, aState>>
Phase2a(bal, rm) ==
/\ ~\E m \in msgs : /\ m.type = "phase2a"
/\ m.bal = bal
/\ m.ins = rm
/\ \E MS \in Majority :
LET mset == {m \in msgs : /\ m.type = "phase1b"
/\ m.ins = rm
/\ m.mbal = bal
/\ m.acc \in MS}
maxbal == Maximum({m.bal : m \in mset})
val == IF maxbal = -1
THEN "aborted"
ELSE (CHOOSE m \in mset : m.bal = maxbal).val
IN /\ \A ac \in MS : \E m \in mset : m.acc = ac
/\ Send([type |-> "phase2a", ins |-> rm, bal |-> bal, val |-> val])
/\ UNCHANGED <<rmState, aState>>
Decide ==
/\ LET Decided(rm, v) ==
\E b \in Ballot, MS \in Majority :
\A ac \in MS : [type |-> "phase2b", ins |-> rm,
bal |-> b, val |-> v, acc |-> ac ] \in msgs
IN \/ /\ \A rm \in RM : Decided(rm, "prepared")
/\ Send([type |-> "Commit"])
\/ /\ \E rm \in RM : Decided(rm, "aborted")
/\ Send([type |-> "Abort"])
/\ UNCHANGED <<rmState, aState>>
-----------------------------------------------------------------------------
Phase1b(acc) ==
\E m \in msgs :
/\ m.type = "phase1a"
/\ aState[m.ins][acc].mbal < m.bal
/\ aState' = [aState EXCEPT ![m.ins][acc].mbal = m.bal]
/\ Send([type |-> "phase1b",
ins |-> m.ins,
mbal |-> m.bal,
bal |-> aState[m.ins][acc].bal,
val |-> aState[m.ins][acc].val,
acc |-> acc])
/\ UNCHANGED rmState
Phase2b(acc) ==
/\ \E m \in msgs :
/\ m.type = "phase2a"
/\ aState[m.ins][acc].mbal \leq m.bal
/\ aState' = [aState EXCEPT ![m.ins][acc].mbal = m.bal,
![m.ins][acc].bal = m.bal,
![m.ins][acc].val = m.val]
/\ Send([type |-> "phase2b", ins |-> m.ins, bal |-> m.bal,
val |-> m.val, acc |-> acc])
/\ UNCHANGED rmState
-----------------------------------------------------------------------------
PCNext == \* The next-state action
\/ \E rm \in RM : \/ RMPrepare(rm)
\/ RMChooseToAbort(rm)
\/ RMRcvCommitMsg(rm)
\/ RMRcvAbortMsg(rm)
\/ \E bal \in Ballot \ {0}, rm \in RM : Phase1a(bal, rm) \/ Phase2a(bal, rm)
\/ Decide
\/ \E acc \in Acceptor : Phase1b(acc) \/ Phase2b(acc)
-----------------------------------------------------------------------------
PCSpec == PCInit /\ [][PCNext]_<<rmState, aState, msgs>>
THEOREM PCSpec => PCTypeOK
-----------------------------------------------------------------------------
TC == INSTANCE TCommit
THEOREM PCSpec => TC!TCSpec
=============================================================================