← 博客

Distributed Systems

RPC Basic

protobuf IDL

  1. define .proto file，定义Request, Response格式，函数签名
  2. protobuf自动生成CalculatorStub
    1
2
3
4
5
6
7
8

    channel = grpc.insecure_channel("服务端IP:端口")

# 客户端Stub（服务端函数的“本地替身”）
stub = CalculatorStub(channel)

# 调用Stub的Add方法（感觉和调用本地函数一样）
request = AddRequest(num1=2, num2=3)
response = stub.Add(request)
  3. server side实现add的具体逻辑

Semantics

At most once

  • have reply, the server恰好执行了一次
  • have no reply, server执行了一次/零次

At least once

  • have reply, the server至少执行了一次
  • have no reply, server执行了零次/一次/多次
  • 对read-only的operation来说，没问题。或者对幂等的操作也没问题，比如MapReduce中的Map阶段

primary backup

general flow

CAP

  • Consistent views of the data at each node
  • Availability of the data at each node
  • Tolerance to network partitions

BASE

  • Basically Available (基本可用): 系统保证能用，但可能偶尔失败或响应慢一点。

  • Soft state (软状态): 数据状态可能会在没有外部输入的情况下发生变化（因为数据正在节点间同步）。

  • Eventual consistency (最终一致性): 系统不保证你读到的立刻是最新的，但保证“过一会儿”所有节点的数据会达成一致。

正常运行流程

Client 发送：客户端只把请求（如 Put）发给 Primary 。

Primary 定序：Primary 收到请求，决定执行顺序（比如先 A 后 B）。

转发 (Forward)：Primary 把操作转发给 Backup 。

Backup 执行：Backup 收到后，执行操作（或记录日志），确保自己状态跟上了 Primary，然后回复 Primary“我好了” 。

Primary 执行并回复：只有收到 Backup 的确认后，Primary 才能执行并回复 Client“成功”

view server

ViewServer维护当前的View
一个View的组成：

  • ViewNum
  • Primary
  • Backup

所有服务器定期给 ViewServer 发 Ping(ViewNum)。如果 ViewServer 发现 Primary 没动静了，就宣布一个新的 View，提拔 Backup 做新的 Primary

state machine replication

$$\text{相同的初始状态} + \text{相同的输入指令} + \text{相同的执行顺序} = \text{相同的最终状态}$$

故障恢复

  1. view server发现某个server没动静了，于是创建新的view. 并且在之后的ViewReply中把这个新 View 告诉它们
  2. 当server收到ViewReply，发现自己是Primary，并且backup节点有更新的时候，它会做这些
    • State Transfer: 把内存中所有k-v打包，发RPC给新的backup
    • Wait for Ack: 在收到新的backup的回复之前，Primary不能处理client的任何请求
    • 只有在收到新backup的回复之后，Primary才会在Ping中使用刚才从view server处收到的viewNum,代表现在的primary/backup结构已经完全和view server刚才指定的一致了
    • state transfer: 只有backup更新时，才需要做。传整个AMOApplication
  3. 假设这一切的背景是：client 给Primary A 发请求，却timeout了，于是client找到view server, view server返回一个完整的view. 那么client 就去请求新的Primary

rule 1: forward before execute

  • primary收到request之后，先转发给backup,让它执行
  • 自己再执行一次，然后发送ack给client
  • primary的转发操作一定要设一个timer

如果不这么做呢？ primary自己执行了，发送ack给client. 但是primary forward在半路被drop,这样primary down之后，backup也没有了之前的操作

Rule 2: split brain

split_brain

Rule 2 (Backup Check): Backup must accept forwarded requests only if view is correct

  • server B 其实没有down。但是view server已经创建了新的view，并且选举C为primary
  • client仍然以为B是primary，给B发送一个put request. B收到之后，把请求转发给C（B仍然以为C是backup）。
  • C收到B转发的request时，发现viewNum=2, 与它自己的viewNum=3不一致，所以不执行请求
  • B timeout, client timeout, client找viewServer 要最新的view

Rule 3: new primary must have been primary or backup

  • 一开始，primary A, backup B，执行put操作
  • A和B都down了，primary C, backup D。执行get操作，无法读到数据
  • 系统刚刚启动，处于 View 0 的状态（没有primary/backup),此时如果一旦有一个server给view server发了ping, view server应该立即形成view 1,把primary分给那个server

rule 4: non-primary reject request

backup rej

rule 4: non-primary must reject client requests

  • A 之前timeout了，但后来又恢复了
  • client 联系A，发送viewNum=1过去
  • A知道现在自己的viewNum=3，因此不接受request

details:

  • 一个server在接收到request的时候，需要检查自己是不是primary。如果不是，拒绝
  • 首先检查自己的ID和当前view的primary ID一致
  • 然后检查request的viewNum和自己的viewNum是否一致
    • 如果client viewNum<自己viewNum: 常见情况，自己已经不是primary了，client仍然以为自己是primary
    • 如果client viewNum>自己viewNum: server无法知道自己到底是不是primary

rule 5: primary fail during state transfer

rule 5: View server cannot change views until current primary has acked the view.

  • A 被选为新的primary, view server给它发viewNum=2
  • A 还没有完成state transfer给B，就down了，所以A没有给view server 发送ping(viewNum=2)
  • viewServer 绝不能 开启 View 3（即选B为新的primary)

这里可以这样做： 在A收到backup的ACK之前，给view server发送的oing都是viewNum=1

另一条需要implement的：

We need to guarantee that state on backup is only overwritten once per view change

Backup 必须记录自己是否已经完成了当前 View 的状态初始化。一旦完成，就必须拒绝或忽略后续任何针对该 View 的全量覆盖请求。

因为stateTransfer()的消息有可能delay，会触发client重传，就有可能导致backup多次收到同样的stateTransfer()。我们必须只执行一次。

stateTransfer(AMOApplication)一次传现在所有的状态，backup server那边记录一下上次完成state transfer的view number，和这次收到的stateTransfer的viewNumber进行比较，再决定是否执行transfer

rule 6: atomic state transfer

atomic transfer

rule 6: Every operation must be before or after state transfer

  • B 是primary，需要把state同步给C.
  • 还没同步完的时候，client给B发request，B需要让C先执行完，自己再执行
  • 但是C先执行request呢？还是先完成state transfer呢？
  • sol: Primary should ignore client requests while in the middle of a state transfer. B先等到C同步完的ack, 再接收client’s request

Paxos

log order

  • log里面有许多slots, servers需要决定每个slot里面放什么operation
  • 每个server发表proposal,关于这个slot i放哪个operation
  • 得票数多的proposal被安排进slot i
  • 没竞争到的operation尝试对slot i+1发起proposal

terms

  • value: operation to be put in the slots
  • proposal: Tuple[Integer, Value]
  • Accept: say yes to a proposal

roles

each server plays all three roles

  • proposers
  • acceptors
  • learners

phase 1: prepare

  1. proposer选择一个全局唯一、递增的proposal number(8)
  2. 给acceptors发送。
  3. acceptor回复：承诺以后不再接受<8的proposal；并回复之前接受过的proposal([2,x],[4,y])

proposer只需要等待多数回复即可

phase 2: accept

  1. 如果刚才acceptor的回复中，已经有人接受proposal, 则proposer必须选择proposal number最大的那个value; 如果没有，proposer可以提任意自己想要的operation。比如这里会选择y
  2. proposer发送accept(p, operation). 比如这里发送(8, y)
  3. 若一个acceptor没有接受过比p更大的proposal,则accept。
  4. 若这个proposal被majority acceptor接受，则确定下来(is chosen)

Liveness

  • Some proposed value is eventually chosen： 最终每个slot总会确定下来一个proposal. 所以acceptor需要能够接受多个proposal,最终选定proposal number最大的value
  • If a value is chosen, a process eventually learns it

safety

  • Only a value that has been proposed can be chosen
  • Only a single value is chosen
  • A process never learns that a value has been chosen unless it has been chosen

propositions

  • p1: An acceptor must accept the first proposal that it receives
  • p2a: if a proposal with value v is chosen, then every higher-numbered proposal accepted by any acceptor has value v
  • p2b: 如果某个值 $v$ 已经被选定（Chosen），那么之后任何 Proposer 提出的更高编号的提案，其值必须也是 $v$
  • p2c: accept阶段，如果一个 Proposer 想要发出编号为 n 的提案，它必须先查看是否有编号小于 n 的提案已经被majority接受。如果有，它必须提议那个编号最大的既定值；如果没有，它可以提议任何值

only want to implement p2c in practice

learn phase

  • once a avlue is chosen, acceptors inform a distinguished learner, who informs the other learners
  • learner leader is the proposer who proposed the chosen proposal

Multi Paxos

alt text

  • Ballot: [Integer, ServerID]

  • pvalue: <Ballot, Slot, Command>

  • election phase: leader is elected and its ballot is adopted by a majority of acceptors

    • acceptor state:
      • its adopted ballot. it promises not to accept pvalues that have smaller ballots
      • a of pvalues that it has accepted
    • p1a: leader send a ballot to the acceptors
    • p1b: acceptor reply with all accecpted pvalues it has, and its ballot

  • proposal phase: leader proproses pvalues. if a major of acceptors accept a pvalue, then the slot is decided

    • p2a: leader send pvalue to the acceptor
      • acceptor will compare the ballot number, if match, than acceptor’s that slot is accepted
    • p2b: ballot+slot
      • when leader receives p2b, it will compare the ballot, if match, mark that slot as accepted by that acceptor

leader

  • election begin upon

    • system initialization
    • current leader timeout due to lack of heartbeat

  • state:

    • active: true/false
    • ballot_num: 0.0
    • proposals: []
    • seen: <1.2, 1, B> (pvalue it has got from p2b)

  • election

    • send p1a to acceptors
      • if majority accpected, set active to true
      • if not, choose a higher ballot value and send again
    • send p2a to acceptors
      • if leader’s p2a get rejected, it realizes it’s not the leader anymore. set active to False, empty the proposal list
      • if get accepted by majority, tell the replicas

  • if get preempted, don’t retry immediately [TODO:need specific design]

  • heartbeat

    • ballot, chosen{slot->command}, first_non_cleared
    • leader sends heartbeat to acceptors, if the ballot matches, acceptor will also mark those slots in chosen to accepted
    • if acceptor’s ballot is lower, it can use the heartbeat to catch up things

  • when heartbeat timeout

    • other servers will compete to be the new leader. every one sends out ballot to others. if a server receives a higher ballot, it will treat itself as a follower, and treat the message as a heartbeat from the new leader

log

  • slot_out

    • next undeicied slot
    • 在 slot_out 之前的所有 Slot（1 到 slot_out - 1）都已经被chosen，并且已经executed
    • increment your slot out when something is executed

  • slot_in

    • next empty slot
    • 指向下一个空白的、可以用来存放新命令的 Log Slot
    • 何时增加？只有当你作为 Active Leader 收到来自Client的新Command时

  • log merging

    • Keep the command with the highest ballot number
    • If we’ve received the same command with the same ballot number from a majority of the nodes, mark the command as chosen
      • Do not need to do this when receiving P1bs
      • Even if the potential leader does not mark slots chosen, if the node becomes leader, it will find out that the slot has been chosen when it later sends out P2as and receives P2bs
    • If no command is found for a slot, put a No-Op command there instead

  • Shadow Log

    • During Phase 1, do not merge P1B responses directly into the main log.
    • Use a temporary data structure (e.g., Map<Slot, Pvalue>) to store the best proposals found so far.
    • Commit Point: Only overwrite the actual log with the temporary map after election is won (Majority achieved).
    • Abort: If election fails or we get preempted, discard the temporary map

  • Log Holes

    • Previous slot is empty, but a later slot has a value.
    • Leader proposes a No-Op command for the empty slot.
    • Once the No-Op is chosen, the application executes it (does nothing), increments slot_out

Garbage Collection

  • Mechanism: Use Heartbeat Replies.
  • Followers send their slot_out (index of next command to execute) back to the Leader in heartbeat replies.
  • Leader calculates min(all_slot_outs).
  • Leader tells everyone: “It is safe to clear logs up to min_slot_out - 1”.
  • Leader 在发送给 Follower 的心跳包（Heartbeat）中，增加一个字段，叫 first_non_cleared

Memory Consistency Models

Consistency

Consistency is a formal contract between you as the programmer and the system builder that defines the allowed semantics and expected return values for operations on a data store.
It strictly specifies the interface of what a system is allowed to do rather than how it is implemented, meaning the underlying data could be cached, replicated, or sharded without changing the rules.
If the system ever returns a value that violates this agreement, the breach of contract is known as an anomaly.

  • Strong Consistency: The system behaves perfectly, as if there is only a single, centralized data store.

  • Weak Consistency: A broad category for models that allow behaviors violating the “single store” illusion (meaning you might sometimes read outdated data).

  • Eventual Consistency: A specific type of weak consistency. Anomalies are allowed, but they are strictly temporary. If operations stop, all users will eventually see the same data.

Lamport’s Register Semantics

  • Safe: Read alone = get old value. Read during write = get anything (even trash).

  • Regular: Read during write = get either old or new value (no trash).

  • Atomic: Regular + No going back. If one reader sees the new value, all later readers must see it too.

Linearizibility

  • 不管后台有多少个服务器在同步数据，对客户端来说，系统表现得就像只有一台单机一样。“我刚写完，你立刻读，一定能读到”
  • if A finishes before B starts, A must happen first in the system’s history.
  • 全序顺序 (Single total order)：系统里发生的所有操作，最终都能排成一个单一的、线性的先后顺序。
  • no stale read(lab3)

Sequential Consistency

  • all operations follow a single total order that respects each node’s local order
  • unlike linearizability, it does not need to respect real-time. it’s okay if operations appear to happen “late” as long as the order is consistent for everyone
  • allow read operations to return stale data(lab2)
  • in database design, it’s called serializability

Causal consistency

causal

  • no global order
  • 维护 happens-before DAG
  • Concurrent writes，也就是互不影响的write, 不同的观察者可以看到不同的顺序

not_causal

  • 因为p1是看到了b之后，再write a的
  • 但p3先看到果，再看到因

Processor Consistency / Read-your-writes

  • Writes done by the same process are seen in that order。如果一个进程先写A，再写B，那么所有其它进程必须先读A再读B
  • Writes by different processes can be seen in different orders by different readers
  • If a process writes a value, a subsequent read of the same value by that process will see the write

Eventual Consistency

  • if all writes stop, eventually all processes read the same value.
  • DNS

Sharding

  • Chunk key-value pairs into shards
  • Dynamically assign shards to Paxos groups
    • Need state transfer to move key-value shards between Paxos groups
    • Want to minimize movement if possible, while maintaining load balance
    • Avoid deadlock when moving shards
  • Clients maintain table of which shards are assigned to which Paxos groups
  • Learn assignment from shard master

consistent hashing

把服务器和 key 都放到一个“环”上。如果新增一个服务器，平均只需要移动K/N个 key

1️⃣ 把 hash 空间看成一个 0 → 2^32 的环

2️⃣ 把每台服务器 hash 到环上一个位置

3️⃣ 把 key 也 hash 到环上

4️⃣ key 属于顺时针方向遇到的第一个服务器

Virtual Nodes

  • 每个物理机器在环上放多个“虚拟节点”，每个节点的坐标是hash(serverID + index)

  • key在环上顺时针找到第一个节点

vnode

vnode越多，key在不同物理机器上的分布越均匀

table indirection

  1. hash(key)=bucket
  2. server = table[bucket]

每个client都有同一张table.
shard master根据负载情况, 调整 table 里的 bucket 分配, 然后broadcast新表

解决了hot key问题 indirection 比如这里bad bunny是个hot key，那么就让这个server1专门服务它的bucket

goals achieved

  • clients all have same assignment
  • workload evenly distributed
  • add/remove node moves only a few keys
  • Even with hot keys
  • divide work of redistributing keys

Cache

Linearizability

Linearizable iff:

  • All operations from one client occur in order
  • All clients see the same total order of operations
  • All operations happen between the client request’s start/end

lease + callback

  • Server tracks which clients cache an object and the lease expiration time for each cached copy.
  • Before processing a write, the server sends callbacks to invalidate all cached copies.
  • If a client does not respond, the server waits until that client’s lease expires before completing the write.

Sequential Consistency

Sequential Consistency iff:

  • All operations from one client occur in order
  • All clients see the same total order of operations

allow

  • Buffer writes at cache: 客户端把write操作更新到cache
  • Invalidate asynchronously: 缓存接收了写入后，会在后台（异步地）去发送通知，让其他节点作废旧数据
  • Allow read-only cache copies to serve stale reads while invalidation proceeds: 在这个invalidation的操作过程中，如果有其它client向其它copy请求同一份数据，允许返回stale data

write strategy

  • write through
    • cache只保存read only data
    • write必须直接发送到后端的中央存储, 中央存储去revoke cache

Eventual Consistency

TTL

lab 4

Implementation

ShardMaster

ShardMaster is an Application that is running within a PaxosServer

1
2
3
4
5

private final int configNum;`
private final Map<Integer, Pair<Set<Address>, Set<Integer>>> groupInfo;
    Integer: group id
    Set<Address>: addresses of all members in that group
    Set<Integer>: all the shard numbers the group is responsible for
  • Keeps track of a current configuration object (ShardConfig)
  • remembers all old configurations
  • Accepts 4 commands:
    • Join
      • Integer for replica group ID + Set of server addresses that should be in the group
      • create a new config; redistributes the shards; return ok
    • Leave
      • contains: Group Id that should “leave”
      • OK on success, ERROR when the current config does not contain group OR the final group is trying to leave (not actually tested)
    • Move
      • contains: Shard number + Replica Group id
    • Query
      • contains: configuration number
      • reply with ShardConfig
  • Respond
    • OK
    • Error
    • ShardConfig