目錄一致性算法背景Paxos實(shí)現(xiàn)集群庫介紹業(yè)務(wù)場景和痛點(diǎn)分布式狀態(tài)機(jī)核心需求類型和常量組件模型應(yīng)用接口Role 類AcceptorReplicaLeader Scout CommanderBootstrap一致性算法背景

1.Paxos一致性算法解決的問題：分布式系統(tǒng)中數(shù)據(jù)不能存在單個(gè)節(jié)點(diǎn)（主機(jī)）上，否則可能出現(xiàn)單點(diǎn)故障；多個(gè)節(jié)點(diǎn)（主機(jī)）需要保證具有相同的數(shù)據(jù)。

2.什么是一致性：一致性就是數(shù)據(jù)保持一致，在分布式系統(tǒng)中，可以理解為多個(gè)節(jié)點(diǎn)中數(shù)據(jù)的值是一致的。

3.一致性模型分類：一般分為強(qiáng)一致性和弱一致性，強(qiáng)一致性保證系統(tǒng)改變提交以后立即改變集群的狀態(tài)。常見模型包括：Paxos，Raft（muti-paxos），ZAB（muti-paxos）； 弱一致性也叫最終一致性，系統(tǒng)不保證改變提交以后立即改變集群的狀態(tài)，但是隨著時(shí)間的推移最終狀態(tài)一致的。常見模型包括：DNS系統(tǒng)，Gossip協(xié)議

4.一致性算法使用案例：Google的Chubby分布式鎖服務(wù)，采用了Paxos算法；etcd分布式鍵值數(shù)據(jù)庫，采用了Raft算法；ZooKeeper分布式應(yīng)用協(xié)調(diào)服務(wù)以及Chubby的開源實(shí)現(xiàn)，采用ZAB算法

simple-paxos就單個(gè)靜態(tài)值達(dá)一致性本身并不實(shí)用，我們需要實(shí)現(xiàn)的集群系統(tǒng)（銀行賬戶服務(wù)）希望就隨時(shí)間變化的特定狀態(tài)（賬戶余額）達(dá)成一致。所以需要使用Paxos就每個(gè)操作達(dá)成一致，將每個(gè)修改視為狀態(tài)機(jī)轉(zhuǎn)換。

Multi-Paxos實(shí)際上是simple Paxos實(shí)例（插槽）的序列，每個(gè)實(shí)例都按順序編號(hào)。每個(gè)狀態(tài)轉(zhuǎn)換都被賦予一個(gè)“插槽編號(hào)”，集群的每個(gè)成員都以嚴(yán)格的數(shù)字順序執(zhí)行轉(zhuǎn)換。為了更改群集的狀態(tài)（例如，處理傳輸操作），我們嘗試在下一個(gè)插槽中就該操作達(dá)成一致性。具體來說，這意味著向每個(gè)消息添加一個(gè)插槽編號(hào)，并在每個(gè)插槽的基礎(chǔ)上跟蹤所有協(xié)議狀態(tài)。

為每個(gè)插槽運(yùn)行Paxos，至少兩次往返會(huì)太慢。Multi-Paxos通過對(duì)所有插槽使用相同的選票編號(hào)集進(jìn)行優(yōu)化，并同時(shí)對(duì)所有插槽執(zhí)行Prepare/Promise。

Client   Proposer      Acceptor     Learner

   |         |          |  |  |       |  | --- First Request ---

   X-------->|          |  |  |       |  |  Request

   |         X--------->|->|->|       |  |  Prepare(N)

   |         |<---------X--X--X       |  |  Promise(N,I,{Va,Vb,Vc})

   |         X--------->|->|->|       |  |  Accept!(N,I,V)

   |         |<---------X--X--X------>|->|  Accepted(N,I,V)

   |<---------------------------------X--X  Response

   |         |          |  |  |       |  |

Paxos實(shí)現(xiàn)

在實(shí)用軟件中實(shí)現(xiàn)Multi-Paxos是出了名的困難，催生了許多論文如'Paxos Made Simple'，“Paxos Made Practical”

首先，multi-poposer在繁忙的環(huán)境中可能會(huì)成為問題，因?yàn)槊總€(gè)群集成員都試圖在每個(gè)插槽中決定其狀態(tài)機(jī)操作。解決方法是選舉一名“l(fā)eader”，負(fù)責(zé)為每個(gè)時(shí)段提交選票。所有其他群集節(jié)點(diǎn)將新操作發(fā)送到領(lǐng)導(dǎo)者執(zhí)行。因此，在只有一名領(lǐng)導(dǎo)人的正常運(yùn)作中，不會(huì)發(fā)生投票沖突。

Prepare/Promise階段可以作為一種leader選舉：無論哪個(gè)集群成員擁有最近承諾的選票號(hào)碼，都被視為leader。leader后續(xù)可以自由地直接執(zhí)行Accept/Accepted階段，而不重復(fù)第一階段。我們將在下文看到的，leader選舉實(shí)際上是相當(dāng)復(fù)雜的。

雖然simple Paxos保證集群不會(huì)達(dá)成沖突的決定，但它不能保證會(huì)做出任何決定。例如，如果初始的Prepare消息丟失，并且沒有到達(dá)接受者，則提議者將等待永遠(yuǎn)不會(huì)到達(dá)的Promise消息。解決這個(gè)問題需要精心設(shè)計(jì)的重新傳輸：足以最終取得進(jìn)展，但不會(huì)群集產(chǎn)生數(shù)據(jù)包風(fēng)暴。

另一個(gè)問題是決定的傳播。在正常情況下，簡單地廣播Decision信息就可以解決這個(gè)問題。但是，如果消息丟失，節(jié)點(diǎn)可能會(huì)永遠(yuǎn)不知道該決定，并且無法為以后的插槽應(yīng)用狀態(tài)機(jī)轉(zhuǎn)換。所以實(shí)現(xiàn)需要一些機(jī)制來共享有關(guān)已決定提案的信息。

使用分布式狀態(tài)機(jī)帶來了另一個(gè)挑戰(zhàn)：當(dāng)新節(jié)點(diǎn)啟動(dòng)時(shí)，它需要獲取群集的現(xiàn)有狀態(tài)。雖然可以通過趕上第一個(gè)插槽以來的所有插槽的決策來做到這一點(diǎn)，但在一個(gè)大的集群中，這可能涉及數(shù)百萬個(gè)插槽。此外，我們需要一些方法來初始化一個(gè)新的群集。

集群庫介紹

前面都是理論介紹，下面我們使用python來實(shí)現(xiàn)一個(gè)簡化的Multi-Paxos

業(yè)務(wù)場景和痛點(diǎn)

我們以簡單的銀行賬戶管理服務(wù)的場景作為案例。在這個(gè)服務(wù)中，每一個(gè)賬戶都有一個(gè)當(dāng)前余額，同時(shí)每個(gè)賬戶都有自己的賬號(hào)。用戶可以對(duì)賬戶進(jìn)行“存款”、“轉(zhuǎn)賬”、“查詢當(dāng)前余額”等操作。“轉(zhuǎn)賬”操作同時(shí)涉及了兩個(gè)賬戶：轉(zhuǎn)出賬戶和轉(zhuǎn)入賬戶，如果賬戶余額不足，轉(zhuǎn)賬操作必須被駁回。

如果這個(gè)服務(wù)僅僅在一個(gè)服務(wù)器上部署，很容易就能夠?qū)崿F(xiàn)：使用一個(gè)操作鎖來確保“轉(zhuǎn)賬”操作不會(huì)同時(shí)進(jìn)行，同時(shí)對(duì)轉(zhuǎn)出賬戶的進(jìn)行校驗(yàn)。然而，銀行不可能僅僅依賴于一個(gè)服務(wù)器來儲(chǔ)存賬戶余額這樣的關(guān)鍵信息，通常，這些服務(wù)都是被分布在多個(gè)服務(wù)器上的，每一個(gè)服務(wù)器各自運(yùn)行著相同代碼的實(shí)例。用戶可以通過任何一個(gè)服務(wù)器來操作賬戶。

在一個(gè)簡單的分布式處理系統(tǒng)的實(shí)現(xiàn)中，每個(gè)服務(wù)器都會(huì)保存一份賬戶余額的副本。它會(huì)處理任何收到的操作，并且將賬戶余額的更新發(fā)送給其他的服務(wù)器。但是這種方法有一個(gè)嚴(yán)重的問題：如果兩個(gè)服務(wù)器同時(shí)對(duì)一個(gè)賬戶進(jìn)行操作，哪一個(gè)新的賬戶余額是正確的？即使服務(wù)器不共享余額而是共享操作，對(duì)一個(gè)賬戶同時(shí)進(jìn)行轉(zhuǎn)賬操作也可能造成透支。

從根本上來說，這些錯(cuò)誤的發(fā)生都是由于服務(wù)器使用它們本地狀態(tài)來響應(yīng)操作，而不是首先確保本地狀態(tài)與其他服務(wù)器相匹配。比如，想象服務(wù)器A接到了從賬號(hào)101向賬號(hào)202轉(zhuǎn)賬的操作指令，而此時(shí)服務(wù)器B已經(jīng)處理了另一個(gè)把賬號(hào)101的錢都轉(zhuǎn)到賬號(hào)202的請(qǐng)求，卻沒有通知服務(wù)器A。這樣，服務(wù)器A的本地狀態(tài)與服務(wù)器B不一樣，即使會(huì)造成賬戶101透支，服務(wù)器A依然允許從賬號(hào)101進(jìn)行轉(zhuǎn)賬操作。

分布式狀態(tài)機(jī)

為了防止上述情況發(fā)生我們采用了一種叫做“分布式狀態(tài)機(jī)”的工具。它的思路是對(duì)每個(gè)同樣的輸入，每個(gè)服務(wù)器都運(yùn)行同樣的對(duì)應(yīng)的狀態(tài)機(jī)。由于狀態(tài)機(jī)的特性，對(duì)于同樣的輸入每個(gè)服務(wù)器的輸出都是一樣的。對(duì)于像“轉(zhuǎn)賬”、“查詢當(dāng)前余額”等操作，賬號(hào)和余額也都是狀態(tài)機(jī)的輸入。

這個(gè)應(yīng)用的狀態(tài)機(jī)比較簡單：

def execute_operation(state, operation): if operation.name == ’deposit’: if not verify_signature(operation.deposit_signature): return state, False state.accounts[operation.destination_account] += operation.amount return state, True elif operation.name == ’transfer’: if state.accounts[operation.source_account] < operation.amount: return state, False state.accounts[operation.source_account] -= operation.amount state.accounts[operation.destination_account] += operation.amount return state, True elif operation.name == ’get-balance’: return state, state.accounts[operation.account]

值得注意的是，運(yùn)行“查詢當(dāng)前余額”操作時(shí)雖然并不會(huì)改變當(dāng)前狀態(tài)，但是我們依然把它當(dāng)做一個(gè)狀態(tài)變化操作來實(shí)現(xiàn)。這確保了返回的余額是分布式系統(tǒng)中的最新信息，并且不是基于一個(gè)服務(wù)器上的本地狀態(tài)來進(jìn)行返回的。

這可能跟你在計(jì)算機(jī)課程中學(xué)習(xí)到的典型的狀態(tài)機(jī)不太一樣。傳統(tǒng)的狀態(tài)機(jī)是一系列有限個(gè)狀態(tài)的集合，每個(gè)狀態(tài)都與一個(gè)標(biāo)記的轉(zhuǎn)移行為相對(duì)應(yīng)，而在本文中，狀態(tài)機(jī)的狀態(tài)是賬戶余額的集合，因此存在無窮多個(gè)可能的狀態(tài)。但是，狀態(tài)機(jī)的基本規(guī)則同樣適用于本文的狀態(tài)機(jī)：對(duì)于同樣的初始狀態(tài)，同樣的輸入總是有同樣的輸出。

因此，分布式狀態(tài)機(jī)確保了對(duì)于同樣的操作，每個(gè)主機(jī)都會(huì)有同樣的相應(yīng)。但是，為了確保每個(gè)服務(wù)器都允許狀態(tài)機(jī)的輸入，前文中提到的問題依然存在。這是一個(gè)一致性問題，為了解決它我們采用了一種派生的Paxos算法。

核心需求

可以為較大的應(yīng)用程序提供一致性服務(wù): 我們用一個(gè)Cluster庫來實(shí)現(xiàn)簡化的Multi-Paxos正確性是這個(gè)庫最重要的能力，因此結(jié)構(gòu)化代碼是很重要的，以便我們可以看到并測試它與規(guī)范的對(duì)應(yīng)關(guān)系。復(fù)雜的協(xié)議可能會(huì)出現(xiàn)復(fù)雜的故障，因此我們將構(gòu)建對(duì)復(fù)現(xiàn)和調(diào)試不常見的故障的支持。我們會(huì)實(shí)現(xiàn)POC代碼：足以證明核心概念是實(shí)用的，代碼的結(jié)構(gòu)化是為了后續(xù)添加此功能對(duì)核心實(shí)現(xiàn)的更改最小我們開始coding吧。

類型和常量

cluster中的協(xié)議需要使用15不同的消息類型，每種消息類型使用collection中的namedturple定義：

Accepted = namedtuple(’Accepted’, [’slot’, ’ballot_num’]) Accept = namedtuple(’Accept’, [’slot’, ’ballot_num’, ’proposal’]) Decision = namedtuple(’Decision’, [’slot’, ’proposal’]) Invoked = namedtuple(’Invoked’, [’client_id’, ’output’]) Invoke = namedtuple(’Invoke’, [’caller’, ’client_id’, ’input_value’]) Join = namedtuple(’Join’, []) Active = namedtuple(’Active’, []) Prepare = namedtuple(’Prepare’, [’ballot_num’]) Promise = namedtuple(’Promise’, [’ballot_num’, ’accepted_proposals’]) Propose = namedtuple(’Propose’, [’slot’, ’proposal’]) Welcome = namedtuple(’Welcome’, [’state’, ’slot’, ’decisions’]) Decided = namedtuple(’Decided’, [’slot’]) Preempted = namedtuple(’Preempted’, [’slot’, ’preempted_by’]) Adopted = namedtuple(’Adopted’, [’ballot_num’, ’accepted_proposals’]) Accepting = namedtuple(’Accepting’, [’leader’])

使用命名元組描述每種消息類型可以保持代碼的clean，并有助于避免一些簡單的錯(cuò)誤。如果命名元組構(gòu)造函數(shù)沒有被賦予正確的屬性，則它將引發(fā)異常，從而使錯(cuò)誤變得明顯。元組在日志消息中k可以很好地格式化，不會(huì)像字典那樣使用那么多的內(nèi)存。

創(chuàng)建消息：

msg = Accepted(slot=10, ballot_num=30)

訪問消息：

got_ballot_num = msg.ballot_num

后面我們會(huì)了解這些消息的含義。代碼還引入了一些常量，其中大多數(shù)常量定義了各種消息的超時(shí)：

JOIN_RETRANSMIT = 0.7 CATCHUP_INTERVAL = 0.6 ACCEPT_RETRANSMIT = 1.0 PREPARE_RETRANSMIT = 1.0 INVOKE_RETRANSMIT = 0.5 LEADER_TIMEOUT = 1.0 NULL_BALLOT = Ballot(-1, -1) # sorts before all real ballots NOOP_PROPOSAL = Proposal(None, None, None) # no-op to fill otherwise empty slots

最后我們需要定義協(xié)議中的Proposal和Ballot

Proposal = namedtuple(’Proposal’, [’caller’, ’client_id’, ’input’]) Ballot = namedtuple(’Ballot’, [’n’, ’leader’])組件模型

實(shí)現(xiàn)multi-paxos的核心組件包括Role和Node。

為了保證可測試性并保持代碼的可讀性，我們將Cluster分解為與協(xié)議中描述的角色相對(duì)應(yīng)的幾個(gè)類。每個(gè)都是Role的子類。

class Role(object): def __init__(self, node):self.node = nodeself.node.register(self)self.running = Trueself.logger = node.logger.getChild(type(self).__name__) def set_timer(self, seconds, callback):return self.node.network.set_timer(self.node.address, seconds, lambda: self.running and callback()) def stop(self):self.running = Falseself.node.unregister(self)

群集節(jié)點(diǎn)的角色由Node類粘在一起，該類代表網(wǎng)絡(luò)上的單個(gè)節(jié)點(diǎn)。在程序過程中角色將添加到節(jié)點(diǎn)中，并從節(jié)點(diǎn)中刪除。

到達(dá)節(jié)點(diǎn)的消息將中繼到所有活動(dòng)角色，調(diào)用以消息類型命名的方法，前綴為do_。 這些do_方法接收消息的屬性作為關(guān)鍵字參數(shù)，以便于訪問。Node``類還提供了``send方法作為方便，使用functools.partial為Network類的相同方法提供一些參數(shù)。

class Node(object): unique_ids = itertools.count() def __init__(self, network, address):self.network = networkself.address = address or ’N%d’ % self.unique_ids.next()self.logger = SimTimeLogger( logging.getLogger(self.address), {’network’: self.network})self.logger.info(’starting’)self.roles = []self.send = functools.partial(self.network.send, self) def register(self, roles):self.roles.append(roles) def unregister(self, roles):self.roles.remove(roles) def receive(self, sender, message):handler_name = ’do_%s’ % type(message).__name__for comp in self.roles[:]: if not hasattr(comp, handler_name):continue comp.logger.debug('received %s from %s', message, sender) fn = getattr(comp, handler_name) fn(sender=sender, **message._asdict())應(yīng)用接口

每個(gè)集群成員上都會(huì)創(chuàng)建并啟動(dòng)一個(gè)Member對(duì)象，提供特定于應(yīng)用程序的狀態(tài)機(jī)和對(duì)等項(xiàng)列表。如果成員對(duì)象正在加入現(xiàn)有集群，則該成員對(duì)象向該節(jié)點(diǎn)添加bootstrap角色，如果正在創(chuàng)建新集群，則該成員對(duì)象添加seed。再用Network.run在單獨(dú)的線程中運(yùn)行協(xié)議。

應(yīng)用程序通過該invoke方法與集群進(jìn)行交互，從而啟動(dòng)了狀態(tài)轉(zhuǎn)換， 確定該提議并運(yùn)行狀態(tài)機(jī)后，invoke將返回狀態(tài)機(jī)的輸出。該方法使用簡單的同步Queue來等待協(xié)議線程的結(jié)果。

class Member(object): def __init__(self, state_machine, network, peers, seed=None, seed_cls=Seed, bootstrap_cls=Bootstrap):self.network = networkself.node = network.new_node()if seed is not None: self.startup_role = seed_cls(self.node, initial_state=seed, peers=peers, execute_fn=state_machine)else: self.startup_role = bootstrap_cls(self.node, execute_fn=state_machine, peers=peers)self.requester = None def start(self):self.startup_role.start()self.thread = threading.Thread(target=self.network.run)self.thread.start() def invoke(self, input_value, request_cls=Requester):assert self.requester is Noneq = Queue.Queue()self.requester = request_cls(self.node, input_value, q.put)self.requester.start()output = q.get()self.requester = Nonereturn outputRole 類

Paxos協(xié)議中的角色包括：client, acceptor, proposer, learner, and leader。在典型的實(shí)現(xiàn)中，單個(gè)processor可以同時(shí)扮演一個(gè)或多個(gè)角色。這不會(huì)影響協(xié)議的正確性，通常會(huì)合并角色以改善協(xié)議中的延遲和/或消息數(shù)量。

下面逐一實(shí)現(xiàn)每個(gè)角色類

Acceptor

Acceptor 類實(shí)現(xiàn)的是Paxos中的 acceptor角色，所以必須存儲(chǔ)最近promise的選票編號(hào)，以及每個(gè)時(shí)段接受的各個(gè)slot的proposal，同時(shí)需要相應(yīng)Prepare和Accept消息。 這里的POC實(shí)現(xiàn)是一個(gè)和協(xié)議可以直接對(duì)應(yīng)的短類，對(duì)于acceptor來說Multi-paxos看起來像是簡單的Paxos，只是在message中添加了slot number。

class Acceptor(Role): def __init__(self, node):super(Acceptor, self).__init__(node)self.ballot_num = NULL_BALLOTself.accepted_proposals = {} # {slot: (ballot_num, proposal)} def do_Prepare(self, sender, ballot_num):if ballot_num > self.ballot_num: self.ballot_num = ballot_num # we’ve heard from a scout, so it might be the next leader self.node.send([self.node.address], Accepting(leader=sender))self.node.send([sender], Promise( ballot_num=self.ballot_num, accepted_proposals=self.accepted_proposals)) def do_Accept(self, sender, ballot_num, slot, proposal):if ballot_num >= self.ballot_num: self.ballot_num = ballot_num acc = self.accepted_proposals if slot not in acc or acc[slot][0] < ballot_num:acc[slot] = (ballot_num, proposal)self.node.send([sender], Accepted( slot=slot, ballot_num=self.ballot_num))Replica

Replica類是Role類最復(fù)雜的子類，對(duì)應(yīng)協(xié)議中的Learner和Proposal角色，它的主要職責(zé)是：提出新的proposal；在決定proposal時(shí)調(diào)用本地狀態(tài)機(jī)；跟蹤當(dāng)前Leader；以及將新啟動(dòng)的節(jié)點(diǎn)添加到集群中。

Replica創(chuàng)建新的proposal以響應(yīng)來自客戶端的“invoke”消息，選擇它認(rèn)為是未使用的插槽，并向當(dāng)前l(fā)eader發(fā)送“Propose”消息。如果選定插槽的共識(shí)是針對(duì)不同proposal，則replica必須使用新插槽re-propose。

下圖顯示Replica的角色控制流程：

Requester    Local Rep   Current Leader

   X---------->|             |    Invoke

   |           X------------>|    Propose

   |           |<------------X    Decision

   |<----------X             |    Decision

   |           |             |

Decision消息表示集群已達(dá)成共識(shí)的插槽， Replica類存儲(chǔ)新的決定并運(yùn)行狀態(tài)機(jī)，直到到達(dá)未確定的插槽。Replica從本地狀態(tài)機(jī)已處理的提交的slot識(shí)別出集群已同意的已決定的slot。如果slot出現(xiàn)亂序，提交的提案可能會(huì)滯后，等待下一個(gè)空位被決定。提交slot后，每個(gè)replica會(huì)將操作結(jié)果發(fā)送回一條Invoked消息給請(qǐng)求者。

在某些情況下slot可能沒有有效的提案，也沒有決策,需要狀態(tài)機(jī)一個(gè)接一個(gè)地執(zhí)行slot，因此群集必須就填充slot的內(nèi)容達(dá)成共識(shí)。為了避免這種可能性，Replica在遇到插槽時(shí)會(huì)提出“no-op”的proposal。如果最終決定了這樣的proposal，則狀態(tài)機(jī)對(duì)該slot不執(zhí)行任何操作。

同樣，同一proposal有可能被Decision兩次。對(duì)于任何此類重復(fù)的proposal，Replica將跳過調(diào)用狀態(tài)機(jī)，而不會(huì)對(duì)該slot執(zhí)行任何狀態(tài)轉(zhuǎn)換。

Replicas需要知道哪個(gè)節(jié)點(diǎn)是active leader才能向其發(fā)送Propose消息， 要實(shí)現(xiàn)這一目標(biāo)，每個(gè)副本都使用三個(gè)信息源跟蹤active leader。

當(dāng)leader 的角色轉(zhuǎn)換為active時(shí)，它會(huì)向同一節(jié)點(diǎn)上的副本發(fā)送一條Adopted消息（下圖）：

Leader    Local Repplica   

   X----------->|          Admopted

當(dāng)acceptor角色向Promise新的leader發(fā)送Accepting消息時(shí)，它將消息發(fā)送到其本地副本（下圖）。

Acceptor    Local Repplica   

   X----------->|          Accepting

active leader將以心跳的形式發(fā)送Active消息。如果在LEADER_TIMEOUT到期之前沒有此類消息到達(dá)，則Replica將假定該Leader已死，并轉(zhuǎn)向下一個(gè)Leader。在這種情況下，重要的是所有副本都選擇相同的新領(lǐng)導(dǎo)者，我們可以通過對(duì)成員進(jìn)行排序并在列表中選擇下一個(gè)leader。

當(dāng)節(jié)點(diǎn)加入網(wǎng)絡(luò)時(shí)，Bootstrap將發(fā)送一條Join消息（下圖）。Replica以一條Welcome包含其最新狀態(tài)的消息作為響應(yīng)，從而使新節(jié)點(diǎn)能夠快速啟用。

BootStrap ReplicaReplica Replica X---------->| | | Join |<----------X X | Welcome X------------------------>| | Join |<------------------------X | Welcome X-------------------------------------->| Join |<--------------------------------------X Welcome class Replica(Role): def __init__(self, node, execute_fn, state, slot, decisions, peers):super(Replica, self).__init__(node)self.execute_fn = execute_fnself.state = stateself.slot = slotself.decisions = decisionsself.peers = peersself.proposals = {}# next slot num for a proposal (may lead slot)self.next_slot = slotself.latest_leader = Noneself.latest_leader_timeout = None # making proposals def do_Invoke(self, sender, caller, client_id, input_value):proposal = Proposal(caller, client_id, input_value)slot = next((s for s, p in self.proposals.iteritems() if p == proposal), None)# propose, or re-propose if this proposal already has a slotself.propose(proposal, slot) def propose(self, proposal, slot=None):'''Send (or resend, if slot is specified) a proposal to the leader'''if not slot: slot, self.next_slot = self.next_slot, self.next_slot + 1self.proposals[slot] = proposal# find a leader we think is working - either the latest we know of, or# ourselves (which may trigger a scout to make us the leader)leader = self.latest_leader or self.node.addressself.logger.info( 'proposing %s at slot %d to leader %s' % (proposal, slot, leader))self.node.send([leader], Propose(slot=slot, proposal=proposal)) # handling decided proposals def do_Decision(self, sender, slot, proposal):assert not self.decisions.get(self.slot, None), 'next slot to commit is already decided'if slot in self.decisions: assert self.decisions[slot] == proposal, 'slot %d already decided with %r!' % (slot, self.decisions[slot]) returnself.decisions[slot] = proposalself.next_slot = max(self.next_slot, slot + 1)# re-propose our proposal in a new slot if it lost its slot and wasn’t a no-opour_proposal = self.proposals.get(slot)if (our_proposal is not None and our_proposal != proposal and our_proposal.caller): self.propose(our_proposal)# execute any pending, decided proposalswhile True: commit_proposal = self.decisions.get(self.slot) if not commit_proposal:break # not decided yet commit_slot, self.slot = self.slot, self.slot + 1 self.commit(commit_slot, commit_proposal) def commit(self, slot, proposal):'''Actually commit a proposal that is decided and in sequence'''decided_proposals = [p for s, p in self.decisions.iteritems() if s < slot]if proposal in decided_proposals: self.logger.info('not committing duplicate proposal %r, slot %d', proposal, slot) return # duplicateself.logger.info('committing %r at slot %d' % (proposal, slot))if proposal.caller is not None: # perform a client operation self.state, output = self.execute_fn(self.state, proposal.input) self.node.send([proposal.caller], Invoked(client_id=proposal.client_id, output=output)) # tracking the leader def do_Adopted(self, sender, ballot_num, accepted_proposals):self.latest_leader = self.node.addressself.leader_alive() def do_Accepting(self, sender, leader):self.latest_leader = leaderself.leader_alive() def do_Active(self, sender):if sender != self.latest_leader: returnself.leader_alive() def leader_alive(self):if self.latest_leader_timeout: self.latest_leader_timeout.cancel()def reset_leader(): idx = self.peers.index(self.latest_leader) self.latest_leader = self.peers[(idx + 1) % len(self.peers)] self.logger.debug('leader timed out; tring the next one, %s', self.latest_leader)self.latest_leader_timeout = self.set_timer(LEADER_TIMEOUT, reset_leader) # adding new cluster members def do_Join(self, sender):if sender in self.peers: self.node.send([sender], Welcome(state=self.state, slot=self.slot, decisions=self.decisions))Leader Scout Commander

Leader的主要任務(wù)是接受Propose要求新投票的消息并做出決定。成功完成協(xié)議的Prepare/Promise部分后Leader將處于“Active狀態(tài)” 。活躍的Leader可以立即發(fā)送Accept消息以響應(yīng)Propose。

與按角色分類的模型保持一致，Leader會(huì)委派scout和Commander角色來執(zhí)行協(xié)議的每個(gè)部分。

class Leader(Role): def __init__(self, node, peers, commander_cls=Commander, scout_cls=Scout):super(Leader, self).__init__(node)self.ballot_num = Ballot(0, node.address)self.active = Falseself.proposals = {}self.commander_cls = commander_clsself.scout_cls = scout_clsself.scouting = Falseself.peers = peers def start(self):# reminder others we’re active before LEADER_TIMEOUT expiresdef active(): if self.active:self.node.send(self.peers, Active()) self.set_timer(LEADER_TIMEOUT / 2.0, active)active() def spawn_scout(self):assert not self.scoutingself.scouting = Trueself.scout_cls(self.node, self.ballot_num, self.peers).start() def do_Adopted(self, sender, ballot_num, accepted_proposals):self.scouting = Falseself.proposals.update(accepted_proposals)# note that we don’t re-spawn commanders here; if there are undecided# proposals, the replicas will re-proposeself.logger.info('leader becoming active')self.active = True def spawn_commander(self, ballot_num, slot):proposal = self.proposals[slot]self.commander_cls(self.node, ballot_num, slot, proposal, self.peers).start() def do_Preempted(self, sender, slot, preempted_by):if not slot: # from the scout self.scouting = Falseself.logger.info('leader preempted by %s', preempted_by.leader)self.active = Falseself.ballot_num = Ballot((preempted_by or self.ballot_num).n + 1, self.ballot_num.leader) def do_Propose(self, sender, slot, proposal):if slot not in self.proposals: if self.active:self.proposals[slot] = proposalself.logger.info('spawning commander for slot %d' % (slot,))self.spawn_commander(self.ballot_num, slot) else:if not self.scouting: self.logger.info('got PROPOSE when not active - scouting') self.spawn_scout()else: self.logger.info('got PROPOSE while scouting; ignored')else: self.logger.info('got PROPOSE for a slot already being proposed')

Leader想要變?yōu)榛顒?dòng)狀態(tài)時(shí)會(huì)創(chuàng)建一個(gè)Scout角色，以響應(yīng)Propose在其處于非活動(dòng)狀態(tài)時(shí)收到消息（下圖），Scout發(fā)送（并在必要時(shí)重新發(fā)送）Prepare消息，并收集Promise響應(yīng)，直到聽到消息為止。多數(shù)同行或直到被搶占為止。在通過Adopted或Preempted回復(fù)給Leader。

Leader    Scout      Acceptor     Acceptor    Acceptor

   |          |          |            |           |   

   |          X--------->|            |           |    Prepare

   |          |<---------X            |           |    Promise

   |          X---------------------->|           |    Prepare

   |          |<----------------------X           |    Promise

   |          X---------------------------------->|    Prepare

   |          |<----------------------------------X    Promise

   |<---------X          |            |           |    Adopted

class Scout(Role):

def __init__(self, node, ballot_num, peers):super(Scout, self).__init__(node)self.ballot_num = ballot_numself.accepted_proposals = {}self.acceptors = set([])self.peers = peersself.quorum = len(peers) / 2 + 1self.retransmit_timer = None def start(self):self.logger.info('scout starting')self.send_prepare() def send_prepare(self):self.node.send(self.peers, Prepare(ballot_num=self.ballot_num))self.retransmit_timer = self.set_timer(PREPARE_RETRANSMIT, self.send_prepare) def update_accepted(self, accepted_proposals):acc = self.accepted_proposalsfor slot, (ballot_num, proposal) in accepted_proposals.iteritems(): if slot not in acc or acc[slot][0] < ballot_num:acc[slot] = (ballot_num, proposal) def do_Promise(self, sender, ballot_num, accepted_proposals):if ballot_num == self.ballot_num: self.logger.info('got matching promise; need %d' % self.quorum) self.update_accepted(accepted_proposals) self.acceptors.add(sender) if len(self.acceptors) >= self.quorum:# strip the ballot numbers from self.accepted_proposals, now that it# represents a majorityaccepted_proposals = dict((s, p) for s, (b, p) in self.accepted_proposals.iteritems())# We’re adopted; note that this does *not* mean that no other# leader is active. # Any such conflicts will be handled by the# commanders.self.node.send([self.node.address], Adopted(ballot_num=ballot_num, accepted_proposals=accepted_proposals))self.stop()else: # this acceptor has promised another leader a higher ballot number, # so we’ve lost self.node.send([self.node.address], Preempted(slot=None, preempted_by=ballot_num)) self.stop()

Leader為每個(gè)有active proposal的slot創(chuàng)建一個(gè)Commander角色（下圖）。像Scout一樣，Commander發(fā)送和重新發(fā)送Accept消息，并等待大多數(shù)接受者的回復(fù)Accepted或搶占消息。接受建議后，Commander將Decision消息廣播到所有節(jié)點(diǎn)。它用Decided或Preempted響應(yīng)Leader。

Leader    Commander   Acceptor     Acceptor    Acceptor

   |          |          |            |           |   

   |          X--------->|            |           |    Accept

   |          |<---------X            |           |    Accepted

   |          X---------------------->|           |    Accept

   |          |<----------------------X           |    Accepted

   |          X---------------------------------->|    Accept

   |          |<----------------------------------X    Accepted

   |<---------X          |            |           |    Decided

class Commander(Role):

def __init__(self, node, ballot_num, slot, proposal, peers):super(Commander, self).__init__(node)self.ballot_num = ballot_numself.slot = slotself.proposal = proposalself.acceptors = set([])self.peers = peersself.quorum = len(peers) / 2 + 1 def start(self):self.node.send(set(self.peers) - self.acceptors, Accept( slot=self.slot, ballot_num=self.ballot_num, proposal=self.proposal))self.set_timer(ACCEPT_RETRANSMIT, self.start) def finished(self, ballot_num, preempted):if preempted: self.node.send([self.node.address], Preempted(slot=self.slot, preempted_by=ballot_num))else: self.node.send([self.node.address], Decided(slot=self.slot))self.stop() def do_Accepted(self, sender, slot, ballot_num):if slot != self.slot: returnif ballot_num == self.ballot_num: self.acceptors.add(sender) if len(self.acceptors) < self.quorum:return self.node.send(self.peers, Decision( slot=self.slot, proposal=self.proposal)) self.finished(ballot_num, False)else: self.finished(ballot_num, True)

有一個(gè)問題是后續(xù)會(huì)介紹的網(wǎng)絡(luò)模擬器甚至在節(jié)點(diǎn)內(nèi)的消息上也引入了數(shù)據(jù)包丟失。當(dāng)所有 Decision消息丟失時(shí)，該協(xié)議無法繼續(xù)進(jìn)行。Replica繼續(xù)重新傳輸Propose消息，但是Leader忽略了這些消息，因?yàn)樗呀?jīng)對(duì)該slot提出了proposal，由于沒有Replica收到Decision所以Replica的catch過程找不到結(jié)果，解決方案是像實(shí)際網(wǎng)絡(luò)堆棧以西洋確保本地消息始終傳遞成功。

Bootstrap

node加入cluster時(shí)必須獲取當(dāng)前的cluster狀態(tài)， Bootstrap role循環(huán)每個(gè)節(jié)點(diǎn)發(fā)送join消息，知道收到Welcome, Bootstrap的時(shí)序圖如下所示：

如果在每個(gè)role（replica，leader，acceptor）中實(shí)現(xiàn)啟動(dòng)過程，并等待welcome消息，會(huì)把初始化邏輯分散到每個(gè)role，測試起來會(huì)非常麻煩，最終，我們決定添加bootstrap role，一旦啟動(dòng)完成，就給node添加每個(gè)role，并且將初始狀態(tài)傳遞給他們的構(gòu)造函數(shù)。

class Bootstrap(Role): def __init__(self, node, peers, execute_fn, replica_cls=Replica, acceptor_cls=Acceptor, leader_cls=Leader, commander_cls=Commander, scout_cls=Scout):super(Bootstrap, self).__init__(node)self.execute_fn = execute_fnself.peers = peersself.peers_cycle = itertools.cycle(peers)self.replica_cls = replica_clsself.acceptor_cls = acceptor_clsself.leader_cls = leader_clsself.commander_cls = commander_clsself.scout_cls = scout_cls def start(self):self.join() def join(self):self.node.send([next(self.peers_cycle)], Join())self.set_timer(JOIN_RETRANSMIT, self.join) def do_Welcome(self, sender, state, slot, decisions):self.acceptor_cls(self.node)self.replica_cls(self.node, execute_fn=self.execute_fn, peers=self.peers, state=state, slot=slot, decisions=decisions)self.leader_cls(self.node, peers=self.peers, commander_cls=self.commander_cls,scout_cls=self.scout_cls).start()self.stop()

以上就是詳解分布式系統(tǒng)中如何用python實(shí)現(xiàn)Paxos的詳細(xì)內(nèi)容，更多關(guān)于python的資料請(qǐng)關(guān)注好吧啦網(wǎng)其它相關(guān)文章！