On Distributed Systems

In the past month or so, so many people have been talking about blockchain that even Uncle Caitou wrote an article, “Leeks Sitting on the Ground.” I shared it on WeChat Moments with this comment:

Blockchain is a feast for technologists who dig into it, industry insiders, and those who provide services to all participants, because they have thought deeply and contributed more than money. Those who act on every rumor, spend their time fishing for tips in Telegram and staring at charts arguing up or down—they are all leeks, big or small.

A friend asked: how does a technical person get into the field?

Although I wouldn’t call myself “in the industry,” I know technology is a lineage—blockchain did not pop out of a stone like Sun Wukong; it has roots. If technologies are banyan trees rising from the ground, gradually thickening and branching over time, then blockchain is a new branch woven from several mature banyan trees—it grows quickly, with boundless vitality, and its canopy already overshadows other branches. If you climb straight onto its canopy and try to grasp its veins, you will get lost in the complexity; but if you trace patiently upward from the roots, the clues are like the “truth” in Qian Zhongshu’s writing: naked and completely exposed.

Today we first seek its most important root: distributed systems. To internet practitioners, this topic looks ridiculous—who would dare say they don’t understand distributed systems? Yet if you only hide behind a load balancer writing stateless services, and have never truly faced the delightful and melancholy uncertainty of distributed systems, then you may not really understand them, and this article is still worth reading.

As usual, let’s start with Wikipedia and get familiar with the concept:

A distributed system is a model in which components located on networked computers communicate and coordinate their actions by passing messages.[1] The components interact with each other in order to achieve a common goal. Three significant characteristics of distributed systems are: concurrency of components, lack of a global clock, and independent failure of components.

• Wikipedia

I have outlined the keywords: communication / coordination, message passing, concurrency, lack of global clock, independent failure. Let’s see how these give rise to uncertainty.

global clock

Let’s start with the global clock. It is one of the roots of many hard problems in distributed systems.

In a single-machine system, whether SMP or NUMA, there is exactly one global clock; this is easy to achieve.

What is time? Setting relativity aside, in a narrow local spacetime, time is the appearance of causality—a cause produces an effect, and this causality creates the concept of time: past, present, and future can better describe causality. We execute an instruction at t0 and get a result at t1; that result cannot appear before the instruction executes. This is the certainty time gives us. Therefore, it is very important for a system to have a consistent time that everyone recognizes and follows.

In a distributed system, each system has its own clock. Even with NTP (Network Time Protocol) synchronization, they cannot be strictly in lockstep. Even if clock differences are small enough to ignore, the delay of messages sent among systems still depends on bandwidth, congestion at the moment, CPU load, and remains highly uncertain. It’s like a team going to a meeting room: if everyone decides when to enter based on their own watch, they will surely be inconsistent; even if their watches agree, people walk at different speeds, so their final arrival times still differ. This inconsistency causes many problems, such as out of sync—everyone has left the meeting while Alice has just arrived at the venue, so she has missed many state updates and doesn’t know whether the next thing on hand should be done. Thus, in distributed systems we often need consistency to ensure certain things are ordered and everyone is on the same page; otherwise the system will go astray.

To solve out-of-sync problems caused by different clocks and inconsistent pace, we need to find a way to form a logical “clock” that everyone recognizes rather than trusting their own. The first implementation of this logical clock was Lamport timestamps (remember Lamport, Turing Award winner and pioneer of distributed systems; he will appear again below). Lamport timestamps have more academic than practical value, and no system actually uses them, but the vector clock evolved from them is widely adopted by AWS S3, DynamoDB, Riak, and other systems to ensure causality for the same object. Let’s look at the vector clock implementation:

The idea is simple: every node has a vector containing all timestamps—a logical “clock.” Each independent node handles its own timestamp to keep it ordered; but when coordination is needed (A sends a message to B), node A sends its knowledge of the “clock,” and node B, upon receiving it, updates all timestamps in its vector that are greater than those it already knows. Algorithm (consult wiki for more precise information):

1. Every node has a timestamp vector initialized to all zeros.
2. If an event occurs at node k, increment its corresponding vector[k] by 1.
3. If node k sends a message to node j, first increment node k’s own vector[k] by 1, then send the entire vector together with the message to node j. Node j increments its own vector[j] by 1, then merges node k’s vector with its own (taking the maximum).

With vector clocks, although there is no absolute global clock, we can guarantee causality in a distributed system, thereby eliminating uncertainty in this dimension (there are other uncertainties!).

As we can see, the vector clock algorithm relies heavily on trust between nodes, so it only applies to a trustworthy distributed environment. Bitcoin, which runs on a P2P network where nodes do not trust one another, cannot ensure this. So how does a distributed system like Bitcoin decide time (causality)? In Bitcoin’s design, Satoshi Nakamoto cleverly used the product of PoW, the block, as the system’s logical time:

The solution we propose begins with a timestamp server. A timestamp server works by taking a hash of a block of items to be timestamped and widely publishing the hash, such as in a newspaper or Usenet post [2-5]. The timestamp proves that the data must have existed at the time, obviously, in order to get into the hash. Each timestamp includes the previous timestamp in its hash, forming a chain, with each additional timestamp reinforcing the ones before it.

So blockchain not only carries the ledger function; each block on the chain is also a timestamp, representing the system’s past, present, and future, thereby coordinating the entire distributed system to move forward in step (let me praise Satoshi once more).

At this point, I believe many people have a question: “Programmer, why does the distributed system I build never need vector clocks or PoW, and has no global clock, yet still runs fine?” Exactly. Your not perceiving it doesn’t mean it doesn’t exist or isn’t important—Postgres, Consul, Kafka in your system, or in other words, all the seemingly centralized parts of a distributed system, use similar mechanisms; they simply shield you from these details.

coordination / communication

Okay, having understood the uncertainty brought by lack of global clock and how to deal with it, let’s look at the next fundamental component that creates uncertainty in distributed systems: communication and coordination.

In a single-machine system, coordination and communication are also easy—within the same thread, synchronization on the stack/register (depending on ABI); across threads, semaphores; across CPUs, spin locks, memory barriers. In any case, everyone lives under the same roof (clock), so there’s always a way.

In a distributed system, things get awkward. Why is Old Wang next door called Old Wang next door? Because there is at least one wall between your two houses (two walls if you live in a big house). Since you cannot look each other in the eye, your communication basically relies on shouting. “Shouting” is classical Chinese; in modern computer civilization, we call it: message passing.

Before sending a message, you must first ensure there is a suitable channel, and you must ensure the channel is established successfully and can be trusted:

“Trustworthy” means messages will not be lost on the network: as long as I send it, the other party’s application will definitely receive it (avoiding the uncertainty shown below):

Moreover, it will neither be out of order like below, nor deliver partial messages to the application (guaranteeing message integrity):

With a trustworthy channel, we can reach consensus through messages. Let’s first look at the obstacles two people encounter when trying to reach consensus:

The sending node or receiving node may crash—even if the network layer guarantees the application will receive the message, we cannot prevent the application from dying while processing it. Therefore, message delivery has two strategies: at least once or at most once. At least once means the same message will be transmitted 1 to n times; at most once means the same message will be transmitted 0 to 1 times. This is easy to understand: if the messaging system has built-in retransmission and persists messages to disk to ensure delivery even if the process crashes, then it is at least once. Conversely, if no such mechanism is built and the message is sent without further concern, that is at most once. In a network environment, message delivery can only be one of these two; it cannot be exactly once. If anyone says otherwise, they are misleading you.

at least once / at most once do not solve the uncertainty problem, so we must try harder—Kafka / AWS Kinesis / AWS SQS implement an essentially once message delivery mechanism. Essentially once is a variant of at least once and requires the message layer and application layer to work together: after the application processes a message, it proactively tells the message layer to delete the corresponding message. If you have used SQS, you should feel this: SQS guarantees that while an application is processing a message, the message is invisible to others in the distributed system. If the application crashes, the message becomes visible again after the visibility timeout. If the application finishes processing, it must explicitly delete the message.

Ultimately, we can reach agreement through message passing mechanisms:

So far, what we’ve discussed is mainly message delivery involving only you and Old Wang. In a distributed system, any two nodes may exchange messages, and because we lack a global clock, we cannot guarantee that messages are globally ordered (TCP only guarantees ordering for messages sent by the same sender). Through vector clocks or similar mechanisms, we can further guarantee that messages are ordered in terms of causality. In most cases, this is good enough.

However, with many participants, even if we guarantee message ordering locally and causally, we still cannot guarantee that all participants reach consensus. If everyone cannot agree on whether to do something (for example, when a piece of data arrives, how to write it, where to write it, who writes it), then such a system remains uncertain.

Thus we have 2PC (two-phase commit), 3PC, Paxos, Raft, and other consensus mechanisms in trustworthy environments; likewise, for the untrustworthy environment blockchain faces (the Byzantine Generals problem), BFT / PBFT were born, along with a pile of P-named consensus algorithms reliable or not—PoW, PoS, DPoS, PoI, PoD, PoDDOS (soon, P will not be enough). Given space constraints, I will discuss consensus algorithms separately.

If you want to learn more about messaging systems and message-passing patterns, you can read my earlier article: The World of 0MQ (and mourn Pieter Hintjens once more).

Pitfalls in Distributed Systems

Above we’ve gone over the most basic elements of distributed systems. Next, let’s step in some pitfalls.

Pitfall one: the network is reliable. In message passing, we go to great lengths to do many things, precisely to fight against an unreliable network. Packet loss / out of order are actually relatively easy problems; the harder problem is split brain. Split brain is the nightmare of distributed systems based on asymmetric consensus (i.e., common master-slave clusters). Once the master is too busy, the slave decides it should step down; after the slave takes over, because of cold start, it also becomes too busy, so it switches back—then master and slave are both abnormally busy for a while, replication fails, data becomes inconsistent, and next both sides believe the other is down and that they are the master. This is split brain. Problems at this level are troublesome to fix (especially for data that uses autoincrement and is referenced by foreign keys).

Pitfall two: message-passing latency is negligible. We know network latency is roughly the distance between two points divided by the speed of light. Beijing to San Francisco, under extremely ideal conditions, a round trip still takes 63ms (9516 x 2 / 300, 000). That’s not a small amount of time: if you use stop-and-wait to confirm details back and forth, you can only do a dozen round trips in 1s. The figure below introduces the magnitude of latency for parts of a complete computer system; everyone should read it carefully and get a feel:

Pitfall three: network bandwidth is not a problem. In the cloud, bandwidth is not a problem. But in a P2P network with many home users, bandwidth—especially uplink bandwidth—is clearly limited. My home 100Mbps network often has only 12Mbps uplink. In my previous article A Brief Analysis of Bitcoin, I discussed why Bitcoin uses 1M blocks: if the network’s original intent is for everyone to participate, then the bandwidth capacity of ordinary users must be considered. Suppose your computer painstakingly computes a block and you must immediately broadcast it to 7 neighbors; if you complete the broadcast within 30s, you need 1.86 Mbps (1MB x 8bit x 7 peers / 30 sec) of bandwidth; if the block is 8M, you need 15 Mbps uplink. So if you’re designing a distributed system that runs in the cloud (> 1Gbps), you need not care much about bandwidth; but if you want to build a blockchain everyone can use, at least don’t set block size off the top of your head.

Pitfall four: participating nodes are homogeneous. In a distributed system like Bitcoin, participants range from phones (wallet software) to specially developed mining machines with ASICs or GPU arrays. The networks they use range from wired to wireless to cellular and satellite. So we must consider participants that differ so greatly and span such a wide range. Miners naturally need to obtain all data, but asking phone users to download a 160G chain is unreasonable. Bitcoin specifically considers this with block headers + Merkle tree path + SPV, allowing small devices to download only a small amount of data (a few hundred megabytes) and act as wallets, verifying transactions relevant to themselves. This is something we need to learn from.

CAP Theorem

In distributed systems, one topic you cannot avoid is the CAP theorem: for Consistency, Availability, and Partition tolerance, you can guarantee only two and must sacrifice the third.

Let me explain briefly:

• Consistency: all clients see the same data. The consistency here refers to atomic consistency (linearizability).
• Availability: every client can read and write at any time.
• Partition tolerance: when a network partition (split brain) occurs, the system can still work. In other words, to support partition tolerance, the system must tolerate the loss of arbitrarily many messages.

Talking about whether a system is CA, CP, or AP is actually oversimplifying a complex problem. But classification has its benefits: it makes comparison and memory easier.

MongoDB is classified as CP in the figure above because writes default to safe=true, i.e., Availability is sacrificed and writes can only go to the master. However, in this case MongoDB still cannot really be called Consistent; only when readConcern is set to linearizability can it count as Consistent. So MongoDB’s CP label is tenuous, and under different settings it is hard to place it in a single category.

Today’s distributed systems actually consider all three of CAP; it’s a matter of priorities—which two do I value more, and which third am I willing to sacrifice? MySQL in a master/slave configuration sacrifices partition tolerance, but we can also configure it as a cluster, sacrificing Availability to achieve Partition Tolerance.