lab3-Fault tolerant Key/Value Service

MIT6.824分布式原理的课程lab，这篇文章记录lab3的详情。

lab3的实验分为两个任务，一是基于lab2完成的raft协议，完成一个不带日志压缩的 Key Value Service系统；二是在已经搭建好的Key Value Service上实现日志压缩(也即 snapshot)。

Key/value service without log compaction

Key Value Service 由客户端和服务端组成，其中客户端处于用户的机器上，服务端和它底层的raft处于服务器上，并且每个服务端与对应的raft处于同一个机器，即raft作为底层，负责一致性，而服务端处于上层，负责接受从客户端发来的请求并结合raft的反馈对状态机做相应的修改。

通常，集群由n个服务端组成，只有一个服务端对应的底层raft为leader，也即只有这个服务端是leader。客户端的交互对象就是这个leader服务端。客户端发过来一个command给leader服务端，然后leader服务端把这条command拿去给集群内的follower达成共识，然后commit，一旦commit后，leader就可以把这个command执行（应用）到自己的状态机。

在客户端看来，就好像服务端并没有分布式集群，它仅仅是在和leader进行交互。

具体到本次lab，服务端负责维护的是一个kv数据库，在lab中我们将其简化为一个map[string] string。服务端需要处理三种请求：

GET(key)：负责接受一个key，并查询其对应的value，如果没有找到，就返回空字符串。
PUT(key,value)：接受kv pair，并将其放入kv数据库。(如果以前有，就覆盖，如果以前没有，就新增)
APPEND(key,value)：接受kv pair，并将其append到数据库。(以前有，就将新value追加在旧value后面，以前没有，就直接新增)

需要注意的是，Raft需要实现线性化语义。即作为客户端，只有在成功收到本次command的结果后，才会发起下一次command。不会出现一次性并行发送n条command，然后分别等待它们的结果的情况。

客户端设计

客户端应该向集群中的leader发送command，但刚开始发送Get、Put之类的时候，它并不知道谁是leader。因此可以采用轮询的策略，当对方不是leader是，客户端会收到 ErrWrongLeader 的错误，这时它就可以向下一个节点发送这条命令了。

raft的线性化语义也必须在客户端得到实现，即：客户端在本条指令未得到结果前不会立刻return。比如下面的Get方法，如果网络拥塞导致超时，那就等待一个时间段然后重试；如果是错误的leader，就重新定位到下一个集群节点，并发送请求；如果是服务端超时，则可能是定位到了partition的leader上，导致command迟迟无法commit，也需要换一个节点发送请求；直到收到正确的结果后才return

另外，对于同一条command，客户端可能会在网络中发送不止一条，所以需要给command编号。

func (ck *Clerk) Get(key string) string {
  
	request := GetArgs{Key: key, ClientId: ck.clientId, OperationId: ck.operationId}
	ck.operationId++
	for {
		response := GetReply{}
		ok := ck.servers[ck.leaderId].Call("KVServer.Get", &request, &response)

		if ok == false {
			ck.leaderId = (ck.leaderId + 1) % len(ck.servers)
			time.Sleep(RetrySendReqTimeMs)
			continue
		}

		if response.Err == ErrNoKey {
			return ""
		} else if response.Err == ErrWrongLeader {
			ck.leaderId = (ck.leaderId + 1) % len(ck.servers)
			continue
		} else if response.Err == ErrRaftTimeout {
			//超时，maybe定位到了partition的旧leader上
			ck.leaderId = (ck.leaderId + 1) % len(ck.servers)
			continue
		} else if response.Err == OK {
			return response.Value
		} else {
			log.Fatal("wrong GET reply args")
		}
	}
	return ""
}

服务端设计

服务端接收从客户端发来的command，并把command通过Start交给raft层。如果raft发现自己不是leader，就会拒绝这个command，并return false，服务端得到false后需要将 ErrWrongLeader 错误反馈给客户端。

如果start成功，则说明command已经交给raft层，由raft层负责达成共识。在达成共识后会通知服务端，服务端就可以apply或者说执行这条command了。

由于考虑到效率问题，整个过程应该是并发的，而考虑到raft层达成一致需要一段时间，因此start 一条command、执行command、反馈结果给客户端这三个过程应该是异步的。

如上图所示，当一个command到达服务端的RPC接口后，服务端将command交给raft层的start函数，并启动一个定时器，然后等待结果。raft层在达成一致后，会把command写入raft.applyCh。服务端的apply Loop收到通知后，开始执行这条command并把结果应用到自己的数据库中。并把结果通过service.applyCh通知到RPC接口，RPC接口收到通知后，返回结果给客户端。如果RPC接口函数迟迟得不到通知，超时定时器就会到期，此时接口函数直接给客户端返回ErrRaftTimeout错误。

图中的raft.applyCh只有一个，用于raft层向service层通知。而service.applyCh有多个，即每次RPC请求，就新创建一个，用于服务端的service Loop向服务端的RPC接口函数通知command结果。

可以看到，raft层只是用来达成一致的，当它认为集群中过半数的节点都同步了某条指令，它就会使用applyCh把它通知给Service层，告诉service层，你可以放心的执行这条command了。

序列号：客户端有可能会重发某条指令（比如网络拥塞导致客户端主动重发）这就要保证重发的指令不能被又一次执行，必须返回上次执行过的结果。解决办法就是给command编号，具体来说，使用<clientId, commandId> pair来标识每一条command，其中clientId表示客户端的识别号，commandId表示指令的Id，指令Id是递增的。服务端记录自己最近已经apply到了哪一条指令，当有新指令需要处理时，先检查它的序列号Id是否小于自己上一次处理过的，如果小于，则说明这条指令已经处理过了，直接给用户回复Ok即可(丢弃也行，因为既然更大的command都已经处理过了，说明之前的command用户早已收到结果了)。

Key/value service with snapshots

partB要求在key value Service 的基础上完成快照功能。

在正常运行的过程中，raft会将自己的log持久化，当系统崩溃并重启后，由于没有持久化service层的状态机，所以整个系统都是0状态，因此raft会将自己的commitIdx和applyIdx(这两个数据不需要持久化)重置为0，并从第一条log开始重新执行到最新的log位置。这会浪费大量的时间，并且为了恢复状态，需要保存从系统开机到当前的所有log，这是很占存储空间的。

快照snapshot解决了这个问题，即在某个时间节点把service层的状态机的数据全部保存一份(在这里就是service层的kv数据库)，当系统崩溃重启后，只需要先把snapshot恢复到service层，然后再依次执行快照时间点之后的所有log。

整个快照机制如上图所示(图来自6.824实验的主页)。

建立快照

每个节点都会建立自己的快照，包括leader和follower。

service层会定期检测raft层保存在硬盘中的持久化数据的大小(主要是log太占空间)，一旦发现持久化数据文件超过了某个大小，就启动快照机制。service层首先将自己维护的数据库序列化，然后通知raft层将raft层的日志截断，丢弃截断的前面的log。从哪里截断呢？应该从service层保存自己快照的最后一条log截断。之后，将service层序列化的数据和raft层的需要持久化的一些状态一起写入硬盘。建立快照就ok了。

service层需要持久化的数据（写入快照的数据）：自己的数据库、lastApplyIndex、lastApplyTerm、recentApplyOpId

raft层需要持久化的数据和lab2一样，仍然是那些

发送快照

快照的发送由leader到follower。通常发生在leader发送心跳数据(包括append log)时，leader发现follower的log落后于leader太远时，会把自己的快照发送给follower。

具体到代码实现上，leader平时使用nextIndex数组来和follower协商，下次发送从哪里开始的log。当leader发现，nextIndex数组已经匹配到自己已经抛弃的log的index时(这些log在建立好快照后被抛弃)，就需要发送snapshot数据而非发送log数据。具体的RPC参数可见原论文的图。

安装快照

安装快照涉及到service层和raft层的互动，比较麻烦。

首先由Leader向follower发送快照数据。follower收到这些数据后，向service层的applyCh写入一条通知，通知的内容就是需要安装快照。然后service层首先调用raft层的condInstallSnapshot，让raft判断是否安装这个快照（如果raft层已经拥有这个快照的状态对应的最后一条log，或者raft层已经有比它更新的快照），就不用安装这个快照。否则，需要安装快照。raft层需要将applyIndex设置为快照状态对应的最后一条log的index，然后返回true。service层收到true后，将快照读取，并利用读取的数据重置自己的状态。

全过程中，raft需要重置的数据，是applyIndex。而commitIndex不用重置，因为commitIndex是从leader处获得的。

一些bug

channel不能加锁。这个和RPC调用不能加锁一样，channel加锁可能导致死锁。不知道为什么前几次实验都通过了，这个bug一直到lab3才暴露出来。

ZhuQiwei's Blog