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。
- 简单来说，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
- 将变量作为参数传递给匿名函数，可以避免闭包问题：
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
- 通过闭包捕获局部变量的副本
在每次循环中创建一个新的局部变量，并让匿名函数捕获该变量：
- 类似于将变量显式传递，
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 层接收到命令的开始的Termmessage.CommandTerm与当前 Raft Leader 所处的 TermcurrentTerm。但是我们之前实现的 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。