Distributed Systems Course by Martin Kleppmann

[Sunt-ing]

https://www.youtube.com/watch?v=UEAMfLPZZhE&list=PLeKd45zvjcDFUEv_ohr_HdUFe97RItdiB

There is a note that you may find useful: https://www.cl.cam.ac.uk/teaching/2122/ConcDisSys/dist-sys-notes.pdf

Also, I have taken a note (in Chinese) for my distributed system course during my college life: https://github.com/Sunt-ing/SUSTech-CS-Notes/tree/master/%E5%B9%B6%E8%A1%8C%E4%B8%8E%E4%BA%91%E8%AE%A1%E7%AE%97

[Sunt-ing]

1.1: Introduction

• Lamport joke: The distributed system is "a system in which the failure of a computer you didn't even know existed can render your own computer unusable."

1.2: Computer networking

• "Hard drives in a van" is actually an efficient way to transfer massive data and AWS actually provides such service!
  • Actually, this reminds me of IP over Avian Carriers.
• Some figures (similar to Jeff Dean's)
  • Network latency
  • In the same building/datacenter ≈ 1 ms
  • One continent to another ≈ 100 ms
  • Hard drive in a van ≈ 1 day
  • Network bandwidth
  • 3G cellular data ≈ 1 Mbit/s
  • Home broadband ≈ 10 Mbit/s
  • Hard drive in a van 50 TB/box ≈ 1 Gbit/s

2.1: The two generals problem

• The two generals problem is first proposed by Jim Gray.
• It can be proved that the two generals cannot reach certainty by exchanging any finite number of messages.

2.2: The Byzantine generals problem

• The word Byzantine had been used in the sense of “excessively complicated, bureaucratic, devious” long before Leslie Lamport adopted the word to describe the Byzantine generals problem.

2.3 System models

• System models
  • Two generals problem: a model of networks, while the nodes are assumpted to be correct.
  • Byzantine generals problem: a model of node behaviour, while the network connections are assumpted to be correct.
• Network behaviour
  • Arbitrary link can be transformed to fair-loss link by TLS
  • Fair-loss link can be transformed to reliable (perfect) link by retry+deduplicate
• Node behaviour
  • Crash-stop (fail-stop)
  • Crash-recovery (fail-recovery)
  • Byzantine (fail-arbitrary)
• Synchrony (timing) assumptions
  • Synchronous: message latency is limited to a known bound
  • Partially Synchronous: message latency is limited to an unknown bound
  • Asynchronous: no time guarantee for message latency

2.4: Fault tolerance

• SLA (Service Level Agreement) is the contract specifying some SLO (Service Level Objective), penalties for violation
• Eventually perfect failure detector: may mislabel a node as failure or not from time to time, but eventually labels a node as a failure if and only the node is a failure.

3.1: Physical clocks

• Leap second is very common but hard to deal with.

3.2: Clock synchronisation

• NTP may adjust the time of a system in the following ways:
  • Slewing the clock when the time skew is small: adjusting the clock speed to run slightly faster or slower as needed
  • Forcibly sets its clock to the estimated correct time when the time skew is large
  • Refuse to adjust the clock when the time skew is very large
• Clock taxonomy
  • Time-of-day clock
  • Time since a fixed date (e.g. 1 January 1970 epoch)
  • May suddenly move forwards or backwards (NTP stepping), subject to leap second adjustments
  • Monotonic clock
  • Time since arbitrary point (e.g. when machine booted up)
  • Always moves forwards at near-constant rate

3.3: Causality and happens-before

• Happen-before relation encodes potential causality

4.1 Logical time Pass.

4.2: Broadcast ordering

• Including FIFO broadcast, causal broadcast, total order broadcast and so on.

4.3: Broadcast algorithms Pass.

5.1: Replication

• Tombstone: flag to indicate a key has been deleted.
• Multi-value register: when using vector clock, suppose there are two writes, namely (x=a, t=t1) and (x=b, t=t2) for variable v. Both of a and b will be recorded if t1 is parallel to t2 according to their vector clocks. The application can choose the appropriate value using its own logic.

5.2: Quorums

• In a system with n replicas, if a read will fetch data from r replicas (read quorum) and a write will write r replicas (write quorum), read can see previously write iff r+w>n.

5.3: State machine replication Pass

6.1: Consensus

• Consensus and total order delivery are formally equivalent.
• Common consensus algorithms:
  • Paxos: single-value consensus
  • Multi-Paxos: generalisation to total order broadcast
  • Raft, Viewstamped Replication, Zab: FIFO-total order broadcast by default
• Consensus system models:
  • Paxos, Raft, etc. assume a partially synchronous, crash-recovery system model.
  • Cannot be possible in an asynchronous model since there is FLP impossibility: There is no deterministic consensus algorithm that is guaranteed to terminate in an asynchronous crash-stop system mode
  • Blockchains: consensus algorithms for a partially synchronous Byzantine system model

6.2: Raft Haven't finished watching it since it's tedious. We may need to revisit it when we are going to implement Raft.

7.1: Two-phase commit

• Atomic commit is more strict than consensus
  • Atomic commit will be aborted if any participant decides to abort or is crashed.

7.2: Linearizability

• ABD algorithm: make read/write quorum linearizability
  • A client must first send the get request to replicas, and wait for responses from a quorum.
  • Then read repair: If some responses include a more recent value than other responses, as indicated by their timestamps, then the client must write back the most recent value to all replicas that did not already respond with the most recent value.
  • The get operation finishes only after the most recent value is stored on a quorum of replicas: a quorum of replicas either responded OK to the read repair, or replied with the most recent value in the first place.
• A linearizable, replicated CAS operation can be implemented using total order broadcast

7.3: Eventual consistency

• The required synchrony of eventual consistency, causal broadcast, FIFO broadcast are all asynchronous.
• The required synchrony of consensus, total order broadcast, linearizable CAS are all partially synchronous.

8.1 Collaboration and conflict resolution

• Conflict-free Replicated Data Types (CRDTs)
  • Operation-based (since the operation is broadcasted)
  • Process:
    • Use map to store kv.
    • Last-writer-wins: A set value request takes a logical timestamp, and the receiver will update the value iff its map doesn't contain the key or the timestamp of the kv pair is smaller than the one from the request.
  • Requirements:
    • Reliable broadcast ensures every operation is eventually delivered to every (non-crashed) replica
    • Applying an operation is commutative: order of delivery doesn’t matter
  • State-based
  • Process:
    • Update is directly merged into the previous state, ignoring its timestamp. Then best-effort broadcast the whole state.
    • Receiving a whole state, a node will merge it with its own state and keep only the newer kv pairs when conflicts occur.
  • Requirements:
    • The merge should be commutative, associative and idempotent.
  • State-based versus operation-based:
  • Op-based CRDT typically has smaller messages
  • State-based CRDT can tolerate message loss/duplication

8.2: Google's Spanner

• Consistency properties:
  • Serializable transaction isolation
  • Linearizable reads and writes
  • Many shards, each holding a subset of the data; atomic commit of transactions across shards
• Many standard techniques:
  • State machine replication (Paxos) within a shard
  • Two-phase locking for serializability
  • Two-phase commit for cross-shard atomicity
• The interesting bit: read-only transactions require no locks!
  • For example, database backup can be conducted in a transaction
  • Spanner avoids locks on read-only transactions by allowing a transaction to read from a consistent snapshot of the database using MVCC
  • Consistent means causality here: if transaction T1 happened before transaction T2, and if the snapshot contains the effects of T2, then it must also contain the effects of T1.
  • The special point of using MVCC in Spanner is how it assigns timestamps to transactions
  • In order to link the transactions with causal order, logical timestamp is used usually
    • But this requires human users to include a properly formed timestamp on every action they perform
  • Another way is to have a single designated server that assigns timestamps to transactions,
    • But in a globally distributed database, that server would become a single point of failure and a performance bottleneck.
    • Moreover, if transactions executing on a different continent from the timestamp server, the delay would make transactions slow
  • TrueTime API
  • It returns a time interval [t_earliest, t_latest] rather than a single timestamp
  • The transaction will not commit until after having waited for a duration=t_latest-t_earliest, and then label the timestamp t_latest to the transaction
  • The waiting ensures that the timestamp intervals of T1 and T2 do not overlap, even if the transactions are executed on different nodes with no communication between the two transactions
  • Google installs atomic clocks and GPS receivers in every datacenter, and synchronises every node’s quartz clock with a time server in the local datacenter every 30 seconds.
  • In the local datacenter, round-trips are usually below 1 ms, so the clock error introduced by network latency is quite small. If the network latency increases, e.g. due to congestion, TrueTime’s uncertainty interval grows accordingly to account for the increased error.