6.584-Lab5B

6.584-Lab5B

• Reference Code
• Reference Blog
• Homework
• Myself Code

Sharded Key/Value Service 梗概

这个图是我从上面参考blog中拿来的，觉得做的不错，借助这张图来讲解一下需要一个什么样的 Service。

ShardCtrler Client：
接收来自客户发出的命令（作业中是test程序/ShardKV Client/Server），四种命令Join/Leave/Move/Query，具体含义看Homework。
shardctrler client 接收到命令后通过 RPC 交给自己下面的 shardctrler cluster/server。

ShardCtrler cluster/server：
接收到来自 shardctrler client 发送包含具体命令的 RPC 后，封装命令并交付给自己下面的 Raft 来实现分布式的一致性。
接着从相应的通道接收 Raft 提交（Appliy）的命令，执行接收到的命令Join/Leave/Move/Query，并生成新的 Configuration。

ShardKV Client：
接收来自客户发出的命令（实际生产中的用户，作业中是test程序），作业中实现的是KV Service 所以命令只包含Put/Get/Append三种命令。
将接收到的命令通过 RPC 交给自己下面的 ShardKV cluster/server。
ShardKV cluster/server：
除了接收来自 shardKV Client 发送来的关于 KV 的三种命令外，还需要定时向 ShardCtrler Client 发送 Query 命令来获取最新的配置，用以得知 Shards 被哪些 group 包含。
Reference Code实现了以下命令：

Shard 的状态有 Serving/GCing/Pulling/BePulling，分别表示为正在服务、垃圾清理、从别的 group 拉取 data、被别的 group 拉取。

• applyConfiguration：得到最新的 Configuration 后应用到本地，更新每个 Shard 的状态。每个 shardKV server 会检查新的 Config 中的每个 shard 所处的 Group，如果这个 shard 现在在自己这但新的 Config 中显示在别的 group 中则会将该 shard 标记为 BePulling；反之现在不在自己这但新的 Config 显示在自己这则标记为Pulling。
• applyInsertShards：将标记为BePulling的 shard 插入到应在的 group 中（在新的 Config 会显示），插入完成后原来拥有这个 shard 的 group 会将这个 shard 标记为 GCing，后续进行垃圾清理。
• applyDeleteShards：将被标记为GCing的 shard 清理掉（初始化为一个 NewShard）。
• applyEmptyShards：当下层的 Raft 进入新的 Term 后，没有任何“命令”作为 log 发送到 Raft的情况下，发送一个空的命令到 Raft，让其作为一个 log 进行占位。

ShardKV cluster/server 发送上述的7种命令到下层的 Raft 层来实现分布式的一致性。等这些命令在下层 Raft “转”一圈后，通过相应的通道接收从 Raft 发来的已经应用（Apply）的命令后再采取相应逻辑执行。

部分代码讲解&踩的坑

以下是我在阅读Reference Code时记录的一些疑问：

Q1： normal command 即 put、get、append是由 Client 发送的；但Configuration/InsertShards/DeleteShards/EmptyShards是从哪里发送的？
A：在server.go启动函数StartServer中，会设置监视器Monitor来定时去看是否需要发送这些命令到函数Execute去进一步执行。

Q2：
Server 中处理normal command即Put/Get/Append命令的函数applyOperation中，判断 Server 是否能处理这个 Command 的函数 canServe 中为什么分片处于垃圾清理状态仍可以 shardstatus == GCing 处理这个 Command ？
为什么在server.go的applyInsertShards中，从别的 Shard 加载完 ShardData 后要把状态从Pulling改为GCing？刚拉取完 新的信息就要进行垃圾清理进行清空？
A：
在applyInsertShards中确实是把Pulling之后的 Shard 状态设置为了GCing，但是在后面的垃圾清理函数gcAction中，先查找状态为GCing的 Shard 都位于那些 Gid 中，我们看一下这个查找函数getShardIDsByStatus：可以看到是在旧的 Config 中找到这个 Shard 所处的 Gid。所以传给垃圾清理函数gcAction中的 Gid 就是要删除 shard 的 Gid。

举个例子：在 Gid_1 中数组 shard[2].satus = Pulling，Gid_2 中数组shard[2].satus = BePulling，表明 shard2 在 Gid_2 中。此时 Gid_1 已经拉取了 shard2 的 data，Gid_1 中数组shard[2].satus = GCing，在垃圾清理的时候，找到 shard2 在旧的 Config 的位置也就是 Gid_2，把 Gid_2 中的 Shard2 = NewShard 初始化为一个空的shard。后面在垃圾清理函数gcAction中遇到状态为GCing的 shard 时，会再次改回Serving。
所以虽然 Gid_1 中 shard[2].satus = GCing，却找的是要清理的 Shard2 的正确的位置即 Gid_2。所以 Gid_1 中 shard[2].satus = GCing表示的不是清理 Gid1 中的 shard2，而表示的是要清理 shard2 之前带过的 Gid2 中的 shard2。

所以状态为 GCing 的 shard 是刚接收完新的 data，后面也不会被垃圾清理，当然可以处理Put/Get/Append命令。

Q3： server.go中为什么ShardKVa中需要设置一个 lastConfig 字段？
A：在server.go的函数migrationAction()中，可能给出了答案，在执行分片迁移的时候，此时currentConfig已经是下一个最新的 configuration 了，但是分片迁移的任务还是需要位于上个 configuration 的 server 去执行的，才能变为下个 configuration 也就是currentConfig

Q4： 在server.go的函数migrationAction()中并发时搭配匿名函数的传参方式不同的区别？
A：参考GPT。

1. 简单来说，goroutine并发的匿名函数直接使用外部变量（闭包）的话，goroutine中使用的外部变量会被goroutine外部改变：

• i 是主 goroutine 的变量，而 goroutines 是在独立的线程中执行的。
• 当 goroutines 执行时，i 的值可能已经被主循环改变，因此打印的结果可能是多个相同的值或不可预测的值。

func main() {for i := 0; i < 5; i++ {go func() {fmt.Println(i) // 闭包变量}()}time.Sleep(time.Second)
}// OutPUt:
4
4
4
4
4

2. 将变量作为参数传递给匿名函数，可以避免闭包问题：

• i 的值在每次迭代时被复制并传递给匿名函数，因此每个 goroutine 都有自己独立的副本。
• 结果是确定的。

func main() {for i := 0; i < 5; i++ {go func(n int) {fmt.Println(n) // 使用传递的参数}(i) // 显式传递 i}time.Sleep(time.Second)
}//OutPut:
0
1
2
3
4

3. 通过闭包捕获局部变量的副本
   在每次循环中创建一个新的局部变量，并让匿名函数捕获该变量：

• 类似于将变量显式传递，n 是每次循环的局部变量，匿名函数捕获的是该变量的副本。
• 结果也是确定的。

func main() {for i := 0; i < 5; i++ {n := i // 创建新的局部变量go func() {fmt.Println(n)}()}time.Sleep(time.Second)
}OutPut:
0
1
2
3
4

踩的坑/需要注意的点

在 Reference Code 的函数applier中，比较了从 Raft 层接收到命令的开始的Term message.CommandTerm 与当前 Raft Leader 所处的 Term currentTerm。但是我们之前实现的 Raft 传出的命令是没有CommandTerm的字段的，秉持少改动底层实现的 Raft 的原则，我在结构体 CommandReply 中设置了 AppliedTerm 字段，将 reply 先传回到函数Execute中，在Execute中保存的有开始传入 Raft 层的 TermstartTerm，这这里进行比较。

---

在 Reference Code 中的 Raft 层实现了 GetRaftStateSize方法用来获取已经持久化的 RaftState 的大小，我在结构体ShardKV中添加了字段persister *raft.Persister用来保存启动函数StartServer传入的persisiter，直接调用官方在 Raft 层中的Persister.RaftStateSize获取已经持久化的 RaftState 的大小。我还像之前参考【香草美人】实现 KVRaft 那样设置了 0.95 的阈值。

---

运行提交命令后，出现了这样的错误：

在运行到某个地方的时候，底层 Raft 实现的 AppendEntries函数出现了 PrevLogIndex < lastIncludedIndex的情况，这样两者相减求相对下标的时候，会出现负数的下标索引。我去查看【香草美人】的代码后发现，他更新了代码，判断了这个情况（不知道之前是漏看了，还是后来他更新的）。

---

在方法checkEntryInCurrentTermAction添加空 Shard 的时候，要判断一下 Raft Leader 当前的 Term 是否等于最后一条日志log的 Term，若不相等则添加一个空的 Shard。
在【参考代码】中，是在 Raft 层实现了函数rf.HasLogInCurrentTerm()来判断，我底层 Raft 是没有这函数的，不想动 Raft 层的代码，我想rf.currentTerm == rf.log[len(log)-1].Term直接判断的，但是发现在其他文件中访问 Raft 层的话，只能访问 Raft 层实现的方法，不能访问 Raft 层的变量rf.log[]、rf.currentTerm，我也就只好也在 Raft 层实现了【参考代码】的函数rf.HasLogInCurrentTerm().

结果

倒腾了好久终于通过官方测试了。这种分布式程序调试太难了，每个发生顺序都不确定，在一堆的输出日志找 bug 太难了 QAQ。