MIT6.824-lab3B-Key/value service with snapshots(实现快照的KV服务)

3B(实现快照的KV服务)

任务

• 完成saveSnapshot函数创建快照。初始化时传入的maxraftstate 表示持久 Raft 状态的最大允许大小，当当前raft日志的大小大于该值，就可以进行一次快照；快照要保存的数据主要是：服务端的data数据、用于线性化语义的lastApplies数据结构；而具体还是调用Raft的Snapshot来进行生成快照。
• 完成readPersist函数读取快照。主要是在初始化、接收到快照命令这两处进行调用，主要就是读取快照中的data数据和lastApplies数据，不同的地方在于，如果不是在初始化中调用，要调用Raft的CondInstallSnapshot来处理日志和状态同步。
• 在合理的地方进行快照创建和快照读取。

任务须知

• 测试代码将 maxraftstate 传递给 StartKVServer()。 maxraftstate 表示持久 Raft 状态的最大允许大小（以字节为单位）（包括日志，但不包括快照）。我们应该将 maxraftstate 与 persister.RaftStateSize() 进行比较。每当应用一个命令，就进行一次快照检查，当服务器检测到 Raft 状态大小大于等于此阈值时，它应该通过调用 Raft 的 Snapshot 来保存快照。如果 maxraftstate 为 -1，则不必进行快照。 maxraftstate 适用于 Raft 传递给 persister.SaveRaftState() 的 GOB 编码字节。

• 在初始化的时候，一定要将传入的persister保存到当前的kvServer中，用处就是获取persister.RaftStateSize()，进行快照生成判断。

• 快照的创建时机：每一次应用一个命令就要进行一次判断；

  快照的读取实际：初始化阶段；从applyCh中接收到快照命令。

• AppendEntries RPC日志内容的深拷贝问题。这个问题在2B中没有发现，是在3B中才发现的。切片默认是进行浅拷贝的，因此leader在向某个follower发送AppendEntries RPC时，在生成args到发送rpc这段时间内(这段时间没有上锁，上锁就太浪费资源了)，如果进行一次Snapshot(生成快照)，那么就会导致args的logEntries发生变化，原本存在的数据，进行一次快照后可能会进行删除，进而导致发送给follower的logEntries的某些日志的Command变为nil了。在kvServer应用这个命令的时候，进行接口的转化(op := cmd.Command.(Op))就会报错。

• 每次通过Snapshot命令读取快照时(即不是初始化调用)，需要调用CondInstallSnapshot来处理日志和状态同步，不调用的话，后续节点就会同步失败，至于为什么初始化阶段不需要调用，因为raft初始化的一段时间内，也会调用CondInstallSnapshot函数进行初始化。

其实3B部分不是很难写，但是离谱的地方是它会暴露我们前面raft部分的一些问题，因此碰到问题了，我们要都看debug日志进行查找问题。

代码

saveSnapshot

//保存快照
func (kv *KVServer) saveSnapshot(logIndex int) {
    if kv.maxraftstate == -1 || kv.persister.RaftStateSize() < kv.maxraftstate {
        return
    }

    //生成快照数据
    w := new(bytes.Buffer)
    e := labgob.NewEncoder(w)
    if err := e.Encode(kv.data); err != nil {
        panic(err)
    }
    if err := e.Encode(kv.lastApplies); err != nil {
        panic(err)
    }
    data := w.Bytes()
    kv.rf.Snapshot(logIndex, data)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

该函数用于生成并保存快照：分为三步处理：

• 快照生成判断。如果maxraftstate不为-1，且当前raft的日志数量大于等于maxraftstate，就进行一次快照生成；
• 生成快照数据。数据就是两种：服务端的data数据、用于线性化语义的lastApplies数据结构；
• 调用Snapshot生成一次快照。具体逻辑就不列出了，是2D部分的代码，主要就是：删除多余日志、保存快照内容、修改raft状态。

readPersist

//读取快照
//两处调用：初始化阶段；收到Snapshot命令，即接收了leader的Snapshot
func (kv *KVServer) readPersist(isInit bool, snapshotTerm, snapshotIndex int, data []byte) {
    if data == nil || len(data) < 1 {
        return
    }
    //只要不是初始化调用，即如果收到一个Snapshot命令，就要执行该函数
    if !isInit {
        res := kv.rf.CondInstallSnapshot(snapshotTerm, snapshotIndex, data)
        if !res {
            log.Panicln("kv read persist err in CondInstallSnapshot!")
            return
        }
    }
    //对数据进行同步
    r := bytes.NewBuffer(data)
    d := labgob.NewDecoder(r)
    var kvData map[string]string
    var lastApplies map[int64]int64

    if d.Decode(&kvData) != nil ||
    d.Decode(&lastApplies) != nil {
        log.Fatal("kv read persist err!")
    } else {
        kv.data = kvData
        kv.lastApplies = lastApplies
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

该函数用于读取快照内容，两处调用：初始化阶段；从applyCh中收到Snapshot命令，即接收了leader的Snapshot。

主要逻辑分为两步：

• 判断是否是初始化，如果不是，就要调用CondInstallSnapshot来处理日志和状态同步，不调用的话，后续节点就会同步失败；至于为什么初始化阶段不需要调用，因为raft初始化的一段时间内，也会调用CondInstallSnapshot函数进行初始化；
• 从快照中获取服务端的data数据、用于线性化语义的lastApplies数据结构。

调用时机

func StartKVServer(servers []*labrpc.ClientEnd, me int, persister *raft.Persister, maxraftstate int) *KVServer {
    ...
    kv.maxraftstate = maxraftstate
    kv.persister = persister

    // You may need initialization code here.
    kv.lastApplies = make(map[int64]int64)
    kv.data = make(map[string]string)

    kv.stopCh = make(chan struct{})
    //读取快照
    kv.readPersist(true, 0, 0, kv.persister.ReadSnapshot())
    ...
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

每一次server初始化时，就要保存传入的persister和maxraftstate，以及根据snapshot读取快照的内容。

//应用每一条命令
func (kv *KVServer) handleApplyCh() {
	for {
		select {
		case <-kv.stopCh:
			DPrintf("get from stopCh,server-%v stop!", kv.me)
			return
		case cmd := <-kv.applyCh:
			//处理快照命令，读取快照的内容
			if cmd.SnapshotValid {
				DPrintf("%v get install sn,%v %v", kv.me, cmd.SnapshotIndex, cmd.SnapshotTerm)
				kv.lock("waitApplyCh_sn")
				kv.readPersist(false, cmd.SnapshotTerm, cmd.SnapshotIndex, cmd.Snapshot)
				kv.unlock("waitApplyCh_sn")
				continue
			}
			//处理普通命令
			...
			} else {
				kv.unlock("handleApplyCh")
				panic("unknown method " + op.Method)
			}

			DPrintf("apply op: cmdId:%d, op: %+v, data:%v", cmdIdx, op, kv.data[op.Key])
			//每应用一条命令，就判断是否进行持久化
			kv.saveSnapshot(cmdIdx)

			kv.unlock("handleApplyCh")
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

在applyCh的处理中，如果接收到一条快照命令，就要调用readPersist函数进行一次快照的读取；

每应用完一条普通命令，就要调用saveSnapshot函数判断是否进行一次快照生成。

测试结果