Designing Data-Intensive Applications 讀後心得 (3)

在前面的心得提到分散式系統的 replication 和 partitioning，以及單機的 transactions。這篇要介紹如何在分散式系統裡處理 transactions 或其它跨多節點的同步操作。

本文的附圖皆取自書中。

Distributed Transactions

假設基於效率需求，已用 user id 作 partitioning。若要記錄 Alice 轉帳給 Bob 並且希望更新是一個 atomic commit，要怎麼做呢？若 Alice 和 Bob 在同一個 partition，用 transaction 即可。但若在不同 partition ，該怎麼辦？

此時要用 distributed transaction，其中一種實作方式是 two-phase commit (2PC)，概念如下圖:

Coordinator 產生全域唯一的 transaction ID，傳送 transaction 到 Database 1 和 2，讓它們分別執行 transaction。然後分兩階段完成作業:

1. Coordinator 送 prepare request 給 Databases。
2. Coordinator 送 commit request 給 Databases，收到 commit request 的 Database 此時才執行 transaction。

這個作法的正確性來自兩個前提:

• Database 一但同意 prepare request 後，就不能改變決定，即使重開機也一樣。一定要等到 commit/abort 才可進行對應動作。
• Coordinator 一但決定要發 commit request 或 abort request，要保證這個決定永遠不會改變。即使重開機也是如此。

這兩個前提很硬，出錯會很慘，要人工介入處理。下圖是一個出錯的情境:

X/Open XA 是實作 2PC 的標準，像 PostgresSQL、mySQL、SQL Server、Oracle 等資料庫都有支援。所以可以用它來跨不同資料庫處理。書上提到多數 XA 的 Coordinator 實作預設沒有 high availability 或是很基本的備份功能，所以 Coordinator 成了分散式系統裡的弱點 (single point of failure)。

Linearizability

如果存在一個「如同只有單份資料並且只提供 atomic 操作」的 database，那很多跨節點的同步需求可透過它處理。像是 Alice 轉帳 500 給 Bob，可以建一個 task ID，然後記錄狀態:

1. ID 1234: Alice -500 (未執行); Bob +500 (未執行)
2. ID 1234: Alice -500 (已執行); Bob +500 (未執行)
3. ID 1234: Alice -500 (已執行); Bob +500 (已執行)

注意，更新金額有可能失敗，所以 idempotent 的更新會比較方便使用，不然要能正確判別是否曾經更新過。

事實上，有許多類似的問題可以用這樣的 database 解決:

Linearizable compare-and-set registers: The register needs to atomically decide whether to set its value, based on whether its current value equals the parameter given in the operation.

Atomic transaction commit: A database must decide whether to commit or abort a distributed transaction.

Total order broadcast: The messaging system must decide on the order in which to deliver messages.

Locks and leases: When several clients are racing to grab a lock or lease, the lock decides which one successfully acquired it.

Membership/coordination service: Given a failure detector (e.g., timeouts), the system must decide which nodes are alive, and which should be considered dead because their sessions timed out.

Uniqueness constraint: When several transactions concurrently try to create conflicting records with the same key, the constraint must decide which one to allow and which should fail with a constraint violation.

這裡的關鍵就是要有 linearizable 的系統，基本概念就是「如同只有單份資料並且只提供 atomic 操作」。以下圖為例:

三個 client 讀寫 x，上面的區間是發出 request 到收到 response 的時間。在這區間中，x 的狀態可能是 0 或 1，重點是一但被任一 client 觀測到狀態改變，之後其它 client 要看到一樣的狀態，像是下圖:

有興趣的話，書上有提供更詳細的拆解，說明符合和不合 linearizability 的狀況，見下圖:

之前說明 transactions 時有提到 serializability，它和 linearizability 是不同概念:

• Serializability: 描述 transactions 之間執行順序有如依序執行。對兩個沒有交集的 transactions 來說，可以同時它們仍符合 serializability。
• Linearizability: 描述單一物件 (非 transaction) 的操作都是 atomic。所以 linearizability 無法保證沒有 write skew (要支援 serializable isolation 的 transaction 才行)。

附帶一提，作者認為 CAP Theorem 容易讓人搞混，比較正確描述 CAP Theorem 的方法是 “either consistent or available when partitioned”。因此，不如直接討論系統有支援 linearizability (如同選擇 consistency) 或是不支援 (如同選擇 availability)。

Total Order Broadcast (Atomic Broadcast)

書中花不少篇幅說明用 total order broadcast 完成一些工作，最後會導出「能實作出 total order broadcast」= 「實作 consensus protocol」。我猜可能是 ZooKeeper 的 consensus protocol 是 ZooKeeper Atomic Broadcast，所以多介紹何謂 atomic broadcast。

我先前讀過一些 blockchain 的 consensus protocol (PaLa、LibraBFT)，有點懶得讀 atomic broadcast。日後有機會再看看吧。

Fault-Tolerant Consensus

前面提到 2PC 的 Coordinator 是 single point of failure，有沒有更好的作法呢？既然有 linearizability 的系統可解決一切問題，要支援 linearizability 得用 consensus protocol，不如就用 consensus protocol 吧。重點是已有像 ZooKeeper 現成的工具可用。

如果不需要 fault-tolerant (像是單機 crash、部份斷網)，那單機 database 就可滿足目的。如果 single leader 掛了，就人工指定下一台。但需要 fault-tolerant 時，就真的得用 consensus protocol。

多數 consensus protocol 會用到 leader-follower 的架構，但和先前說的 single leader 主要差異是會定義 epoch (一個數字，在 Paxos 稱作 ballot number，在 Raft 稱作 term number)，並且每個 epoch 選出唯一的 leader。

這些 Consensus protocol 大致規則是:

• leader 由過半成員決定。
• leader 提案更新資料，提案要獲得過半成員同意。
• 成員只投票給他們認為最新 epoch 的 leader。

雖然 consensus protocol 和 2PC 看起來很像，主要差異有:

• 2PC 的 coordinator 不是選出來的。
• 2PC 要全員回應，consensus protocol 只要過半。
• consensus protocol 有同步狀態的協定，讓有問題的節點可追回最新狀態。

附帶一提，通常 consensus protocol 的節點數是固定的，還有用 timeout 來判斷其它成員是否有問題。

其它雜項

ZooKeeper

記錄幾個 ZooKeeper 相關的事:

• 提供 linearizable write，但預設有可能讀到舊資料，也有提供 linearizable read，會比較慢。
• Apache Curator 是使用 ZooKeeper 的函式庫，提供更高階的操作，像是 distributed lock 或 leader election。

Lock 和 Fencing Tokens

使用 distributed lock 可能會有意外狀況，如下圖所示 ：

這個例子是 client 1 取得 lock 後，所用的執行環境執行 garbage collection，因此暫停操作。等 GC 返回後已超出使用 lock 的時間，但應用程式沒有自覺，而沒有發揮 lock 的效果。

下圖是解法之一，稱為 fencing tokens:

取 lock 的時候同時獲得全域唯一的流水號，寫回 storage 時要帶上流水號，storage 會看流水號拒絕超時的操作。ZooKeeper 的 transaction ID zxid 可用作流水號。

Linearizability 和 Quorums

前面提到 fault-tolerant consensus protocol 有用 quorum，是否單用 quorum 就能有 linearizability？答案是否，下圖是反例:

• Writer 將x 由 0 改為 1。
• Reader A 觀察到 x = 1，但 Reader B 卻觀察到 x = 0 (舊的狀態)。

FLP Result

證明沒有演算法可以作到 consensus，不過這個證明是基於 asynchronous system model，不能用 timeout 的前提。實務上會用 timeout，所以有方法可以作到 consensus。

結語

跨節點同步操作需要 linearizability，作法和從單節點到多節點過程一樣:

• 先單機提供 linearizability，容易實作、效率好。
• 需要 high availability 所以需要replication。
• 用 single leader replication 比較單純，但是要支援自動 failover，避免成為 single point of failure。
• 為了 failover，需要多數節點即時取得共識，所以使用quorum，還有要有自動同步狀態的協定。

提供 linearizability 的系統效率必然比沒提供的系統差，所以只有針對必須有linearizability 的工作 (例如選 leader、distributed lock)，才使用它。

許多系統像是 Kafka 將此專業外包給如 ZooKeeper 這類輔助系統，以提供 high-availability 的 linearizability。

相關文章

• Designing Data-Intensive Applications 讀後心得 (1): 談分散式系統的 replication 和 partitioning。
• Designing Data-Intensive Applications 讀後心得 (2): 談單一節點的 transactions 是什麼。
• Designing Data-Intensive Applications 讀後心得 (3): 談跨節點維持一致資料的方法。
• Designing Data-Intensive Applications 讀後心得 (4): 前三篇挑重點章節講，這篇依章節順序摘要各章內容，大概做個記錄。