info

intro

drivens

性能
容错
物理分布
安全隔离

challenges

并发
部分错误
性能调优

framework

存储
通信（网络）
计算

abstraction & implementation

我们希望构建一个系统，它看起来就像一个非分布式存储和计算系统，但是实际上又是一个有极高的性能和容错性的分布式系统

一些实现工具

RPC
线程
并发控制

CAP

Consistency（一致性）
Availability（可用性）
Partition tolerance（分区容忍性）

map reduce

https://pdos.csail.mit.edu/6.824/papers/mapreduce.pdf

https://zhuanlan.zhihu.com/p/122571315

以单词计数为例，基本的示意图如下

一些术语

整个 Map Reduce 计算称为 Job
每一次 Map Reduce 调用称为 Task

优化

Master 节点可以找出输入文件具体存在哪台 GFS 服务器上，并把对应于那个输入文件的 Map Task 调度到同一台服务器上
这样可以避免网络传输带来的瓶颈

crawler

https://go.dev/tour/concurrency/10

mutex

func ConcurrentMutex(url string, fetcher Fetcher, f *fetchState) {
  f.mu.Lock()
  already := f.fetched[url]
  f.fetched[url] = true
  f.mu.Unlock()

  if already {
    return
  }

  urls, err := fetcher.Fetch(url)
  if err != nil {
    return
  }
  var done sync.WaitGroup
  for _, u := range urls {
    done.Add(1)
    go func(u string) {
      defer done.Done()
      ConcurrentMutex(u, fetcher, f)
    }(u)
  }
  done.Wait()
  return
}

使用 lock 保护临界区

使用条件变量同步各个线程，从而避免主线程立即 exit

channel

func worker(url string, ch chan []string, fetcher Fetcher) {
  urls, err := fetcher.Fetch(url)
  if err != nil {
    ch <- []string{}
  } else {
    ch <- urls
  }
}

func coordinator(ch chan []string, fetcher Fetcher) {
  n := 1
  fetched := make(map[string]bool)
  for urls := range ch {
    for _, u := range urls {
      if fetched[u] == false {
        fetched[u] = true
        n += 1
        go worker(u, ch, fetcher)
      }
    }
    n -= 1
    if n == 0 {
      break
    }
  }
}

func ConcurrentChannel(url string, fetcher Fetcher) {
  ch := make(chan []string)
  go func() {
    ch <- []string{url}
  }()
  coordinator(ch, fetcher)
}

使用 channel 进行同步

coordinator 即为 master 节点

注意这里的对 channel 使用了 range，并在 master 节点内部使用一个变量记录 worker 的数量，当 worker 变为 0 时，即退出 range，否则将会陷入死循环

由于这里 master 节点和所有的 worker 节点共用一个 channel，所以不能简单的 close channel

可以参考 https://imil.net/blog/posts/2019/understanding-golang-channel-range-again/

rpc

client-server 模型

server 注册 KV，其中包含了数据和行为

func server() {
  kv := new(KV)
  kv.data = map[string]string{}
  rpcs := rpc.NewServer()
  rpcs.Register(kv)
  l, e := net.Listen("tcp", ":1234")
  if e != nil {
    log.Fatal("listen error:", e)
  }
  go func() {
    for {
      conn, err := l.Accept()
      if err == nil {
        go rpcs.ServeConn(conn)
      } else {
        break
      }
    }
    l.Close()
  }()
}

func (kv *KV) Get(args *GetArgs, reply *GetReply) error {
  kv.mu.Lock()
  defer kv.mu.Unlock()

  reply.Value = kv.data[args.Key]

  return nil
}

行为中定义了一系列接口，供 client 使用

server 只需要获取 connection 并按照接口调用即可

func connect() *rpc.Client {
  client, err := rpc.Dial("tcp", ":1234")
  if err != nil {
    log.Fatal("dialing:", err)
  }
  return client
}

func get(key string) string {
  client := connect()
  args := GetArgs{"subject"}
  reply := GetReply{}
  err := client.Call("KV.Get", &args, &reply)
  if err != nil {
    log.Fatal("error:", err)
  }
  client.Close()
  return reply.Value
}

实际的实现中涉及到桩和序列化等技术

client 调用 client stub，并将调用参数 push 到栈中，这个调用是在本地的
client stub 将这些参数包装，并通过系统调用发送到服务端机器，打包的过程叫 marshalling
client 操作系统将消息传给传输层，传输层发送信息至服务端
服务端的传输层将消息传递给 server stub
server stub 解析信息，该过程叫 unmarshalling
server stub 调用程序，并通过类似的方式返回给 client
client 拿到数据解析后，将执行结果返回给调用者

RPC 可以实现三种语义

At-Most-Once

client 不会自动重试一个请求

At-Least-Once

client 会不断重试请求，直到收到请求被执行的肯定确认

例如 TCP 的超时重传

适用于幂等操作

Exactly-Once

在这种模式下，请求既不能重复，也不能丢失

Go RPC 实现了 At-Most-Once 语义

distributed storage system

困难与妥协

追求性能 -> 分片
部分错误 -> 容错
容错 -> 复制
复制 -> 不一致
不一致 -> 低性能

并不乐意为好的一致性付出相应的性能代价

gfs

https://pdos.csail.mit.edu/6.824/papers/gfs.pdf

master node

两个映射关系

filename -> array of chunk handles
chunk handle -> array of chunk servers
1. 版本号 version
2. 哪个 chunk server 上持有 primary chunk，只有 primary chunk 才能进行写操作
3. primary chunk 的租约过期时间 lease expiration

以上数据都存储在内存中，有些数据还应同步数据存储在磁盘，例如

array of chunk handles
version

由于 master node 重启后需要和所有 chunk servers 通信，所以相关的信息不必持久化存储

另外，master node 还会定期将所有数据在磁盘上生成 checkpoint，并为所有的写操作进行记录 log

read file

流程如下

client 发送 filename 和 offset 给 master node
master node 通过映射关系得到 array of chunk servers，并发送给 client
client 同其中一个 chunk server 通信

client 会缓存 chunk 和 server 的映射关系

write file

流程如下

client 向 master node 请求存储了某个文件最后一个 chunk 的 chunk server
1. 因为当有多个 clients 同时写同一个文件时，client 并不能知道文件究竟有多长
master node 若发现该 chunk 的 primary server 不存在时，会找出所有存有该 chunk 最新副本的 chunk servers
1. 这通过 version 来完成
2. 即副本中保存的 version 与 master node 中记录的 chunk 的 version 一致
3. 有可能找不到存有该 chunk 最新副本的 chunk server，那么 master node 可以向 client 返回请稍后重试
master node 从这些 chunk servers 中选取一个作为 primary server，其余作为 secondary servers
master node 增加 version
1. 这只会发生在指定一个新的 primary server 时
master node 向这些 chunk servers 发送消息，包括谁是 primary，谁是 secondary，最新的 version
1. 还会给 primary server 的租约过期时间 lease expiration
2. 这种机制可以确保不会同时有两个 primary server
3. 若 master node 发现无法与 primary server 通信，就开始选取新的 primary server，很有可能旧的 primary server 还可以与 client 通信，最终就会有两个 primary server 处理写请求，导致有两个不同的数据拷贝，这被称为脑裂 split-brain
4. 使用 lease expiration 机制，primary server 只在租约内有效
5. 即使在租约内 primary server 真的挂了，master node 也只会默默等待
master node 将 version 写入磁盘
1. 有可能 master node 向 chunk servers 发送完消息之后就崩溃了
2. 重启时可能发现 chunk servers 上报的 version 高于 master node 存储的 version
3. 此时使用更高的 version 作为 chunk 的最新 version，master node 就可以从崩溃中自动恢复了
4. 所以将 version 写入磁盘在发送消息之后
master node 向 client 返回谁是 primary，谁是 secondary
client 将要追加的数据发送给 primary 和 secondary servers
1. 这些数据不会追加到文件中，而是写入到一个临时位置
当所有的 servers 都返回了确认消息，client 向 primary server 发送消息确认追加数据
1. primary server 追加数据，并通知所有的 secondary servers 追加数据，成功则回复 primary server
当 primary server 成功追加数据，并收到了所有的 secondary servers 的回复，则向 client 返回写入成功，否则返回写入失败
1. 写入失败时，一个 chunk 的部分副本成功完成了数据追加
2. 注意此时 primary 和 secondary servers 拥有相同的 version
3. client 应重新发起追加数据的请求
4. 会导致数据乱序，但这种状态是可接受的
5. 所以 gfs 并不是强一致的系统

vm-ft

https://pdos.csail.mit.edu/6.824/papers/vm-ft.pdf

复制 replication 能够处理单台计算机的 fail-stop 故障

当任何地方出现故障时，就停止运行，而不是运算出错误结果

这篇论文介绍了两种复制的方法

状态转移 State Transfer
- 传输 memory
复制状态机 Replicated State Machine
- 传输来自 client 的 operation 或者其他外部事件
这里主要介绍后者，在机器级别复制状态机

basic

下面介绍 VMware FT 的工作原理

VMware FT 需要两个物理服务器，每个物理服务器都运行 VMM (Virtual Machine Monitor)

Primary 虚拟机在其中一个物理服务器上，Backup 虚拟机在另一个物理服务器上

在初始状态，两个虚拟机具有完全相同的内存镜像

client 向 Primary 虚拟机发送了一个请求，这个请求以网络数据包的形式发出

这个网络数据包产生一个中断，之后这个中断送到了 Primary 虚拟机的 VMM

此时 Backup 虚拟机对 client 不可见

Primary 虚拟机的 VMM 会做两件事情

VMM 模拟网络数据包到达的中断，以将相应的数据送给 Primary 虚拟机

Primary 虚拟机里的服务会生成一个回复报文，并通过 VMM 模拟的虚拟网卡发出，之后 VMM 可以看到这个报文，它会实际的将这个报文发送给 client

VMM 将网络数据包拷贝一份，并通过网络送给 Backup 虚拟机所在的 VMM

Backup 虚拟机所在的 VMM 也会模拟网络数据包到达的中断，并生成一个回复报文，但是它的 VMM 知道这是 Backup 虚拟机，于是会丢弃这里的回复报文

下面是一些术语

将 Primary 到 Backup 之间同步的数据流的通道称之为 Log Channel
从 Primary 发往 Backup 的事件被称为 Log Entry

当 Primary 虚拟机挂了后，Backup 虚拟机会接手，并宣称它有 Primary 的 MAC 地址

以 Ethernet 为例，此时 Backup 虚拟机对 client 可见

sync

然而，想让 Primary 和 Backup 之间保持同步，有如下困难

client 请求的网络数据包产生中断的时机
weird instructions，如随机数生成器或获取当前的时间
多核 CPU 指令的并行，论文中完全没有讨论这一点

为此，需要在 Log Entry 中提供一些信息以供同步

事件发生时的指令序号
1. 自机器启动以来指令的相对序号
2. 这里依赖硬件的定制，让 Backup 虚拟机执行到特定的位置就停止，以等待下一个 Log Entry
类型是 network packet 还是 weird instructions
数据
1. 对于 network packet 而言就是 client 发送的数据
2. 对于 weird instructions 而言是 Primary 虚拟机执行指令的结果，Backup 虚拟机需要伪造指令，并提供与 Primary 相同的结果

output

还需要仔细考虑对于 client 请求的响应，考虑如下情形

Primary 虚拟机生成一个回复报文后崩溃了
但是 Log Entry 并未发送给 Backup 虚拟机
这样，当 Backup 虚拟机上线后，其状态和 Primary 虚拟机是不一致的

所以需要在 Log Entry 发送给 Backup 虚拟机，并收到 Backup 虚拟机的 ACK 后，Primary 虚拟机才能将报文发送给 client

另一种情形是

Primary 虚拟机将报文发送给 client 后崩溃了
此时 Backup 虚拟机上线，Backup 会执行 buffer 中的 Log Entry，并发送相同的报文发送给 client

然而，这个报文的 TCP 序列号与之前生成报文的 TCP 序列号是一样的，这样 client 的 TCP stack 会发现这是一个重复的报文，并丢弃它

在其他情形中，重复输出是难以避免的

test-and-set

最后需要考虑的情形是

Primary 和 Backup 之间的通信断了
这时，它们都会以为对方挂了，自己需要上线并接管服务

这便出现了 Split Brain 现象

为此，副本在上线前需要获得一个外部服务 Test-and-Set 的锁

可以发现主从切换依赖于 Test-and-Set 服务在线，所以实际上 Test-and-Set 服务也应该进行 replication，否则这就是个单点故障

raft

https://pdos.csail.mit.edu/6.824/papers/raft-extended.pdf

之前介绍的 fault-tolerant 系统存在一个共性，它们都是多副本系统，并且需要一个单节点来决定，在多个副本中，谁是 primary，这就容易引入单点故障，在分区的情形下，需要小心的设计策略来避免 Split Brain

在构建能自动恢复，同时又避免 Split Brain 的多副本系统时，人们发现，关键点在于过半票决 Majority Vote，其要点如下

server 的数量是奇数
必须凑够过半的服务器来批准完成任何的操作
1. 如果网络存在分区，那么必然不可能有超过一个分区拥有过半数量的服务器
2. 这里的过半是所有服务器数量的一半，而不是当前可用的服务器数量的一半

Raft 的核心便是 Majority Vote

basic

Raft 会 Library 的形式存在于服务中

对于一个 KV 数据库而言，对应的状态就是 Key-Value Table，应用程序往下，就是 Raft 层，所以，KV 数据库需要对 Raft 层进行函数调用，来传递自己的状态并接收 Raft 反馈的信息

Raft 本身也会保持状态，最重要的就是 Raft 会记录操作的 log

下面考虑宏观上的流程

client 对 leader 发起请求
应用程序将来自 client 的请求对应的操作向下发送到 Raft 层
Raft 节点之间相互交互，直到过半的 Raft 节点将这个新的操作加入到它们的日志中，并通知 leader 的 Raft 层
1. leader 的 Raft 层发送 AppendEntries 给其他 followers，并等待其响应
leader 的 Raft 层会向上发送一个通知到应用程序，可以真正的执行这个操作了
1. leader 的 Raft 层会在下一次 AppendEntries 时顺带发送 committed 给其他 followers，可以真正的执行这个操作了

log

log 是实现复制状态机的重要机制

leader 为操作定序
followers 用来存放临时操作，还未收到 leader 的 committed 信息
leader 存放所有的操作，因为这些操作可能需要重传给 followers
持久化存储操作，便于快速恢复

leader election

下面考虑 leader 的选举策略

使用 term number 来区分不同的任期和 leader

每一个任期最多有一个 leader，可能没有 leader，即没有收到过半的 vote

每个 Raft 节点都有一个选举定时器 Election Timer，如果在 Timer 时间耗尽之前，当前节点没有收到任何当前 leader 的消息，这个节点会认为 leader 已经下线，并开始一次选举

开始一次选举就会增加 term number，之后当前的节点发出请求投票 RequestVote 的 RPC 给其余节点

注意，网络分区时，旧的 leader 可能还在小的分区运行，但是它发出的 AppendEntries 将无法凑够过半 followers 的回复，所以无法完成任何操作

考虑到多个节点可能同时开始选举，并 Split Vote，因为当前的节点一定会投票给自己，这样这些选举都会失败

为此，需要为 Timer 随机的选择超时时间，有如下限制

不同节点的 Timer 的超时时间差能够完成一轮选举
超时时间的下限应大于 leader 的 heartbeat 间隔
超时时间的上限决定了系统能多快从故障中恢复

log sync

下面考虑 log 的同步问题

考虑如下情形

	10	11	12
server 1	3
server 2	3	3	4
server 3	3	3	5

10 / 11 / 12 代表 log slot

其中的数值 x 代表 term x 内记录的 operation

这种情形是由于两次 leader 在发送 AppendEntries 前就崩溃了，导致只有 leader 在 log 中有记录

下一个 term number 为 6，假定选取 server 3 作为 leader

	10	11	12	13
server 1	3
server 2	3	3	4
server 3	3	3	5	6

此时 leader 发送 AppendEntries 给 followers，信息包括

prevLogIndex 为 12，即 leader 上一个 log slot 的索引
prevLogTerm 为 5，即 leader 上一个 log slot 的 term number
在当前 log slot #13 的 term number 为 6

然而 followers 发现 AppendEntries 的信息与自己的不匹配，于是均拒绝

同时，leader 为每个 follower 维护了 nextIndex，leader 之前发送的是有关 slot #13 的 log，这意味着 leader 对于其他两个服务器的 nextIndex 都是 13

为了响应 followers 返回的拒绝，leader 会减小对应的 nextIndex，然后发送的 AppendEntries 消息

prevLogIndex 为 11
prevLogTerm 为 3
[(#12, 5), (#13, 6)]，即从 nextIndex 开始的所有 log 信息

此时 server 2 匹配，更新自己的 log 信息，并回复 leader

	10	11	12	13
server 1	3
server 2	3	3	5	6
server 3	3	3	5	6

server 1 仍拒绝，再次减小 server 1 的 nextIndex，并发送 AppendEntries 消息

此时 server 1 匹配，更新自己的 log 信息

	10	11	12	13
server 1	3	3	5	6
server 2	3	3	5	6
server 3	3	3	5	6

于是副本之间的 log 就保持了同步

election restriction

上文假定选取 server 3 作为 leader

实际上并非任意节点都可以成为 leader

考虑如下情形

	10	11	12
server 1	5	6	7
server 2	5	8
server 3	5	8

这种情形是由于 server 1 连续两次成为 leader，但是在发送 AppendEntries 前就崩溃了，之后 server 2 和 server 3 之间达成了同步

此时就不能选举 server 1 作为 leader，因为 server 2 和 server 3 之间已经组成了过半服务器

论文中给出了选举的限制

候选人最后一条 log 信息的 term number 大于投票者最后一条 log 信息的 term number
或者，候选人最后一条 log 信息的 term number 等于投票者最后一条 log 信息的 term number，且候选人的 log 记录长度大于等于投票者 log 记录的长度

根据规则一，显然 server 1 最后一条 log 信息的 term number 为 7，小于 server 2/3 最后一条 log 信息的 term number 8，所以 server 2/3 都不会投票给 server 1

根据规则二，显然 server 2/3 之间可以互投

fast backup

当 log 存在冲突时，leader 会逐渐减小对应的 nextIndex，在某些情景中，这是不够快速的

可以让 followers 在回复 leader 的 AppendEntries 拒绝消息中，携带一些额外的信息，来加速日志的恢复

XTerm
- 这个是 follower 中与 leader 冲突的 log 对应的 term number
- 如果 follower 在对应位置没有 log，那么 XTerm 会返回 -1
XIndex
- 这个是 follower 中，对应 term number 为 XTerm 的第一条 log entry 的 slot number
- 配合 XTerm 返回 term number 使用
XLen
- XLen 表示空白的 log slot 数
- 配合 XTerm 返回 -1 使用

考虑如下场景，server 2 将要发送一条 term number 为 6 的 AppendEntries 消息给 follower，nextIndex 为 4

场景一

	1	2	3	4
server 1	4	5	5
server 2	4	6	6	6

server 1 返回 XTerm 为 5，XIndex 为 2

leader 发现自己没有 term number 为 5 的 log

于是 leader 设置 nextIndex 为 2

场景二

	1	2	3	4
server 1	4	4	4
server 2	4	6	6	6

server 1 返回 XTerm 为 4，XIndex 为 1

leader 发现自己有 term number 为 4 的 log

于是 leader 设置 nextIndex 为自己在 XTerm 位置的 log entry 后面，也就是 slot #2

场景三

	1	2	3	4
server 1	4
server 2	4	6	6	6

server 1 返回 XTerm 为 -1，XLen 为 2

于是 leader 设置 nextIndex 为 2

persistence

以下三个字段需要被持久化

log
currentTerm
votedFor

重启之后，应用程序可以通过重复执行每一条 log 来完全从头构建自己的状态

这是一种简单且优雅的方法，但是会很慢，并且大量的 log 会占据空间，于是需要引入日志压缩这个概念

log snapshot

注意到，对于大多数的应用程序而言，state 的大小远小于 log 的大小

在某些时间点，log 和应用程序的 state 是可以互换的，它们是用来表示应用程序状态的不同事物

例如，有如下的 log

x = 1
x = 2
y = 7

对应的 state 为

x = 2
y = 7

所以，当 Raft 认为它的 log 过于庞大时，Raft 会要求应用程序在 log 的特定位置，对其状态做一个快照，然后丢弃所有那个点之前的 log 记录

快照的创建依赖于应用程序

当重启时，应用程序可以根据之前创建的快照重建自己的状态，然后执行这个快照对应 log 之后的所有命令

然而，这引入了大量的复杂性，例如，当 follower 的 log 在 leader 的快照之前就结束，那么 leader 就无法通过 AppendEntries 同步 follower 的 log

一种解决方案是，如果 leader 发现有任何一个 follower 的 log 落后于 leader 要做快照的点，那么 leader 就不丢弃快照之前的 log，然而，如果 follower 离线时间太长，leader 就无法进行创建快照从而丢弃日志

论文中提出了另一种解决方案，引入 InstallSnapshot 这个 rpc，如果 follower 的 log 短于 leader 通过 AppendEntries 发给它的 log，leader 会通过 InstallSnapshot 将自己 log 开头对应的快照发给 follower，之后立即通过 AppendEntries 将后面的 log 发给 follower

linearizability

一个系统的执行历史是一系列的 client 请求，或许这是来自多个 client 的多个请求。如果执行历史整体可以按照一个顺序排列，且排列顺序与 client 请求的实际时间相符合，那么它是线性一致的

具体的限制如下

如果一个操作在另一个操作开始前就结束了，那么这个操作必须在执行历史中出现在另一个操作前面
执行历史中，读操作，必须在相应的写操作之后

下面是几个例子

开头和结尾代表 request 和 response 的时刻，其实际的读取和写入可能在其中的任意时间点发生

第一个例子是满足 linearizability 的

可以线性化为

write X = 1
read X = 1
write X = 2
read X = 2

第二个例子是不满足 linearizability 的

因为存在循环

第三个例子是不满足 linearizability 的

这里强调对于整个请求历史记录，只存在一个序列，不允许不同的 client 看见不同的序列

最后一个例子

这里读取 X = 1 或者 X = 2 都满足 linearizability

其返回值取决于应用程序的行为，例如 prefer fresh or stale data

zookeeper

https://pdos.csail.mit.edu/6.824/papers/zookeeper.pdf

consistency guarantees

如果我们有了 $n$ 倍数量的服务器，是否可以为我们带来 $n$ 倍的性能

对于 raft 而言，当我们加入更多的服务器时，leader 几乎可以确定是一个瓶颈，因为 leader 需要处理每一个请求，它需要将每个请求的拷贝发送给每一个 follower

考虑将只读请求发送给 follower，然而 follower 并不能保证数据是最新的，那么就无法构建一个线性一致的系统

linearizability 保证不会提供过时的数据

所以 zookeeper 选择了妥协，放弃线性一致性，即不提供线性一致的读

zookeeper 保证如下的一致性

写请求是线性一致的
对于某个特定的 client 的请求，都会按照 client 指定的顺序来执行

具体而言

对于多个 clients 的一系列写请求
1. 其意义是显然的
2. client 会对它的写请求打上序号
对于某个特定的 client 的一系列读请求
1. 后续的读请求，必须要在不早于当前读请求对应的 log 点执行
2. 考虑不同的 followers 响应一系列读请求
3. 任何时候一个 follower 回复 client 的读请求，都需要带上 zxid
4. client 会记住最大的 zxid，当 client 发出一个请求到一个相同或者不同的 follower 时，它会在它的请求中带上这个最大的 zxid
5. follower 就可以判断当前状态是否不旧于之前读请求的状态
对于某个特定的 client 的一系列读写请求
1. 例如向 leader 发送了一个写请求，之后立即读同一份数据
2. 读请求发送给了某一个 follower，那么 client 需要看到自己刚刚写入的值
3. 实现机制之一便是 zxid
若某个 client 发起写请求，另一个 client 发起读请求，并不能保证数据是最新的
1. 也就是上面提到的，不提供线性一致的读

sync

zookeeper 有一个操作类型是 sync，它本质上就是一个写请求

当想读出 zookeeper 中最新的数据时，就发送一个 sync 请求，它的效果相当于一个写请求

ready file / znode

我们假设有另外一个分布式系统，这个分布式有一个 Master 节点，而 Master 节点在 Zookeeper 中维护了一个配置，这个配置对应了一些 file

现在 Master 更新这个配置，同时有大量的 clients 需要读取相应的配置

问题是，尽管配置被分割成了多个 file，我们还能有原子效果的更新吗

首先我们假设有一些 ready file，如果 ready file 存在，那么允许读这个配置。如果 ready file 不存在，那么说明配置正在更新过程中，我们不应该读取配置

考虑如下的执行顺序

write order	read order
delete(“ready file”)
write file 1
write file 2
create(“ready file”)
	exists(“ready file”)
	read file 1
	read file 2

这是理想情形

考虑另一种情形

write order	read order
	exists(“ready file”)
	read file 1
delete(“ready file”)
write file 1
write file 2
create(“ready file”)
	read file 2

此时就不能保证原子效果的更新了

为此引入 watch 操作，当 ready file 有任何变更时，会立即 notify 相应的 client

那么在 read file 2 前，client 就可以得知数据的不一致

每一个 replica 都会维护自己的 watch table，若当前 replica 崩溃了，client 应该重置所有数据，并与新的 replica 建立连接

api

Raft 不是一个你可以直接交互的独立的服务，你必须要设计你自己的应用程序来与 Raft 库交互

是否存在一些有用的，独立的，通用的系统可以帮助人们构建分布式系统，对于一个通用的服务，API 应该是怎样的

对于这些问题，zookeeper 给出了一种方案

Zookeeper 使用层级化的命名方式

Zookeeper 的 API 某种程度上来说像是一个文件系统，这里的文件和目录都被称为 znodes

每一个 znode 都有一个表示当前版本号的 version，当 znode 有更新时，version 也会随之增加

Zookeeper 中包含了 3 种类型的 znode

Regular znodes
- 这种 znode 一旦创建，就永久存在，除非手动删除
Ephemeral znodes
- 如果 Zookeeper 认为创建它的 client 挂了，它会删除这种类型的 znodes
- 这种类型的 znodes 与 client session 绑定在一起
Sequential znodes
- 当你想要以特定的名字创建一个文件，Zookeeper 实际上创建的文件名是你指定的文件名再加上一个数字
- Zookeeper 会确保这里的数字不重合，同时也会确保这里的数字总是递增的

Zookeeper 以 RPC 的方式暴露以下 API

CREATE(PATH, DATA, FLAG)
- exclusive
DELETE(PATH, VERSION)
- 可以传入一个 version 表明，只有当 znode 版本匹配时才删除
EXISTS(PATH, WATCH)
- 判断文件是否存在和 watch 文件的变化，在 Zookeeper 内是原子操作
GETDATA(PATH, WATCH)
- 这里的 watch 监听的是文件的内容的变化
SETDATA(PATH, DATA, VERSION)
LIST(PATH)

counter

使用 zookeeper 实现简单的计数器

naive 的想法

X = GETDATA("f")
SETDATA("f", X + 1)

存在两个问题

GET 读取的数据可能不是最新的
SET 操作可能返回 false，也就是说其他的 client 抢先修改了值

下面给出正确的实现

WHILE TRUE:
    X, V = GETDATA("f")
    IF SETDATA("f", X + 1, V):
        BREAK

通过比对 version 参数可以解决上面的两个问题

由于 Zookeeper 的数据都在内存中，所以通常用 Zookeeper 来存储配置，而不是大型网站的真实数据
- vm-ft 中的 test-and-set 服务就可以通过类似的策略实现
该策略易受到 herd effect 的影响，对于 $n$ 个的 client 的并发请求，需要 $\mathcal{O}(n^2)$ 的 rpc 才能全部处理完成
这个例子，其实就是大家常说的 mini-transaction
- 这里之所以是 transactional 的，是因为一旦我们操作成功了，我们就对 counter 达成了读 - 更改 - 写的原子操作
- 之所以称之为 mini-transaction，是因为这里并不是一个完整的数据库事务 transaction

simple locks

使用 zookeeper 实现互斥锁

WHILE TRUE:
    IF CREATE("f", data, ephemeral=TRUE):
        RETURN
    IF EXISTS("f", watch=TRUE):
        WAIT

注意这里的 ephemeral flag，可以保证 client 崩溃后可以及时的释放锁
注意 CREATE 的 exclusive 特性
由于锁文件有可能在调用 EXISTS 之前就释放了，所以需要加上 if
由于判断文件是否存在和 watch 文件的变化是原子操作，所以当文件存在时，一定可以在 delete 之前 watch 到该文件
同样，该策略易受到 herd effect 的影响

simple locks without herd effect

使用 zookeeper 实现可扩展的互斥锁

也就是对于 $n$ 个的 client 的并发请求，只需要 $\mathcal{O}(n)$ 的 rpc 就能全部处理完成

CREATE("f", data, sequential=TRUE, ephemeral=TRUE)
WHILE TRUE:
    LIST("f*")
    IF NO LOWER #FILE:
        RETURN
    IF EXIST(NEXT LOWER #FILE, watch=TRUE):
        WAIT

注意这里的 sequential flag
这里的 f* 代表以 f 开头的所有文件
通过只 watch NEXT LOWER #FILE 减少 rpc 的数量
由于在分布式系统中可能会出现 Partial Failure，所以这里的锁并不能保证原子性
- 在一个分布式系统中，Zookeeper 实现的锁有如下的使用场景
  - 每一个获得锁的 client，需要做好准备清理之前锁持有者因为故障残留的数据
  - Soft Lock 用来保护一些不太重要的数据，例如运行 MapReduce Job 时，可以用这样的锁来确保一个 Task 同时只被一个 Worker 节点执行，也可以用来选举 Master

summary

Zookeeper 通过从多个副本读数据提升了性能，但同时又牺牲了一些一致性
Zookeeper 的 API 设计使得 Zookeeper 可以提供通用的协调服务

6.824 Note

TOC

info

intro

drivens

challenges

framework

abstraction & implementation

CAP

map reduce

crawler

mutex

channel

rpc

distributed storage system

gfs

master node

read file

write file

vm-ft

basic

sync

output

test-and-set

raft

basic

log

leader election

log sync

election restriction

fast backup

persistence

log snapshot

linearizability

zookeeper

consistency guarantees

sync

ready file / znode

api

counter

simple locks

simple locks without herd effect

summary