MIT-6.824-2022-Lab3 & Lab2D

KVRAFT算法及RAFT算法快照部分实现

前言

这个部分包含了Lab2D以及Lab3的内容，但是综合来说好像最后的运行速度比较慢，见谅见谅。

本系列其他链接

LAB2ABC	LAB2D & LAB3	Lab4
	本文	还得等等

• 所述实验均有完整的配有中文注释的代码，已上传至Github，详见lankoestee/MIT-6.824-2022。

任务基本介绍

Key-Value数据库操作

在本节中，需要实现一个能够保存Key-Value的数据库，并分别对客户端Client和服务器Server的实现，其底层架构是我们在Lab 2中实现的Raft算法。相较于Lab 2中普通的进行集群选主、日志同步和持久化的实现，Lab 3中所实现的KVRaft是由一定价值的，也就是其可以实现对与键值的存储。

对于本次任务而言，实际也是对于大部分的Key-Value数据库而言，都要实现3种方法。

• Put：对指定的Key直接设置Value；
• Append：对指定Key追加Value；
• Get：获取指定Key的Value；

对于该任务而言，其需要实现的整体结构如图-1所示。其分为了客户端Client和服务器Server两个部分，其中的服务器部分各自拥有一个Raft层，用于服务器的互相信息传递和同步，实现Raft的分布式集群功能。对于Client而言，多个的Server就是一个通过Raft连接的整体，依赖于Raft的日志同步功能，所有的服务器都将和复写相同的日志，将KV保存到各个服务器所对应的数据库。

对于客户端而言，其作用便是通过Put、Append和Get方法与服务器集群进行交互，对各个服务器中的简直进行复写、追加和获取操作，因此，对于Key-Value的交互而言，其三个操作方法本质上都是在传递对应的Key的Value，包括了读和写。

其中的服务器KVServer除了提供基础的数据库功能外，还需要保证可用性，一致性和安全性，Raft协议作为其底层可以进行有效的实现。这需要利用到我们在Lab 2中搭建的Raft架构。KVServer以5个节点为一组，每组KVServer上会有一个Leader和四个Follower。在这里，只有Leader会处理客户端发来的RPC请求。每次客户端收到对于数据库的处理操作（Put、Append和Get），都需要将操作封装成命令Command后通过接口投递到Raft层，然后Raft层将操作通过基本的心跳同步给Follwer以增加安全性和可靠性。

实现要求和实现顺序

在实现的KVRaft中，必须为Put、Append和Get方法提供强一致性。这样的强一致性是由下面的几个要求组成的。

• 如果一次调用一个，那么Get/Put/Append方法应该像系统只有一个副本那样工作，每次调用都应该观察前面的调用序列所暗示的对状态的修改；
• 对于并发调用，返回值和最终状态必须相同，就像操作以某种顺序一次执行一个操作一样；
• 如果调用在时间上有重叠，则进行并发调用，例如客户端X调用了Put方法，然后客户端Y调用了Append方法，那么需要先返回X的结果，也就是一次调用必须观察在调用开始之前完成的所有调用的效果。
• 系统必须具有幂等性，即客户端对服务器完全相同服务的调用无论多少次，其结果都和进行一次调用是一样的，这也是实现高并发的基本保障之一。

这样的强一致性对于应用程序来说很方便，因此几乎意味着，所有客户端看到的状态都是一样的并且看到了都是最新的状态。对于单机来说，提供强一致性相对容易。但是如果服务是有副本的，就是较为复杂的操作了。因为所有的服务器必须为并发请求选择相同的执行顺序，并且必须避免使用不是最新的状态来回复客户端。

该代码由两个部分组成，分别是Lab 3A和Lab 3B，但是由于Lab 3B依赖于之前尚未实现的Lab 2D，因此我将分3个部分进行说明和讲述，其中Lab 2D和Lab 3B被放入了一个章节中。

1. Lab 3A：需要实现一个没有快照的，基于Raft的Key-Value数据库服务，由于无需快照，该部分使用的Raft代码为其中作业所实现的Lab 2C代码；
2. Lab 2D：该部分是完成Lab 3B的基础，要求在Lab 2C代码的基础上实现Raft的快照机制；
3. Lab 3B：该部分是基于Lab 3A和Lab 2D而实现的一个含有快照机制的Key-Value数据库服务，对部分日志进行丢弃，其所使用的Raft代码是上面实现的Lab 2D部分代码。

在具体的实现方面，在Raft的论文中给出了客户端进行请求的相关方法，如图-2所示。

在Raft的原论文中，客户端Client想服务器进行请求的各个参数如下。

• clientId：调用请求的客户端id；
• sequenceNum：用于消除重复请求；
• command：请求状态机的命令，可能会影响状态。

服务器返回的各个参数如下。

• status：如果状态机成功应用命令则返回OK；
• response：如果请求成功，则为状态机的输出；
• leaderHint：如果服务器知道，则返回最近的Leader的地址。(用于不为Leader的服务器接收到之后将正确的Leader地址返回让客户端请求)

这样的ClientRequestRPC是比较复杂的，我们于Lab 3中仅采取其中的简化版本，即只保存当前最大请求的序列号，而且不提供重定向到Leader的功能。据此，可以进行后续的代码实现和分析了。

实现难点

在本实验中，需要实现一个高并发的操作和线性一致性的操作，实际上，Raft层的设计已经足以能够保证线性一致性，也就是能能够AppendEntries RPC等方法完成多个数据之间的数据同步，但对于高并发而言，Lab 3B对高并发的要求实际而言不是特别的高，在一半情况下，为了面对高并发的情景，通常需要使用负载均衡技术进行控制。然而在KVServer的实现中，LAB 3的要求描述并未使用负载均衡技术，而是所有Client的处理节点均是KVServer的Leader节点，而若Client将请求发送到了其他的Follower节点，则会无法处理，并要求客户端自己进行重新发送，也就是发送给下一个的KVServer节点，这样的设计显然不是为高并发进行设计的。但是我们需要实现最基础的并发控制。

并发控制而言，即是通过锁和线性化的设计以完成我们的目的。锁的存在保证了节点在处理一个线程的时候不会被另一个线程给打断，能够在处理完该线程之前不进行另一个线程，起到数据的保护作用。为了实现线性化的操作，我们也需要进行一定的调整，但总体而言较为简单，对于仍然被锁住的内容只需要引入一个合理的超时机制便可以进行实现了。

此外，KVRaft的实现是需要能够容错的，这个错误并不是指Client向Server发送的指令的错误，这样的请求我们一般认为是完全正确的。在测试中所发生的错误更多的是KVServer节点的错误，也就是KVServer节点发生宕机、网络分区、通讯故障等方面时拥有快速的修复能力。这一部分实际上是我们无需考虑的，因为在Lab 2A-2C的实现中已经基本解决了这一点，Raft中集群选主的设计能够有效面对多种意外情况，同时日志同步也保证了各个节点日志的一致性。但是我们仍然要使用快照机制，以提高系统对以外情况的处理性能，同时减小其内存占用。

死锁也是在实现过程中经常出现的问题，对于锁的使用，有mu.Lock的上锁过程就一定要对应mu.UnLock的解锁过程，防止某一个节点或进程上时间的占用锁。

KVRaft基本实现(Lab 3A)

任务分析

Lab 3A主要实现的是没有快照机制的Key-Value服务，客户端将Put、Append和Get的RPC发送到其关联的Raft层的Leader的键值服务器KVServer。KVServer将Put、Append和Get操作提交给Raft层，这样，Raft日志就会保存一系列的数据库读写操作。所有的键值服务器都将按顺序从Raft层获取信息，并执信其日志操作，将其操作到自己所对相应的Key-Value数据库。其目的是让各个服务器所维护的Key-Value数据库拥有相同的副本。

存在这样一种情况，客户端不知道此时谁是服务器集群的Leader，客户端可能会将Put、Append和Get的RPC请求发送到错误的KVServer，或者无法到达KVServer。此时，客户端应该通过将RPC请求发送到不同的KVServer来重试，如果KVServer将其方法操作提交到它的Raft日志，Raft层中的Leader将会通过响应其RPC将结果报告给发出的客户端。如果操作提交失败（例如，如果Leader被替换），服务器会报告错误，然后Client会尝试使用不同的服务器。

当服务器节点收到请求时候，应该先通过Start结构将请求操作同步到各个节点，让后各个节点根据请求对应的操作（Put、Append和Get），来执行对应的逻辑，这样就可以保证各个节点的一致性。

对于三个操作而言，Put操作和Append的操作需要进行时序的同步是很好理解的，但是对于Get操作，似乎没有进行同步后一致性处理的必要，可以使用图-3对其进行说明。假设严格按照编号时序执行，理论上Client-B应该Get到A的Put操作结果，但是由于Put需要通过Start去同步给其他节点，同步成功后（超过半数节点顺利同步），才会返回给KVServer，KVServer才能返回给Client-A，而如果Get操作不需要Start，直接返回，就有可能导致历史一致性错误。

此外，在KVRaft的实现中，还存在几个需要注意的点。

• 在Start操作过后需要等待Start的结果，收到结果后才能返回给Client操作是否成功。在本次实验中，单个的Client操作并不会并发，只有上一个操作执行结束了才会执行下一个。
• 由于Start操作是异步的，该操作无法保证成功（可能出现的一个问题是Leader会改变），因此在等待过程中，需要设定一个超时等待时间，超时后需要检查当前是否还是Leader，如果还是Leader则继续等待。
• 需要增加一个客户端RPC上的SequenceId用来充当序列号的作用，以保证Client的操作不会被重复执行。

基于上述的分析过程，我们可以得到如图-4所示的KVServer处理客户端请求的大致流程。

而需要有另一处地方处理Start成功后，Apply Channel才会返回成功的ApplyMsg，这里选择另外发起一个信息处理的协程，如图-5所示。

据此，我们可以对其进行初步的实现了。

功能设计与实现

数据结构的设计

在6.824源代码的common.go程序中，给我们了诸多需要实现的数据结构，其中最为主要的是Put、Append和Get三个操作方法的数据结构。其中，Put和Append的方法一个是改写，一个是追加，他们的方法是近乎一致的，故可以使用同一个数据类型进行描述。

在下面的代码中PutAppendArgs是客户端向服务器发送的Put和Append请求的RPC内容，而PutAppendReply是服务器返回给客户端的内容，因为他是一个写操作，如果一切正常的话，是需要客户端委托服务器写即可，也就是传送一个Key-Value给服务器，服务器理论而言不需要给客户端任何回应，但为了避免写入失败或写请求的RPC传输失败的情况，服务器需要向客户端返回一个错误码，这也是仅有的需要回复的内容。同样需要注意的是，与我们在任务分析中所说的相同，为了防止一个命令被重复执行，加入操作需要SeqId是十分重要的。

type PutAppendArgs struct {
	Key     string // 键名
	Value   string // 值
	Op      string // 操作类型（Put 或 Append）
	ClerkId int64  // 客户端 ID
	SeqId   int    // 操作序号
}

type PutAppendReply struct {
	Err Err // 错误类型
}

而对于Get操作，由于其传送的数据是不同的，也就是客户端仅需要向服务器传送一个Key，而服务器也需要向客户端传送一个Value，故其数据结构是与Put和Append的结构是有所不同的。客户端向服务器传送的RPC，也就是GetArgs的数据结构，其中不再需要Op来区分操作类型，也不再需要Value带来值，仅需要键名、客户端ID以及操作需要即可。但是对于服务器的回应PutReply而言，就需要附加上一个Value值来进行传递。同时为了避免读取失败和明确失败的原因，加入一个错误码也是十分有必要的。

type GetArgs struct {
	Key     string // 键名
	ClerkId int64  // 客户端 ID
	SeqId   int    // 操作序号
}

type GetReply struct {
	Err   Err    // 错误类型
	Value string // 值
}

当KVServer收到客户端发来的命令之后，会调用底层Raft中的Start命令，并且传送一个Op结构给Raft层，这个Op结构不再对Put/Append和Get进行区分，而是新建一个command的字符串，用于保存操作方法的类型，以便进行区分。其实现的代码如下。

type Op struct {
	Key      string // 操作的键名
	Value    string // 操作的值
	Command  string // 操作的命令类型（Get、Put、Append）
	ClientId int64  // 客户端的唯一标识符
	SeqId    int    // 操作的序列号
	Server   int    // 服务端的标识符
}

此外，关于客户端和服务器方面，也有两个相应的数据结构以存储对应的客户端的信息，分别是客户端的结构Clerk，和服务器的结构KVServer，其中KVserver具有Raft的相关类似特征，在后面的视线中也需要进行持久化存储，而客户端则是一个普通的客户端，可以通过调用labrpc包进行实现。

type Clerk struct {
	servers  []*labrpc.ClientEnd
	leaderId int   // 当前leader的ID
	clerkId int64 // clerk的唯一ID
	seqId    int   // 请求的序列号
}

type KVServer struct {
	mu           sync.Mutex
	me           int
	rf           *raft.Raft
	applyCh      chan raft.ApplyMsg
	dead         int32 // 通过Kill()方法设置
	maxraftstate int   // 当日志增长到一定大小时进行快照

	dataSource map[string]string   // 存储键值对的数据源
	messageMap map[int64]*ClerkOps // 客户端ID与ClerkOps结构体的映射表
	messageCh  chan raft.ApplyMsg  // 用于接收Raft层的ApplyMsg消息的通道
	persister  *raft.Persister     // 持久化存储
}

客户端发送请求

从任务分析和上面的数据结构分析中可以得知，客户端必须实现两种请求功能，一是Put和Append，二是Get，两种方法所依赖的数据结构是不同的，因此其方法也有所不同。

在PutAppend函数中，需要传入的参数是插入的键名，对应的值，以及操作名称，也就是Put或者Append。而在Get函数中，仅仅需要传入查询的键名即可。就其主体部分，实际上都是大差不差的。Clerk客户端首先会将请求发送至其认为的Leader服务器上，如果该Leader依然是Leader，则会回复正确信息，或向客户端返回查询键所对应的值。而若此时Leader发生了变化，Clerk则会向ServerId的下一个服务器发送请求，直至找到了新的Leader，此时才会收到处理成功的信息，或受到返回对应键的值。

// 客户端向服务器添加键值
func (ck *Clerk) PutAppend(key string, value string, opName string) {
	args := PutAppendArgs{
		Key:     key,             // 待写入或追加的键值对的键
		Value:   value,           // 待写入或追加的键值对的值
		Op:      opName,          // 操作类型，可以是 "Put" 或 "Append"
		ClerkId: ck.clerkId,      // Clerk 的唯一标识符
		SeqId:   ck.allocSeqId(), // 分配的序列号
	}
	reply := PutAppendReply{}
	server := ck.leaderId
	for {
		ok := ck.SendPutAppend(server%len(ck.servers), &args, &reply) // 发送 PutAppend 请求给指定的服务器
		if ok {
			if reply.Err == ErrWrongLeader { // 如果收到了 ErrWrongLeader 错误，表示当前服务器不是 Leader
				server += 1
				time.Sleep(50 * time.Millisecond)
				continue // 重试下一个服务器
			}
			ck.leaderId = server // 更新 Leader 的标识符
			break // 获取到响应，退出循环
		} else {
			server += 1
		}
		time.Sleep(50 * time.Millisecond) // 等待一段时间后继续重试
	}
}

// 客户端获取对应键的值
func (ck *Clerk) Get(key string) string {
	args := GetArgs{
		Key:     key,             // 待获取的键值对的键
		ClerkId: ck.clerkId,      // Clerk 的唯一标识符
		SeqId:   ck.allocSeqId(), // 分配的序列号
	}

	reply := GetReply{}
	server := ck.leaderId
	for {
		ok := ck.SendGet(server%len(ck.servers), &args, &reply) // 发送 Get 请求给指定的服务器
		if ok {
			if reply.Err == ErrWrongLeader { // 如果收到了 ErrWrongLeader 错误，表示当前服务器不是 Leader
				server += 1
				continue // 重试下一个服务器
			}
			ck.leaderId = server // 更新 Leader 的标识符
			break // 获取到响应，退出循环
		} else {
			server += 1
		}
		time.Sleep(50 * time.Millisecond) // 等待一段时间后继续重试
	}

	return reply.Value // 返回获取到的键值对的值
}

服务器处理请求

对于客户端所提交的请求，服务器也有相关的办法进行处理，他们实际上是两个文件中的重名函数，都叫做PutAppend和Get，但是其进行的处理是完全不同的。对于服务器中的这两个函数而言，其主要任务是负责和本地数据库进行交互，同时Leader要将该信息通过Raft层及时地通知其他服务器。其他Follower也会将其写入本地的数据库中。

对于服务器而言，首先应该接收来自客户端的函数，此后，需要异常情况的处理。一种很可能出现的情况是客户端误认为Leader暂时没有回复，导致其重发请求，这可能会导致两者的序列号不能做到匹配，此时就应该进行错误的设置，同时将错误码设为OK，说明Leader已经处理过该条信息了，防止Client由于没有受到回应继续向服务器发送大量信息。

随后，Server将会对该信息进行二次封装，封装为适合Raft进行处理的数据结构，也就是上面所提到的Op结构体。该结构体将不会再对PutAppend以及Get结构体进行进一步的结构体上的区分，而是通过结构体中的Command变量进行区分。封装好Op结构体后，将会调用Raft层中的start方法，将该信息作为日志传入Raft层进行相关调用，并调用WaitApplyMsgByCh函数进行等待。

func (kv *KVServer) PutAppend(args *PutAppendArgs, reply *PutAppendReply) {
	kv.mu.Lock()
	ck := kv.GetCk(args.ClerkId) // 获取客户端的PutAppendArgs结构体
	// 已经处理过了
	if ck.seqId > args.SeqId { // 如果PutAppendArgs结构体的序列号大于请求的序列号
		kv.mu.Unlock()
		reply.Err = OK // 设置错误码为OK
		return
	} // 表示收到了PutAppend请求
	// 开始一个命令
	logIndex, _, isLeader := kv.rf.Start(Op{
		Key:      args.Key,
		Value:    args.Value,
		Command:  args.Operate,
		ClientId: args.ClerkId,
		SeqId:    args.SeqId,
		Server:   kv.me,
	})
	if !isLeader { // 如果当前不是领导者
		reply.Err = ErrWrongLeader // 设置错误码为ErrWrongLeader
		kv.mu.Unlock()
		return
	}

    // 将当前命令的日志索引设置为ClerkOps结构体的消息唯一标识符
	ck.msgUniqueId = logIndex
	
    // 等待通道
	reply.Err = OK                                      // 设置错误码为OK
	Msg, err := kv.WaitApplyMsgByCh(ck.putAppendCh, ck) // 等待从通道接收到PutAppend操作的结果
	kv.mu.Lock()
	defer kv.mu.Unlock()
    // 表示收到了PutAppend操作的结果
	reply.Err = err
	if err != OK { // 表示领导者发生了变更
		return
	}
}

func (kv *KVServer) Get(args *GetArgs, reply *GetReply) {
	kv.mu.Lock()
	ck := kv.GetCk(args.ClerkId) // 获取客户端的ClerkOps结构体
    // 收到Get请求
	// 开始一个命令
	logIndex, _, isLeader := kv.rf.Start(Op{
		Key:      args.Key,
		Command:  "Get",
		ClientId: args.ClerkId,
		SeqId:    args.SeqId,
		Server:   kv.me,
	})
	
    // ---------- 中间此段与PutAppend是一样的 ---------- 

	_, foundData := kv.dataSource[getMsg.Key]
	if !foundData {
		reply.Err = ErrNoKey
		return
	} else {
		reply.Value = kv.dataSource[getMsg.Key] // 执行了Get操作
	}
}

服务器等待请求

上述我们提到，Leader会调用WaitApplyMsgByCh对Raft层的处理进行等待，具体的等待方式即是通过Raft层中的GetState函数获取当前的状态。并通过设置定时器的方法处理超时问题。若当前的服务器是集群的Leader且Raft层工作正常，将会返回OK，否则将会重置定时器，直至Raft层发生了任期错位的情况，也就是执行WaitApplyMsgByCh函数时的起始任期与当前问题发生了不一致的情况，则可以说明本节点不是Leader，无权进行操作，并返回ErrWrongLeader的错误。

如果Raft层一切顺利，将会进行下一步的processMsg的操作，也就是处理Raft层传来的ApplyMsg消息，这代表Raft层已经完成了相关的同步，需要KVServer向客户端发送反馈。如果日志的序列号大于ClerkOps结构体的序列号，则说明日志已经失效，将会忽略回应，也就是不对客户端进行相关的反馈，直接进入下一轮的等待或循环。而若向Raft层发送command的唯一标识符(msgUniqueId)与Raft层传回的命令索引(CommandIndex)，是相同的，则说明对应的信息写入正确，需要进行通知，也就是正式的写入本地的数据库中，也就是KVServer结构体所对应的dataSource，这代表服务器本地的数据库，对于KVRaft的实现而言，这个数据库是需要保持一致的。

为了方便理解，我将下述代码中的Debug部分的日志打印操作均去除了，但是在提交的代码中有所呈现。

func (kv *KVServer) WaitApplyMsgByCh(ch chan Op, ck *ClerkOps) (Op, Err) {
	startTerm, _ := kv.rf.GetState()                // 获取当前服务器的任期号
	timer := time.NewTimer(1000 * time.Millisecond) // 创建一个定时器，设置超时时间为1秒
	for {
		select {
		case Msg := <-ch: // 从通道接收到消息
			return Msg, OK // 返回接收到的消息和OK错误码
		case <-timer.C: // 定时器超时
			curTerm, isLeader := kv.rf.GetState()  // 获取当前服务器的任期号和领导状态
			if curTerm != startTerm || !isLeader { // 如果当前任期号不等于开始任期号，或者当前不是领导者
				kv.mu.Lock()
				ck.msgUniqueId = 0 // 将ClerkOps结构体的消息唯一标识符重置为0
				kv.mu.Unlock()
				return Op{}, ErrWrongLeader // 返回空的操作和ErrWrongLeader错误码
			}
			timer.Reset(1000 * time.Millisecond) // 重新设置定时器超时时间为1秒
		}
	}
}

func (kv *KVServer) processMsg() {
	for {
		applyMsg := <-kv.applyCh                                                    // 从通道接收应用层提交的日志
		Msg := applyMsg.Command.(Op)                                                // 解析日志中的Op结构体
		kv.mu.Lock()
		ck := kv.GetCk(Msg.ClientId) // 获取客户端的ClerkOps结构体
		// 当前不处理该日志
		if Msg.SeqId > ck.seqId { // 如果日志的序列号大于ClerkOps结构体的序列号
			kv.mu.Unlock()
			continue
		}

		_, isLeader := kv.rf.GetState()

		// 检查是否需要通知
		needNotify := ck.msgUniqueId == applyMsg.CommandIndex
		if Msg.Server == kv.me && isLeader && needNotify { // 如果当前服务器是领导者，并且需要通知客户端
			// 通知通道并重置时间戳
			ck.msgUniqueId = 0
			kv.NotifyApplyMsgByCh(ck.GetCh(Msg.Command), Msg) // 通过通道通知客户端
		}

		if Msg.SeqId < ck.seqId { // 如果日志的序列号小于ClerkOps结构体的序列号
            // 打印日志，表示忽略该日志
			kv.mu.Unlock()
			continue
		}

		switch Msg.Command { // 根据命令类型执行相应的操作
		case "Put":
			kv.dataSource[Msg.Key] = Msg.Value // 执行Put操作，将键值对写入数据源
			// 打印日志，表示执行了Put操作
		case "Append":
			// 打印日志，表示执行了Append操作
			kv.dataSource[Msg.Key] += Msg.Value // 执行Append操作，将值追加到键对应的现有值后面
		case "Get":
			// 打印日志，表示执行了Get操作
		}
		ck.seqId = Msg.SeqId + 1 // 更新ClerkOps结构体的序列号
		kv.mu.Unlock()
	}
}

运行结果

根据如上的主要代码，再辅以我在Lab 2C中实现的Raft层的相关功能，可以顺利的使用go test -run 3A命令通过相关的测试，测试结果如图-8所示。其能够通过原代码给出的所有测试。

实验心得

在Lab 3A的实现中，最为困难的是服务器与状态机的交互，也就是如何使用底层的Raft层进行操作。在代码实现的过程中，我也是基本按照上面展示的顺序进行实现的。首先必须确定实验过程中的基本结构体，也就是PutAppend和Get的相关请求以及应答。

在后面的服务器和客户端的实现中，实际上，客户端的实现相较于服务器是较为简单的，也是较为容易着手的地方。Client所需要实现的两个重要内容，一个是发送，一个是重试。因为Client本来使用的也是更高级的调用方法，不会涉及到多么底层的调用，只需要发布所需要的键值或者需要替换或添加的就好了。至于重试的方面，就是为了将请求传入真正的Leader中。

而服务器的实现较为复杂，最为主要的是其与Raft层的交互，KVRaft的实现相较于之前的Lab 2实际上已经降低了部分难度了，通过Start方法调用Raft层，可以实现强一致性的安全同步。那么在KVServer端要解决的首要问题，便是并发控制，也就是按照先来后到的顺序保证信息能够按照先来后到的顺序进行处理，同时使用锁保护正在处理的线程。

KVRaft快照实现(Lab 2D, Lab 3B)

任务分析

在KVRaft的系统中，随着系统的不断运行，底层raft保存的日志越来越多，这将带来三个问题。

• 每一次持久化数据的时间会越来越长，拖慢日志同步的时间，因为持久化数据时需要用到rf.mu锁；
• 当底层raft宕机后，通过持久化数据恢复的时间会非常漫长；
• 脱离集群很久的节点重新回到集群时，需要同步的日志会非常的多。

也是因此，有必要定时对上层server的状态保存快照，压缩底层raft的日志，这样一来，不论是持久化数据的时间、恢复的时间，已经脱离集群很久节点的同步时间都会大大缩短，提高系统的性能。其进行同步的过程如图-12所示。其快照的示意如图-11所示。

为此，我们必须在我们原来的Raft结构中进行更改，使其能够实现快照的功能，在这里，我是基于我在Lab 2C中完成的功能进行更改的。

在Lab 2C中，我们已经实现了持久化的保存，并且能够实现集群选主和日志同步的等基本功能。Lab 2D要求我们实现一个会被周期性调用的Snapshot方法，这将进行快照保存，也就是对于一个最高的日志索引，将之前的存入Persister中并在Raft层中丢弃，以起到定期保存的作用，减少Raft层的内存占用。同时还需要实现一个InstallSnapshot方法，也就是让Leader节点告知一个落后的Raft节点去用快照信息替换其自身的状态。

在快照的实现过程中，Raft算法通过设置LastIncludedIndex和LastIncludedTerm两个变量来进行快照的记录。当节点中记录的日志达到一个阈值之后，服务层会生成一个快照，并传递给Sanpshot函数（在2D的测试中，是每10个日志生成一个快照）。

快照的实现主要有两个需要注意的点，一是快照实现时需要对日志进行剪裁，二是确定快照应用所需要的条件。在我们的实现中，仅是通过Leader节点进行快照的保存，这与6.824 Lab 2D的要求是相同的，我们不必实现通过偏移机制去分割快照，而是应该由Leader直接将完整的快照数据发送给存在数据缺陷的Follower。但其实这样会带来一个后果，就是当Follower与Leader的日志相差较大的时候，Leader一次性向Follower发送了过多的日志，可能会影响整体的网络带宽，不过这不是我们在这里需要考虑的事情。

具体而言，其要实现如下的交互流程。

1. Leader节点发起快照请求：Leader节点首先会判断是否需要进行快照，如果需要则生成快照数据，并通过安装快照请求将快照数据发送给其他节点。
2. Follower节点接收快照请求：Follower节点接收到Leader节点发送的安装快照请求后，会根据请求中的信息进行处理。
3. Follower节点处理快照请求：Follower节点首先会检查快照请求的任期是否比自己当前的任期更低，如果是则拒绝安装快照请求。然后，Follower节点会检查是否存在更新的快照，如果存在则拒绝安装快照请求。接着，Follower节点会设置等待中的快照数据和相关参数，并重置选举超时时间。
4. Leader节点接收快照安装回复：Leader节点接收到其他节点发送的快照安装回复后，会根据回复中的信息进行处理。
5. Leader节点处理快照安装回复：Leader节点首先会检查回复中的任期是否比自己当前的任期更低，如果是则更新自己的任期。然后，Leader节点会根据回复中的信息确定是否存在更新的快照，如果存在则更新自己的快照数据。接着，Leader节点会根据回复中的信息更新状态机相关参数，并将快照数据写入持久化存储。
6. 状态机应用快照：一旦Leader节点将快照数据写入持久化存储并更新状态机参数后，状态机会将快照数据应用到自己的状态中，从而使得状态机与Leader节点保持一致。

对于Lab 3B而言，其与Lab2D的区别并不显著，但是它充分利用了Lab 2D中所实现的日志压缩以及快照机制，并且，每个KVServer都会执行快照任务。一个在其中存在的问题是，由于进行了快照机制，日志会只剩下后面的部分，这会对日志的下标产生严重的影响，需要对所有的下标进行全面的替换。为此，根据原论文的图5.3，如图-7所示，我们需要添加两个变量。并设计一个全新的RPC结构。在这个RPC中，与AppendEntries RPC所不同的是，其设计的数据传送部分data[]是一个bytes形式的数据，而非日志同步时使用的log[]类型的数据，这有几个好处。更小的数据传输量将会加快落后Follower节点的恢复速度，并减少网络传输的数据量。

回到我们主要实现的部分，也就是Lab 3B的部分，它需要每个节点不断进行快照检查，并在不断进行快照的判断。具体而言，也就是比较测试代码传入的maxraftstate（进行快照保存的最小单位）和从持久化池persister中读出的RaftStateSize（尚未进行保存的状态条目）进行比较，以判断是否需要进行快照操作。

此外，在Lab 3B的实现中还需要注意以下几点。

• Lab 3B的测试主要检查了Raft层的Log是否超标了，因此进行快照的操作需要放在ProcessMsg中进行判断，主要负责判断两件事情。一是检查ApplyMsg是否为SnapshotMsg，二是负责判断当前的persisiter.RaftStateSize是否超标了，也就是超过了maxraftstate，以判断是否需要进行快照操作。
• 当KVServer从ApplyChannel中读取出来的RPC信息为InstallSnapshot时，直接采用Snapshot中的数据覆盖原生的数据即可。
• 需要对一些状态进行持久化，比如说客户端的Clerk，以防止重复执行相同命令，保证幂等性。

还需要说明的是，Lab 3B相较于Lab 3A而言，其发生的改进只是基于KVServer进行的相关改进，实际上与Client并无关系。客户端与服务器之间的RPC数据传输一切照常。

功能设计与实现

快照安装基本结构(Lab 2D)

为了实现快照的安装，我按照图-7的要求，设计了简单的快照发送结构体和快照回复RPC结构体，以实现快照的应用功能。

type InstallSnapshotArgs struct {
	Term              int    // 领导者任期
	LeaderId          int    // 领导者ID，以便于跟随者重定向请求
	LastIncludedIndex int    // 快照中包含的最后日志条目的索引值
	LastIncludedTerm  int    // 快照中包含的最后日志条目的任期号
	Data              []byte // 快照数据
}

type InstallSnapshotReply struct {
	Term int // 当前任期，用于领导者更新自己
}

创建快照的实现(Lab 2D)

为了更好的传输快照信息，需要设计一个Snapshot方法进行快照的创建，这也是测试中需要频繁进行调用的函数。由于担心可能出现的调用时延等问题，保险起见，我们仍然需要对传入的参数进行一定的校验。

如果我们当前的截断下标都已经比请求中给出的高了，那说明相关快照信息已经存储过了；而如果请求中给出的下标都比我们提交的水位还高（不应该发生），那就拒绝这次请求。理论而言，第二种情况是不太可能出现的，但是为保险起见，还是把它加了上去。

其最为重要的实现，就是进行切片以生成所需要的新的日志，并更新状态机相关参数。

// 进行快照的创建
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	lastLogIndex, _ := rf.lastLogInfo()

	// 如果快照中包含的最后日志条目的索引值大于等于当前索引，则说明已经进行过快照
	if rf.lastIncludedIndex >= index {
		return
	}

	// 如果提交的索引小于当前索引，未提交的日志无法进行快照
	if rf.commitIndex < index {
		return
	}

	// 切片生成所需要的新日志，并更新状态机相关参数
	newLog := rf.getSlice(index+1, lastLogIndex+1)
	newLastIncludeTerm := rf.getEntry(index).Term
	rf.lastIncludedTerm = newLastIncludeTerm
	rf.log = newLog
	rf.lastIncludedIndex = index
	rf.snapshot = snapshot

	// 将快照数据写入持久化存储
	rf.persistAndSnapshot(snapshot)
}

安装快照请求的发送以及接收处理(Lab 2D)

为了实现安装快发送的功能，我设计了一个leaderSendSnapshot的函数，它用于向节点发送快照。它由Leader进行调用，并在监听来自Follower的回复，以确认它们是否根据安装快照请求进行了日志更新的操作。同样的，在正式发送恢复之前，需要进行一系列的检查，这包括了两个检查的条件。

• 如果回复的任期比当前节点的任期更低，则忽略回复；
• 如果安装快照请求后当前节点的任期发生了改变，则忽略回复。

发送并接受回复成功后，Leader将会更新指定Follower的下一个期望日志索引。

// 由Leader发送快照
func (rf *Raft) leaderSendSnapshot(args *InstallSnapshotArgs, server int) {
	reply := &InstallSnapshotReply{}
	ok := rf.sendInstallSnapshot(server, args, reply)
	if ok {
		rf.mu.Lock()
		defer rf.mu.Unlock()
        
		if reply.Term < rf.currentTerm {
			return // 如果回复的任期比当前节点的任期更低，则忽略回复
		}
		if rf.currentTerm != args.Term {
			return // 如果安装快照请求后当前节点的任期发生了改变，则忽略回复
		}
		rf.checkTerm(reply.Term)                                     // 检查并更新当前节点的任期
		newNext := args.LastIncludedIndex + 1                        // 计算新的nextIndex
		newMatch := args.LastIncludedIndex                           // 计算新的matchIndex
		rf.matchIndex[server] = max(newMatch, rf.matchIndex[server]) // 更新matchIndex
		rf.nextIndex[server] = max(newNext, rf.nextIndex[server])    // 更新nextIndex
	}
}

Follower在接收到由Leader发来的快照安装请求后，会调用InstallSnapshot函数，该函数将处理由Leader发来的快照。并会在这个过程中重置自身的选举超时时间和心跳超时时间。对于Leader发来的请求，Follower同样需要进行一些检查，这包括了两个检查。

• 如果当前任期大于请求任期，则重置当前任期为请求发来的Leader任期，则拒绝安装快照请求；
• 已存在更新的快照，同样拒绝安装快照请求。

其函数InstallSnapshot的实现如下。

func (rf *Raft) InstallSnapshot(args *InstallSnapshotArgs, reply *InstallSnapshotReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	if args.Term < rf.currentTerm {
		reply.Term = rf.currentTerm
		return // 当前任期大于请求任期，重置当前任期为请求发来的Leader任期，拒绝安装快照请求
	}
	rf.checkTerm(args.Term)
	reply.Term = rf.currentTerm

	rf.setElectionTimeout(randHeartbeatTimeout()) // 重置选举超时时间，等待下一个可能的心跳超时

	if rf.waitingIndex >= args.LastIncludedIndex {
		return // 已存在更新的快照，拒绝安装快照请求
	}
	rf.leaderId = args.LeaderId
	rf.waitingSnapshot = args.Data           // 设置等待中的快照数据
	rf.waitingIndex = args.LastIncludedIndex // 设置等待中的快照索引
	rf.waitingTerm = args.LastIncludedTerm   // 设置等待中的快照任期
}

需要注意的是，当Follower接收到InstallSnapshot的RPC请求时，不能直接将其应用在内存中snapshot字段上。此时不能立刻应用快照，需要保证Raft和状态机都应用快照成功，不然可能会导致状态机和Raft之间存在不一致。

在这里，需要将快照信息先通过applyCh传给状态机，状态机接受到了快照之后，再通过调用Raft代码上的CondInstallSnapshot来协商，Raft代码在返回结果中会告知状态机是否需要安装快照，同时视情况更新自己的状态信息。

为此，我也实现了CondInsatllSnapshot函数，该函数是状态机一侧进行调用的，用于和Raft协商确定快照是否是需要进行安装的。如果已经确定当前节点的commitIndex还小于快照中给出的截断位置，那么就可以开启安装流程，否则就返回false告诉状态机不要应用这个快照，因为他还不够新。安装的时候我们也采取类似的策略，日志在快照给出的位置进行截断即可，如果我们的日志非常短还没有到截断位置，那就清空日志即可。其中CondInstallSnapshot函数的实现如下。

// 状态机进行安装条件协商
func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	lastLogIndex, _ := rf.lastLogInfo()

	// 如果快照的最后日志条目索引小于当前日志条目索引，忽略快照
	if rf.commitIndex >= lastIncludedIndex {
		return false
	}

	// 如果快照的最后日志条目索引大于当前日志条目索引，截断日志
	if lastLogIndex >= lastIncludedIndex {
		rf.log = rf.getSlice(lastIncludedIndex+1, lastLogIndex+1)
	} else {
		// 如果不是，清空日志
		rf.log = []LogEntry{}
	}

	// 更新状态机相关参数
	rf.lastIncludedIndex = lastIncludedIndex
	rf.lastIncludedTerm = lastIncludedTerm
	rf.lastApplied = lastIncludedIndex
	rf.commitIndex = lastIncludedIndex
	rf.snapshot = snapshot

	// 将快照数据写入持久化存储
	rf.persistAndSnapshot(snapshot)
	return true
}

日志同步部分的微调(Lab 2D)

在Leader发送日志的过程中，可能由于Follower节点日志过于落后，导致无法找到对应的日志，也就是Leader发现自己的lastIncludedIndex不高于Follower的nextIndex的时候，这种时候我们需要发送InstallSnapshot RPC。同理当AppendEntries请求发生回退的时候，极有可能我们会回退到某个Leader找不到的古早日志位置，这个时候我们也需要发送快照。故我们要对在Lab 2B日志同步部分实现的代码进行修改，具体修改的部分是最后的一部分，发送快照以解决找不到日志的问题。

// 提交日志条目
func (rf *Raft) leaderSendEntries(args *AppendEntriesArgs, server int) {
	reply := &AppendEntriesReply{}
	ok := rf.sendAppendEntries(server, args, reply)
	if ok {
        
		// ---------- 前序内容与Lab 2C一致 ----------

		if reply.Success {
			// ---------- 前序内容与Lab 2C一致 ----------
		} else {
			// ---------- 前序内容与Lab 2C一致 ----------

			// 如果 nextIndex 小于等于 lastIncludedIndex，说明需要发送快照
			if nextIndex <= rf.lastIncludedIndex {
				rf.sendSnapshot(server)
			} else if lastLogIndex >= nextIndex { // 如果 lastLogIndex 大于等于 nextIndex，说明日志不一致，重新发送日志条目
				newArg := &AppendEntriesArgs{
					Term:         rf.currentTerm,
					LeaderId:     rf.me,
					PrevLogIndex: nextIndex- 1,
					PrevLogTerm:  rf.getEntry(nextIndex - 1).Term,
					Entries:      rf.getSlice(nextIndex, lastLogIndex+1),
				}
				go rf.leaderSendEntries(newArg, server)
			}
		}
	}
}

此外，除了Leader无法找到过于早的日志，Follower也可能出现这种情况，例如一个Follower刚进行了一轮快照，截断了若干日志。接着接收到了Leader发来的日志，在AppendEntries中我们需要做的第一步就是比对PrevLogIndex位置上的日志是否对应，如果不对应则回退，但是有可能这个位置已经被截断丢弃了。

由于日志没法从快照中重新恢复出来，那么我们只能默认被快照的这些日志，既然已经被提交，加上被快照，那么自然就是一致的。我们只能将Leader发来的日志条目，重新截断，在Follower能接受的位置重新开始比对。对AppendEntries的代码修改如下所示。

// AppendEntries Follower接收Leader的追加/心跳包
func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	
    // ---------- 前序内容与Lab 2C一致 ----------

	// 如果PrevLogIndex小于当前节点的lastIncludedIndex，表示之前的日志已经被快照截断了
	if args.PrevLogIndex < rf.lastIncludedIndex {
		alreadySnapshotLogLen := rf.lastIncludedIndex - args.PrevLogIndex
		// 如果已经快照截断的日志长度小于等于收到的Entries的长度，说明日志部分匹配，需要将截断的部分补充回来
		if alreadySnapshotLogLen <= len(args.Entries) {
			newArgs := &AppendEntriesArgs{
				Term:         args.Term,
				LeaderId:     args.LeaderId,
				PrevLogTerm:  rf.lastIncludedTerm,
				PrevLogIndex: rf.lastIncludedIndex,
				Entries:      args.Entries[alreadySnapshotLogLen:],
				LeaderCommit: args.LeaderCommit,
			}
			args = newArgs
			Debug(dWarn, "S%d Log entry at PLI already discarded by snapshot, readjusting. PLI: %d, PLT:%d, Entries: %v.",
				rf.me, args.PrevLogIndex, args.Entries)
		} else {
			// 如果已经快照截断的日志长度大于收到的Entries的长度，说明已经匹配，可以直接返回成功
			Debug(dWarn, "S%d Log entry at PLI already discarded by snapshot, assume as a match. PLI: %d.", rf.me, args.PrevLogIndex)
			reply.Success = true
			return
		}
	}
    
    // ---------- 后续内容与Lab 2C一致 ----------

}

状态的保存及读取(Lab 3B)

在有了Lab 2D的实现基础后，Lab 3B的实现理论而言就稍微简单了许多，但是其对底层Raft层的快照要求较高。在后续的检测中容易检测出Raft层的问题。但是我在这里的实现好似并没有遇到这样的问题。

在Lab 3B的实现中，快照机制的调用不再是测试代码中的直接调用，而是有了更为高级上游的调用方式，也就是直接保存KVServer的当前状态，包括Raft层的状态和数据的状态，而不仅仅是Raft的日志状态。

故此，就有了两个简单的函数，分别是saveKVState和readKVState，它们分别负责状态的保存和读取。

func (kv *KVServer) saveKVState(index int) {
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	cks := make(map[int64]int) // 创建一个map，用于存储客户端ID和序列号
	for ckId, ck := range kv.messageMap {
		cks[ckId] = ck.seqId // 将客户端ID和序列号添加到map中
	}
	e.Encode(cks)
	e.Encode(kv.dataSource)
	kv.rf.Snapshot(index, w.Bytes()) // 将数据写入快照
}

func (kv *KVServer) readKVState(data []byte) {
	if data == nil || len(data) < 1 {
		return // 如果数据为空或长度小于1，则直接返回
	}
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	cks := make(map[int64]int) // 创建一个map，用于存储客户端ID和序列号
	dataSource := make(map[string]string) // 创建一个map，用于存储数据源
	if d.Decode(&cks) != nil || d.Decode(&dataSource) != nil {
		DPrintf("[readKVState] decode failed ...")
	} else {
		for ckId, seqId := range cks { // 遍历客户端ID和序列号的map
			kv.mu.Lock()
			ck := kv.GetCk(ckId) // 根据客户端ID获取客户端
			ck.seqId = seqId // 更新客户端的序列号
			kv.mu.Unlock()
		}
		kv.mu.Lock()
		kv.dataSource = dataSource // 更新数据源
		kv.mu.Unlock()
	}
}

进行快照时机的选择(Lab 3B)

再设计了状态的保存和读取之后，何时需要进行状态的保存便是一个重要的问题。实际上，保存和添加函数的调用都是不太频繁的，仅在一些特定的地方需要进行调用。其中，读取函数需要在启动初始化KVServer的时候和processMsg读到了Snapshot Msg的时候调用，而且也仅是保存当前处理条目之前的状态。而保存函数也只需要在processMsg中进行调用即可，两个函数在pocessMsg和StartKVServer中的调用如下所示。

func (kv *KVServer) processMsg() {
	for {
		// 从通道接收到消息
		applyMsg := <-kv.applyCh
		Msg := applyMsg.Command.(Op)
		DPrintf("[KVServer-%d] Received Msg from channel. Msg=%v", kv.me, applyMsg)

		// 检查是否需要读取快照，如果需要，则进行快照
		if applyMsg.SnapshotValid {
			kv.readKVState(applyMsg.Snapshot)
			continue
		}

		kv.mu.Lock()
		ck := kv.GetCk(Msg.ClientId) // 获取客户端的ClerkOps结构体
		if Msg.SeqId > ck.seqId {
			kv.mu.Unlock()
			continue
		}

		// 检查是否需要保存快照
		_, isLeader := kv.rf.GetState()
		if kv.needSnapshot() {
			kv.saveKVState(applyMsg.CommandIndex - 1)
		}

		// ---------- 后续内容与Lab 3A相同 ----------
	}
}

func StartKVServer(servers []*labrpc.ClientEnd, me int, persister *raft.Persister, maxraftstate int) *KVServer {
	labgob.Register(Op{})

	kv := new(KVServer)
	kv.me = me
	kv.maxraftstate = maxraftstate  // 该参数由测试代码传入

	kv.applyCh = make(chan raft.ApplyMsg, 1000) // 创建一个通道，用于接收Raft层的ApplyMsg消息
	kv.rf = raft.Make(servers, me, persister, kv.applyCh) // 创建一个Raft结构体
	kv.mu.Lock()

	// 进行一系列初始化
	kv.dataSource = make(map[string]string)
	kv.messageMap = make(map[int64]*ClerkOps)
	kv.messageCh = make(chan raft.ApplyMsg, 1000)
	kv.persister = persister
	kv.mu.Unlock()

	// 读取被持久化的快照数据
	kv.readKVState(kv.persister.ReadSnapshot())
	go kv.processMsg()
	return kv
}

可以看到，在processMsg中进行调用时，是需要进行一定的检查的，自然不能每次进入该函数都进行一次调用，也就是在前面说的，比较kv.persisiter.RaftStateSize和maxraftstate两个变量，以进行保存快照的判定，其函数实现如下。

func (kv *KVServer) needSnapshot() bool {
	return kv.persister.RaftStateSize() >= kv.maxraftstate && kv.maxraftstate != -1
}

运行结果

根据上述的代码，我分别进行了Lab 2D和Lab 3B的运行，其中，Lab 3B的运行是使用了Lab 2D最终完成的Raft层作为交互层的，也就是进行了import操作。其运行结果如图-9和图-10所示。两者能够十分顺利的通过测试。

实验心得

Lab 2D和Lab 3B的实现也是不轻松的，主要的难点还是在Lab 2D上。快照的实现需要进行大量的代码工作，为此，我对在Lab 2C中书写的代码进行了一定的重构，最主要的工作量是日志系统的重构，改为直接从Raft结构体获取日志对象。

在Lab 2D中，调试的工作也变的困难起来了，由于需要对原本已经实现的结构进行诸多的修改，而部分修改可能会直接导致代码甚至无法进行日志同步的基本要求，尤其是对AppendEntries函数的修改至关重要。而且，Lab 2D的测试时间很长，微小的代码改动也只能通过动辄3分钟起步的冗长测试进行修改。

对比而言，Lab 3B的实验就简单了许多，我也查阅的不少资料，确定了保存和读取状态的函数只需要在processMsg和系统初始化阶段进行执行，而保存状态和读取状态函数的写法也较为简单，代码的书写并未造成太大的问题。实际上Lab 3B的测试对KVServer的要求并不高，而是对Raft的快照机制拥有很高的要求，但幸运的是，作为底层的Raft层并未出现什么问题，测试也较为顺利的通过了。