MIT6.5840-2023-Lab4B: Sharded K/V Service-Sharded K/V Server

实验内容

实现一个分片 k/v 存储系统，分片指如所有以“a”开头的键可能是一个分片，所有以 “b”开头的键可能是另一个分片。每个副本组仅处理几个分片的 Put、Append 操作，实现并行操作，系统总吞吐量（单位时间的放入和获取）与组的数量成正比。

分片 k/v 存储系统由多个副本组和一个分片控制器组成，分片控制器管理配置信息，决定哪个副本组为哪个分片服务。
某些组的负载可能远高于其他组，实现在多个副本组之间转移分片，以达到负载均衡。同时可能会添加新的副本组以增加容量，或者现有的副本组可能会脱机以进行修复，因此必须移动分片以继续满足要求。

主要挑战是处理重新配置，即将分片重新分配给副本组。在单个副本组中，所有组成员必须就客户端的 Put/Append/Get 请求在进行重新配置时达成一致。如 Put 可能与重新配置同时到达，重新配置导致副本组不再对 Put 的 key 对应的分片负责。所以组中的所有副本必须就 Put 发生在重新配置之前还是之后达成一致。若在重新配置之前，则 Put 应生效，并且该分片的新所有者需要看到生效效果；若在重新配置之后，Put 将不会生效，客户必须重新请求该 key 的新所有者。

推荐的方法是让每个副本组使用 Raft 不仅记录请求的ID，还记录重新配置的ID。需要确保任何时间最多只有一个副本组为一个分片提供服务。

重新配置还涉及到副本组之间的交互，如在配置 10 中，组 G1 负责分片 S1，在配置 11 中，组 G2 负责分片 S1；在 10 到 11 的重新配置期间，G1 和 G2 必须使用 RPC 将分片 S1 的内容（键/值对）从 G1 移动到 G2。

实验环境

OS：WSL-Ubuntu-18.04
golang：go1.17.6 linux/amd64

Part B: Sharded Key/Value Server

Impl：shardkv/client.go, shardkv/common.go, and shardkv/server.go

实现一个分片 k/v 存储系统，每个副本组仅处理其所负责的分片中的键，并支持 Get、Put、Append 操作。利用 PartA 中实现的 ShardCtrler 维护配置信息。

No-credit challenge exercises

Garbage collection of state
当副本组失去对分片的所有权时，该副本组应消除其数据库中对应的key，从而避免浪费。但是，这在分片迁移上会有一些问题。假设我们有两个组G1和G2，并且有一个新的配置C——将Shard S从G1移至G2。如果G1转换为配置C时从其数据库中删除了S中的所有key，则G2在尝试移动到C时如何获取S的数据？

Client requests during configuration changes
处理配置更改的最简单方法是禁止所有客户端操作，直到过渡完成。虽然从概念上讲简单，但这种方法在生产级系统中是不可行的。每当将机器带入或取出时，它会导致所有客户暂停。
即使配置转换仍在进行中，副本组仍然能够使用新配置已经join的分片。

addConfig()

定时检测是否有更新的配置，若当前正在更新配置，则等待更新完成再继续检测；否则向raft提交命令，等待达成共识然后执行。

func (kv *ShardKV) addConfig() {for kv.killed() == false {if _, isLeader := kv.rf.GetState(); !isLeader {time.Sleep(100 * time.Millisecond)continue}kv.mu.Lock()isAllServing := truefor _, shard := range kv.shards {if shard.State != Serving {isAllServing = falsebreak}}kv.mu.Unlock()if isAllServing {config := kv.manager.Query(kv.config.Num + 1) // next configif config.Num == kv.config.Num+1 {command := Command{CommandType: AddConfig,Data:        config,}reply := CommonReply{}kv.startCommand(command, &reply)}}time.Sleep(100 * time.Millisecond)}
}

insertShard()

定时检测是否需要从其他副本组pull shard，向pulling状态分片所在的副本组发送GetShards-rpc，得到分片信息后，向raft提交插入分片的命令，达成共识后执行。

func (kv *ShardKV) insertShard() {for kv.killed() == false {if _, isLeader := kv.rf.GetState(); !isLeader {time.Sleep(100 * time.Millisecond)continue}kv.mu.Lock()GID2ShardIds := kv.getLastGID2ShardIds(Pulling) // gid in last configwg := &sync.WaitGroup{}wg.Add(len(GID2ShardIds))for gid, shardIds := range GID2ShardIds {configNum, servers := kv.config.Num, kv.lastConfig.Groups[gid]go func(gid int, shardIds []int, configNum int, servers []string) {defer wg.Done()// get pulling shards in other groupfor _, server := range servers {args := PullShardArgs{GID:       gid,ShardIds:  shardIds,ConfigNum: configNum,}reply := PullShardReply{}srv := kv.make_end(server)ok := srv.Call("ShardKV.GetShards", &args, &reply)if ok && reply.Err == OK {reply.ConfigNum = configNumcommand := Command{CommandType: InsertShard,Data:        reply,}kv.startCommand(command, &CommonReply{})}}}(gid, shardIds, configNum, servers)}kv.mu.Unlock()wg.Wait()time.Sleep(100 * time.Millisecond)}
}

adjustGCing()——Garbage collection of state

定时检测是否有GCing状态分片，这种状态表示分片已经复制到副本组中。若有，则向这些分片之前所在的副本组发送DeleteShards-rpc，删除其中BePulling状态分片；然后向raft提交将GCing状态修改为Serving状态的命令。

func (kv *ShardKV) adjustGCing() {for kv.killed() == false {if _, isLeader := kv.rf.GetState(); !isLeader {time.Sleep(100 * time.Millisecond)continue}kv.mu.Lock()GID2ShardIds := kv.getLastGID2ShardIds(GCing)wg := &sync.WaitGroup{}wg.Add(len(GID2ShardIds))for gid, shardIds := range GID2ShardIds {configNum, servers := kv.config.Num, kv.lastConfig.Groups[gid]go func(gid int, shardIds []int, configNum int, servers []string) {defer wg.Done()// remove gcing shards in other groupfor _, server := range servers {args := RemoveShardArgs{ShardIds:  shardIds,ConfigNum: configNum,}reply := RemoveShardReply{}srv := kv.make_end(server)ok := srv.Call("ShardKV.DeleteShards", &args, &reply)if ok && reply.Err == OK {args := AdjustShardArgs{ShardIds:  shardIds,ConfigNum: configNum,}command := Command{CommandType: AdjustState,Data:        args,}kv.startCommand(command, &CommonReply{})}}}(gid, shardIds, configNum, servers)}kv.mu.Unlock()wg.Wait()time.Sleep(100 * time.Millisecond)}
}

adjustBePulling()

主要解决快照记录的时间点不同，不同服务器回退到了不同的状态。定时检测状态为BePulling的分片，若配置落后于被插入分片的副本组（即原先分片状态为Pulling的副本组，而快照回退正常，所有分片正常迁移，状态均为Serving，而不是GCing）的配置，则可以判断持久化状态回退失误，将BePulling状态转为Serving。

func (kv *ShardKV) adjustBePulling() {for kv.killed() == false {if _, isLeader := kv.rf.GetState(); !isLeader {time.Sleep(100 * time.Millisecond)continue}kv.mu.Lock()GID2ShardIds := kv.getNowGID2ShardIds(BePulling)wg := &sync.WaitGroup{}wg.Add(len(GID2ShardIds))for gid, shardIds := range GID2ShardIds {configNum, servers := kv.config.Num, kv.lastConfig.Groups[gid]go func(gid int, shardIds []int, configNum int, servers []string) {defer wg.Done()// get bepulling shards in now groupfor _, server := range servers {args := CheckArgs{ShardIds:  shardIds,ConfigNum: configNum,}reply := CheckReply{}srv := kv.make_end(server)ok := srv.Call("ShardKV.CheckShards", &args, &reply)if ok && reply.Err == OK {args := AdjustShardArgs{ShardIds:  shardIds,ConfigNum: configNum,}command := Command{CommandType: AdjustState,Data:        args,}kv.startCommand(command, &CommonReply{})}}}(gid, shardIds, configNum, servers)}kv.mu.Unlock()wg.Wait()time.Sleep(100 * time.Millisecond)}
}