Background

The implementation of paxos-kv in 200 lines of code introduces a very concise distributed KV storage system, which is based on classic-paxos to achieve distributed consistency. In the intuitive explanation of paxos, we mentioned that each write operation, which means each paxos instance, requires 2 rounds of RPC to complete, resulting in low efficiency.

A common optimization is multi-paxos (or raft), but the flaw of this design in cross-datacenter (or cross-cloud) deployment environments is that writes from remote datacenters require 2 RTTs to complete:

client → leader → followers → leader → client

It cannot achieve multi-active across locations. In a scenario with 3 nodes, the write efficiency of 2/3 is reduced to 2 RTTs.

This article approaches the problem of multi-active across locations from another angle, in a 3-replica cluster deployed across 3 datacenters:

• Any node can write,

• Any write can be strictly completed within 1 RTT.

This is the improved version of paxoskv we will introduce today: mmp-3: multi-master-paxos 3-replica implementation.

Similarly, the principle of show me the code remains unchanged: the multi-active code implemented in this article for 3 nodes can be found at: https://github.com/openacid/paxoskv/tree/mmp3

Multi-active across locations is increasingly recognized as an important issue in the distributed field. Datacenters are becoming single machines, and the multi-machine distributed model in a single datacenter can no longer meet the availability requirements of large-scale deployments.

Almost all distributed storage deployed in online environments need to be deployed across datacenters (or across clouds). Everyone is actively working to solve these problems:

• Either use queues or other eventual consistency methods to complete cross-datacenter replication, which may lead to data inconsistency, as two conflicting pieces of data may be written simultaneously; the business layer needs to participate in resolving such conflicts.

• Or split the data, assigning more writes in location A to be led by the datacenter in A, and assigning more writes in location B to be led by the datacenter in B.

• Or designate one datacenter as primary: deploy 2 replicas in one datacenter and 1 replica in another datacenter to form a 3-replica cluster. This, however, means that a failure in datacenter A will lead to global unavailability, while datacenter B can only provide additional data redundancy without increasing data availability.

Paxos can achieve 1 RTT writes in smaller clusters by customizing paxos. If using majority-quorum, it supports up to 5 replicas for multi-active.

The multi-active design defined in epaxos briefly introduces the design of 3 nodes but does not provide implementation details. The handling of various conflicts and repair processes is not clearly defined.

• Additionally, the apply algorithm in epaxos has an unsolvable livelock problem: determining the instance order through SCC cannot guarantee completion within a finite time.

• Furthermore, the design of epaxos lacks a rnd record (the last-seen-ballot or vbal in paxos), leading to incorrect consistency implementation.

• And the dependencies between instances can create inconsistencies during the repair process.

• Epaxos requires another seq to determine the order between instances, whereas in the design of mmp3, seq is unnecessary; it only needs dependencies to determine the apply order.

Multi-Master Paxos – 3

We start by analyzing the problem from classic-paxos.

xp’s tips: To implement a stable distributed system, it is best to use raft, as it works out of the box. To learn distributed systems, it is best to start with paxos. The seemingly simple design of raft hides some obscure conditions, and its correctness proof is more complex than that of paxos.

We need to achieve two objectives:

• Complete a commit in 1 RTT.

• Allow 3 nodes to write simultaneously without conflicts.

1 RTT Classic-Paxos

If classic-paxos does not require 2 RTTs, we do not need multi-paxos or raft to optimize latency.

This is achievable in a 3-node system.

First, let’s make some basic assumptions: A replica in the system is a replica (or called a server or node), and it is both a proposer and an acceptor. When a replica receives a write request, it uses its local proposer to complete the commit.

Review Classic Paxos

The write process of classic-paxos introduced in paxos-kv in 200 lines of code is as follows, where the proposer P0 on replica-0 sequentially completes phase-1, phase-2, and commit:

🤔 Think about the above process…

Optimizing Classic Paxos to 1 RTT

Since the proposer itself is merely a data structure, in paxos, it does not need to have any binding relationship with acceptors. Therefore, we can allow the proposer to run on any replica: sending the proposer to run on another replica allows the transmission of messages to be transformed into the transmission of the proposer.

To achieve the 2/3 majority required by paxos, it only needs to send the proposer to another replica. Since this proposer will always have only one instance, there will be no inconsistency (the proposer either works on R0 or on R1).

If the proposer were sent to 2 replicas, it would become more complicated; for example, in a 5-node system with quorum=3, two different proposers might attempt to use different values.

By sending the proposer, paxos can be optimized into the following 1 RTT implementation: P0 sequentially executes phase-1 and phase-2 on R1, and then is sent back to R0:

In the process of transmitting the proposer, unlike the original paxos, both round-trip processes must include the complete information of the proposer:

• In the process from R0 to R1, it must carry the value the user wants to submit so that after Prepare succeeds on R1, it can directly run Accept;

• In the process from R1 to R0, it must carry the execution results of Prepare and Accept from R1.

After this one round of RPC, R0 and R1 can form a majority, and then R0 can directly commit.

Note that in this model, aside from the change in the location of the proposer, there is no difference from classic-paxos! This means that anything that paxos can accomplish can also be accomplished here.

Now we have completed the first task. If we rewrite the 200 lines of code implementation of paxos-kv using this model, we can achieve 1 RTT commits on a 3-replica system, but multiple write points will still have conflicts. For example, if R0 and R1 simultaneously initiate a write for the same paxos instance, R0 may find that the local instance has already been covered by P1 with a higher ballot after receiving the returned P0, and will need to elevate P0’s ballot and retry.

This is the second problem we need to solve: avoiding write conflicts between different replicas.

Multi-Column Log

When 2 replicas simultaneously write to one instance, it creates a livelock, preventing 1 RTT writes from being guaranteed. To avoid conflicts, we need to ensure that each replica cannot generate conflicting instances, so we need to allocate separate instance spaces to each replica.

In the implementation of mmp3, with 3 replicas, we need 3 columns of instances, with each replica only writing to one column.

For example:

• R0 maintains a proposer P0, continuously running paxos on each replica’s column A instances,

• R1 maintains proposer P1, only writing to each replica’s column B instances.

This structure is somewhat similar to 3 standard raft groups, each deployed on 3 replicas, where the leader of the i-th raft group is R[i].

Thus, because there are no instance conflicts, any write request received on any replica only requires 1 RTT to complete the instance submission.

However!

The instances in these 3 columns are currently still independent. In order to apply the instances to the state machine, all instances on each replica must be applied in the same order. (Unlike the instances in raft, which are simply monotonically increasing; as long as the instances are consistent, the apply order will be consistent.)

Therefore, in mmp3, in addition to the consistency of instance content, additional constraints between each column’s instances need to be added to ensure consistent apply order. The instances in the 3 columns have a (weaker but consistent) topological order, thus in mmp3, the value that paxos needs to determine (Value) includes two:

• The data the user wants to submit: a log operation for the state machine: instance.Val,

• It is also necessary to determine the relationship of this instance with other instances.

Using Paxos to Determine Relationships Between Instances

This relationship is described as: an instance X sees which other instances: represented by X.Deps, which is used to determine the apply order between instances:

For example, in a single-machine system, concurrently writing 3 pieces of data a, b, c, the order can be determined as follows: If a did not see b when it was written, then a runs before b. Thus visibility indicates the order between instances.

Of course, this idea is more complex in distributed systems, as multiple replicas do not have the lock protection found in single machines, and the same instance seen on different replicas may also differ.

Ultimately, the data structure of instances in mmp3 compared to classic-paxos adds a Deps field:

• instance.Deps: which other instances were seen.

The implementation of Deps includes changes in the following steps:

Proposer Chooses the Value of Deps

Based on the 1-RTT classic-paxos:

• When initializing instance X (i.e., after creating X, when executing prepare on the local replica), initialize the set of all known instances on the current replica as X.Deps (including all instances visible on the replica, as well as those instances seen by these instances, although indirectly visible instances may not exist on the current replica),

• When executing accept, the final value of X.Deps is the union of the Deps values obtained from two prepares to be used as the accept value.

For example, for instance a4, on the replica that created it and on the other replica it was copied to, it saw b2, c2 and b1, c3, respectively, leading to the two a4.Deps values being: [4, 2, 2] and [4, 1, 3]:

Thus, the Deps value used for running accept for a4 would be [4, 2, 3]:

In classic-paxos, the values seen during the prepare phase must be used, while in mmp3, the values of all Deps seen during the prepare phase are taken as the union. This does not violate any constraints of paxos; it merely assumes that the value in classic-paxos is arbitrary and not necessarily the union. In mmp3, the values of Deps seen during the prepare process can be regarded as a value with VBal equal to 0.

Readers can verify that this does not violate any constraints required by classic-paxos.

Since the value of X.Deps is also determined through paxos, it ensures that the Deps of each instance submitted on each replica are consistent. At this point, using a determined algorithm based on the value of each instance’s Deps to decide the apply order can ensure that the final state of the state machine is consistent across multiple replicas.

The above two points satisfy the first requirement of the apply algorithm: Consistency. Furthermore, the apply order also needs to provide another guarantee called Linearizability: if propose A occurs after commit B, then A should be applied after B.

This is an intuitive requirement: if a command set x=1 is sent to the storage system and returns OK (committed), then any subsequent get x command sent to the storage should definitely see the value x=1.

In fact, xp believes that using absolute time precedence globally in a distributed system is not a rational choice. However, it is easier for businesses to use.

Next, we design an algorithm to meet the requirements of Linearizability:

Apply Algorithm: Ordering Nodes in a Directed Graph with Cycles

Interfering Instances

In mmp3, it is set that any 2 instances are interfering, meaning that swapping the apply order of 2 instances will lead to different results (even if they may be interchangeable).

Epaxos considers that the instances set x=1 and set y=2 can be applied in any order because the value of x is independent of y, but the instances set x=y and set y=2 cannot be applied interchangeably, as the change in order will produce different results for x. This is also why epaxos needs to reduce the number of interfering instances to achieve 1 RTT, hence this design.

In a 3-replica system, mmp3 requires only 1 RTT regardless of conflicts, so we need not worry about the additional RTT cost caused by conflicts of interfering instances. We can simply assume that any 2 instances are interfering, which simplifies the problem.

Lemma-0: Dependency Relationships Between Instances

Define A depends on B, i.e., A → B if A.Deps ∋ B.

Since mmp3 assumes that any 2 instances are interfering and that any 2 instances submitted must have an intersection in their quorum, there must be at least one dependency relationship between any 2 instances, meaning that the relationship between A and B can only be:

• A → B

• B → A

• A ↔ B

Dependency relationships form a directed graph that may have cycles, for example, executing in the following time order:

• R0 proposes a1, a1.Deps = [1, 0, 0],

• R1 proposes b1, b1.Deps = [0, 1, 0],

• R0 sends a1 to R1, a1.Deps = [1, 1, 0]

• R1 sends b1 to R0, b1.Deps = [1, 1, 0]

• R0 commits a1

• R1 commits b1

Thus, a1 ∈ b1.Deps and b1 ∈ a1.Deps.

Dependency relationships are intuitive. In this dependency relationship graph, we will attempt to find a finite-sized set to implement an effective apply algorithm.

Lemma-1: Using Deps to Determine Linearizability

First, we have a small conclusion:

If A is proposed after B is committed, then A.Deps must include all instances in B.Deps.

Because if B has been committed, then B.Deps, which is the set of all other instances seen by B, has already been replicated to some quorum. Therefore, when A runs paxos, it must see the value of B.Deps that B committed.

Moreover, since A.Deps is the union of the Deps values seen during the prepare phase, A.Deps must include all instances in B.Deps.

Thus, the thought process for implementing the apply algorithm is:

• If A.Deps ⊃ B.Deps, apply B first to ensure Linearizability.

• In other cases, the order of applying does not violate Linearizability, so mmp3 uses the size sorting of the instance’s (columnIndex, index) to determine the apply order.

Epaxos provides a simple and crude method to determine the apply order in a directed graph with cycles: starting from a node in the graph, find the largest strongly connected subgraph (SCC) (a node without outgoing edges is also an SCC), then sort the nodes in that SCC by some property (for example, epaxos uses (seq, instanceId) to sort the nodes) and apply them in order.

However, epaxos’s SCC algorithm has a problem: an SCC may grow indefinitely. For instance, if A is committed before another interfering instance B is proposed, and then another interfering instance C appears before B is committed, …

In this case, epaxos’s approach cannot guarantee finding the SCC within a finite time.

Epaxos suggests pausing the proposal of new instances for a short period to break the SCC, which is also not easy to implement since it requires n-1 replicas to pause simultaneously. As long as 2 replicas continue to write new instances, it is possible to create an arbitrarily large SCC.

Lemma-2: No Need for SCC

The second conclusion:

If A and B do not belong to the same SCC, i.e., A ∈ SCC₁ and B ∉ SCC₁, then:

• A → B ⇒ A.Deps ⊃ B.Deps.

• B → A ⇒ B.Deps ⊃ A.Deps.

Because according to Lemma-0, any 2 instances must have at least one dependency relationship. If X ∈ B.Deps and X ∉ A.Deps, then there must be X → A, leading to A → B → X → A forming an SCC.

Thus, regardless of whether A and B are in the same SCC, the conditions for guaranteeing Linearizability can be determined using Deps, so our algorithm does not need to find SCCs but only needs to traverse the dependency relationships.

Reducing Traversal Count: Only Consider the Oldest Instance

The above apply algorithm can be further optimized to consider at most 3 instances:

Assuming a1 and a2 are two adjacent instances on column-A, then it must hold that a1 ∈ a2.Deps. According to the apply algorithm design, a1.Deps ⊃ a2.Deps must not hold; thus, a2 will not be applied before a1:

• If a1 does not depend on a2, a1 must be applied first.

• If a1 depends on a2, but a1’s (a3.columnIndex, a3.index) is smaller, then a1 will also be applied before a2.

Therefore, we only need to consider the oldest un-applied instance on each column to find the next instance to apply. In mmp3, there can be at most 3 (but the algorithm itself is not limited to 3).

Lemma-3: The Number of Deps Sets Determines Linearizability

Define a dependency count: |X.Deps| as the number of columns where X depends on un-applied instances.

For example, a3.Deps = [3, 2, 2]:

• If the applied instances are [2, 1, 1], i.e., a1, a2, b1, c1, then at this point a3 depends on an un-applied instance in all 3 columns: |a3.Deps|=3.

• After c2 is applied, then |a3.Deps| = 2.

It is clear that a conclusion can be drawn: A.Deps ⊃ B.Deps ⇒ |A.Deps| > |B.Deps|.

Finally, the apply algorithm is:

Find the next committed, un-applied instance X on a column, traverse X.Deps, and obtain the oldest un-applied instance on columns that have not been traversed. After the traversal, select the instance to apply to the state machine that has the smallest (|X.Deps|, X.columnIndex).

On the next apply, reconstruct this graph and find the second instance to execute.

It is necessary to traverse again because the second instance sorted previously may still be the second after a new instance is added.

Thus, on each replica, the Deps values of committed instances are the same, and the dependency graph formed by the oldest 3 instances is also the same. Therefore, finding the first instance to apply will be the same, and repeating this step will yield the second instance to apply, and so on, until the state machine on each replica reaches a consistent state, ensuring Consistency.

Example of Apply Execution

For example, the following 20 instances have a Deps relationship that forms a directed graph, and the final generated apply order is a single path:

RPC Timeout Retries

Paxos assumes it operates in an unreliable network environment. In standard implementations, if a request times out, it should theoretically be retried. The running environment of mmp3 assumes the same as classic-paxos, and also requires timeout retries. Here, there is a slight difference from classic-paxos, which is that when retrying, it must elevate its own BallotNum, re-execute prepare locally, and resend RPC with the new BallotNum.

This is because during the prepare process, the Deps values obtained on each replica may differ.

For example, the instance X proposed by R0 may receive different X.Deps values after prepare on R1 and R2 (the instances contained in the two replicas differ). Using the same BallotNum cannot distinguish which one is the latest value. Retrying with an elevated BallotNum ensures that the final determined value can be recognized.

A repair process (for instance, after R0 crashes, R1 or R2 can rerun paxos to repair) can discover this issue through timeout checks. The content to be repaired still includes two: the command Val to be executed by the instance, and the instances it sees: Deps.

Since these two values are established through classic-paxos, the repair process is also straightforward: elevate BallotNum and run paxos once again. This is equivalent to seizing leadership from R0 and giving it to another replica.

Code and Testing

The git repo mmp3 contains the multi-master three-replica implementation described in this article (mmp3 branch), where the main logic for instance submission on the server side is implemented in mmp.go, and the apply algorithm is implemented in apply_*.

In addition to basic unit tests, the most important test is: Test_set_get, which conducts random read and write stress tests on a three-replica cluster. This test simulates sending and receiving network errors (20% probability each), and in this case, checks:

• All write requests are submitted.

• Instances on the 3 replicas are consistent.

• The apply order on the 3 replicas is consistent, as well as the final state in the state machine.

Limitation

The design of mmp3 only supports 3-node systems. Additionally, this implementation does not include member change implementations.

Conclusion

mmp3 is a fully peer-to-peer design implementation of multi-master consensus. Previously, while trying to implement a multi-master storage based on epaxos, several issues that were not easy to fix were discovered (initially there were a few easily fixable problems), leading to the decision to design a new system.

Looking forward to exchanging ideas with anyone interested in this direction~

Reference:

• 200 lines of code implementing a paxos-based kv storage.

• Classic paxos: http://lamport.azurewebsites.net/pubs/pubs.html#paxos-simple

• Reliable distributed systems – an intuitive explanation of paxos.

• Multi-master-paxos-3: https://github.com/openacid/paxoskv/tree/mmp3

• Minority implementation of majority read and write.

Article link: https://blog.openacid.com/algo/mmp3

Recommended Reading

• Baido’s large-scale Service Mesh implementation practice.

• How to cope with 1 billion-level traffic during the Spring Festival red envelope activity? See the big shots’ summary.

• The path of microservices splitting.

• From traffic marking to machine marking – Dada’s full-link pressure testing exploration and practice.

• The tragedy of hybrid deployment – discussing CPU isolation technology for cloud-native resource isolation.

Original and architectural practice articles are welcome for submission through the WeChat public account menu ‘Contact Us’.

The 2021 GIAC will be held on July 30-31 in Shenzhen. Click to read the original text for more details.