Ingredients of Storage System — Part II

TL;DR: This note is a follow up of part I and captures some of my learnings about distributed storage system.

Key Problems for Distributed Storage System

Distributed computation is necessary to scale and make computer system fault tolerant. The two key techniques in distributed system are partition and replication.

Partitioning makes it possible to store more data that individual machine can support. Partitioning also enables parallelism to reduce the completion time of jobs. However, implementing transactions across partitions is more challenging and expensive.

Replication is key technique to achieve fault tolerance in distributed system. If data is replicated on multiple servers, the failure of a server or even multiple servers will not affect the overall availability of the system. However, it is desirable to have a logical consistent view of the data across all the replicas. This is challenging since servers are physically independent and practically there is no way to make them equivalent. We need rules and processes to derive a consistent logical view from the replicas.

Another important concern is ordering of events. The state of the system can be modeled as application of a sequence of events. Different ordering can map to different state. Ordering events on a single server with multiple CPUs is relatively easy to achieve using synchronization primitives. The independence of physical servers makes it impossible to define meaningful total ordering between events across multiple servers. However, it is possible and useful to define partial ordering in distributed system.

To summarize, following are three key problems we need to address in distributed storage system:

• Consistent View of Replicas
• Distributed Transaction
• Ordering of Events

For the first problem, please refer to a note I wrote earlier. We will discuss about the second and third problem in following sections.

Distributed Transaction

Two phase commit is an old technique to implement transactions across multiple systems. In two phase commits, there are two kinds of roles: coordinator and participant. Coordinator will send request to participants and use the response from participants to make an overall decision.

From https://medium.com/geekculture/distributed-transactions-two-phase-commit-c82752d69324

In practice, two phase commit is not commonly used. The reason is that coordinator and participant failures are hard to handle. For example, if coordinator fails, we will have no way to know the outcome of a transaction. There are algorithms to try to figure out what happened based on the surviving servers, but these algorithms are complicated and still have gaps.

One of the key insights in Spanner is that failures, no matter in coordinator and participant, can be handled using the standard technique in distributed system — replication.

Hybrid Time

In distributed system, there are two kinds of time: physical and logical.

Physical clock on different servers at the same instant might be different. In a data centers, time on different servers is typically synchronized using protocol such as NTP so the time difference between servers is bounded. Google Spanner uses atomic clock to reduce the time drift between servers to a few milliseconds. Without atomic clock, the time drift within a data center can be bounded within a few hundred milliseconds.

Logical clock is another clock Lamport proposed to track the causal relationship in distributed system. Essentially each process keeps a counter and increases the counter whenever there is an event. When processes communicate with each other, the receiver side will update its counter to be larger than the sender counter. Informally, this clock preserves happens before relationship (->) in distributed. Happens before relationship are defined for two cases: (1) events in the same process (2) send and receive event for two processes communicating.

Hybrid time is pair of clocks <physical, logical>. The physical component is time on servers. The logical component is a counter. The generation and update rule of hybrid time is very similar to logical clock, so it provides the same property logical clock provides. In addition, Hybrid time helps us to know roughly when an event happened​. We will see why this additional benefit is useful later in this note.

Following is the Pseudocode for Hybrid Time (from this blog):

type Timestamp of {int physical, int logical}

 int last physical = 0
 int next logical = 0

 PhysicalClock pc

function NOW : Timestamp
 Timestamp now;
 int cur physical = pc.now().physical;
 if cur physical ≥ last physical then
 now.physical = cur physical;
 now.logical = 0
 last physical = cur physical;
 next logical = 1;
 else
 now.physical = last physical;
 now.logical = next logical;
 next logical++;
 end if
 return now;
 end function

function UPDATE(Timestamp in) : void
 int Timestamp now = Now();
 if now.physical > in.physical then
 return;
 end if
 last physical = in.physical;
 next logical = in.logical + 1;
 end function

‘NOW’ is called when we need to generate a timestamp for an event. ‘UPDATE’ is called when we receive a message from a remote process.

MVCC for Distributed System

We explained how MVCC works on a single server. We can recap the desired properties of MVCC here:

• Reader should see writes from committed transactions.
• Reader should not see writes from transactions committed after reader started.

The key to implement MVCC is managing the timestamps of transactions and records. The challenge for implementing MVCC in distributed system is that timestamps can be allocated at different nodes that have different time. Two bad cases are possible due to the time difference: (1) committed transactions might have higher timestamps than the timestamp for reader so we could miss records of committed transactions (2) transactions can also commit at lower timestamp after reader started so reader can see different versions of a record.

There are few solutions to this:

• Global Time Oracle
• TrueTime
• HybridTime

Global time oracle means we have a central place to allocate timestamps. This is the method used in Google Percolator. The benefit is that system is quite simple. In theory, the oracle can become a bottleneck. But given how little the Oracle does, this will not be an issue in practice. Sharing the same oracle across data center can introduce high latency, so Percolator is designed to work inside a single data center.

As mentioned earlier, TrueTime in Spanner can guarantee uncertainty time bound within a few milliseconds. Spanner adds an additional wait step after transaction commits and before sending responses to client. This makes sure any reader transaction initiated by the client after receiving the response will have higher timestamp than committed transactions.

We mentioned earlier that we can achieve some uncertainty bound without TrueTime, but the bound is too large so adding wait like Spanner will introduce long latency into transactions. With Hybrid time, we can detect potential ordering issues and retry readers if necessary. This might delay the readers, but it only happens when there are concurrent read/write within some time bound.

Case Study — YugaByteDB

YugaByteDB is a Geo-Distributed transactional system. It is a good example for us to understand how the techniques discussed earlier can be combined to build a real system. The official YugaByteDB doc is well-written and contain a lot of useful details. I will capture some key points I learned in this section.

Following is the write path in YugaByteDB:

From Transactional IO path | YugabyteDB Docs

A transaction can generate provisional records on multiple servers. The provisional records are replicated using Raft (a popular consensus protocol.) After all the provisional records are written, the transaction manager (coordinator in 2PC) can decide to commit the transaction. The commit status is stored in a status tablet, which is replicated. The coordinator and participant failure in 2PC is handled like Spanner. The status tablet also achieves atomicity in commit. The moment the tablet acks the status record is committed is the moment the transaction is committed.

The status update is shared to the participants asynchronously. Provisional records will be turned into official records with transaction commit timestamp and cleaned up afterwards. A natural question at this point is that how we can make sure readers will see the changes of a committed transaction if we update status and records asynchronously. This question will be answered below.

For a committed transaction, the records on all involved tablets can be in three different states:

• All records for a committed transaction have been changed to official records
• Provisional records on some tablets are turned into official records
• None of the tablet have changed provisional records into official records

Obviously, all these cases should be equivalent as far reader is concerned. To achieve this, reader transactions need to examine both provisional records and official records. Readers will check if records belong to a committed transaction by talking to the status tablet. This check was done on the tablet that reads the data.

From Transactional IO path | YugabyteDB Docs

If we think about it for a while, there is a race condition that can cause problem. A transaction might change its commit status after the reader starts. It is possible reader will be told that the transaction is not committed earlier and will be told the transaction is committed later. The key to address this problem and many others in YugaByteDB is using timestamps.

Before we discuss the solutions in detail, it is useful to look at how various timestamps are chosen:

• Reader timestamp: reader timestamp is chosen as the hybrid timestamp of the server that is serving the request.
• Provisional records: the timestamp of a provisional record is the hybrid time on the leader tablet server.
• Commit timestamp: the committed timestamp is the timestamp on the status tablet. Given the rules to update timestamp, committed timestamp will be higher than all provisional records.

When the commit timestamp is within uncertainty bound, YugaByteDB will retry the reader transactions. The retry can help avoid the inconsistent commit status on different tablets involved in reader.

As mentioned earlier, another bad case we want to avoid transactions to be committed at a lower timestamp after a reader starts. In YugaByteDB, this is achieved by defining a safe time at each tablet a reader will talk to. Request (which carries a timestamp) from the reader will be hold until the safe time on all the tablets pass the reader timestamp. This will guarantee that new events on these tablets will get higher timestamp. So, it is impossible for new transactions to be committed at lower timestamp.

References

• Ingredients of Storage System — Part I
• How to Understand Paxos?
• CSEP 545 Lectures Archive (washington.edu)
• MIT 6.824 Distributed Systems (Spring 2020) — YouTube
• CSE138 (Distributed Systems) lectures, Spring 2021 — YouTube
• Transaction Processing: Concepts and Techniques
• Transaction Processing: Management of the Logical Database and its Underlying Physical Structure
• Designing Data-Intensive Applications
• Database System Concepts
• Database Systems: The Complete Book
• The Internals of PostgreSQL : Introduction (interdb.jp)
• Living Without Atomic Clocks | Where CockroachDB & Spanner Diverge
• Yes We Can! Distributed ACID Transactions with High Performance
• CockroachDB’s Consistency Model (cockroachlabs.com)
• Distributed SQL Databases Deconstructed | YugaByte — YouTube
• Raft Consensus Algorithm
• Consensus Protocols: Two-Phase Commit | Paper Trail (the-paper-trail.org)
• OSDI12 — Spanner: Google’s Globally-Distributed Database — YouTube
• Spanner Internals Part 1: What Makes Spanner Tick? (Cloud Next ’19) — YouTube
• Spanner Internals Part 2: Global Meta-Data and Scalable Data Backend