流程图
到了这个lab5才算是真正看清除了整个lab的样子, 之前还一直纳闷lab2好像没什么用…
这个系统的核心思想是 分而治之。通过将整个键空间划分为多个分片(Shard),并将这些分片分配给不同的、可独立运行的服务器组(Shard Group),系统能够实现:
- 水平扩展: 吞吐量随着服务器组数量的增加而(近乎)线性增长。
- 故障隔离: 一个服务器组的故障不会影响其他分片的数据服务。
- **灵活管理: 可以动态地调整分片与服务器组的映射关系,以应对负载变化、机器增减等情况。
- 并发操作: 可以同时处理不同服务器组的kv数据
个人理解:
- 这个kvsrv就是lab2实现的那个client和Server, 配置(主要用于记录分片和ShardGrp的映射)以KV形式(value就是配置)存在Server中, 可以通过clerk(其实就是client)去访问或更新配置
- 一个ShardGrp就是一个raft集群, 一个ShardGrp可以存储一个或多个分片, 一个ShardGrp对应于若干个Server, 通过一个client访问
- 有三个client:
- shardkv1/client.go : 这个是模拟真正的用户
- shardkv1/shardgrp/client.go : 用于访问某个分片组的client
- kvsrv1/client.go: 用于访问配置的
- 访问流程:
- 首先shardkv1/client.go的Get/Put去访问shardkv1/shardcfg.go, 去得到key对应的分片, 进而得到对应的Server组
- 创建一个shardkv1/shardgrp/client.go 用于访问该分片组
- 在shardkv1/shardgrp/server.go中存真正的kv数据, 获取到数据后再原路返回
lab5A
这个难度是hard, 其实主要是理解上的难度, 搞明白每块代码要干什么就好了
下面来逐步完成实验要求:
Task1: 通过TestInitQuery5A
在shardctrler/shardctrler.go 中实现这两个方法 :
InitConfig方法接收测试人员以shardcfg.ShardConfig的形式传递给它的第一个配置。InitConfig 应该将配置存储在实验室 2 的kvsrv实例中。
Query方法返回当前配置;它应该从kvsrv读取先前由InitConfig存储在那里的配置 。
实现InitConfig和Query,并将配置存储在kvsrv中。代码通过第一个测试后,就大功告成了。请注意,此任务不需要任何 shardgrp。
Hint:通过从kvsrv 存储和读取初始配置来实现InitConfig和Query:使用ShardCtrler.IKVClerk的 Get / Put方法与kvsrv通信,使用ShardConfig的String方法将ShardConfig转换为可以传递给Put 的字符串,并使用shardcfg.FromString()函数将字符串转换为ShardConfig。
个人理解:
- 就是用kv的形式,去存储配置信息, 用Get获取配置, 用put初始化或更新配置
相关代码如下:
const (configKey = "configKey" // 用于存储配置的键
)func (sck *ShardCtrler) InitConfig(cfg *shardcfg.ShardConfig) {// 您的代码在这里configValue := cfg.String() // 将配置转换为字符串sck.IKVClerk.Put(configKey, configValue, 0)
}// 返回当前配置
func (sck *ShardCtrler) Query() *shardcfg.ShardConfig {// 您的代码在这里configValue, _, err := sck.IKVClerk.Get(configKey)if err != rpc.OK || configValue == "" {return nil}// 调用shardcfg.FromString解析配置字符串config := shardcfg.FromString(configValue)if config == nil {return nil}return config
}
Task2: 通过 TestStaticOneShardGroup5A
在shardkv1/shardgrp/server.go 中 实现一个初始版本的shardgrp,并在shardkv1/shardgrp/client.go中实现一个相应的 clerk ,代码来自你的实验 4 中的kvraft解决方案。 在shardkv1/client.go 中实现一个 clerk ,它使用Query方法查找指定键对应的 shardgrp,然后与该 shardgrp 进行通信。当你的代码通过Static测试后,就完成了 。
Hint:从kvraft client.go 和 server.go复制Put和Get 的代码,以及从kvraft 复制所需的任何其他代码。
Hint:shardkv1/client.go 中的代码为整个系统提供了Put / Get文员:它通过调用Query方法找出哪个 shardgrp 持有所需密钥的分片,然后与持有该分片的 shardgrp 对话。
Hint:实现 shardkv1/client.go,包括其Put / Get方法。使用shardcfg.Key2Shard() 查找键对应的分片编号。测试人员将 ShardCtrler对象传递给 shardkv1/client.go中的MakeClerk 。使用Query方法检索当前配置 。
Hint:要从 shardgrp 中提交/获取密钥,shardkv 管理员应该通过调用shardgrp.MakeClerk为该 shardgrp 创建一个 shardgrp 管理员,并传入配置中找到的服务器以及 shardkv 管理员的ck.clnt。使用 ShardConfig中的GidServers()方法获取分片所属的组。
Hint:当回复可能丢失时,shardkv1/client.go的 Put 方法必须返回ErrMaybe ,但此 Put 方法会调用shardgrp的 Put 方法与特定的 shardgrp 进行通信。内部的 Put 方法可以通过错误来指示这种情况。
Hint:创建后,第一个 shardgrp(shardcfg.Gid1)应初始化自身以拥有所有分片。
个人理解:
-
就是在分组(每个组处理若干个分片, 每个分片交给一个组处理)的情况下, 重新实现Get和put, 当然涉及到:
- shardkv1/client.go
- shardkv1/shardgrp/client.go
- shardkv1/shardgrp/server.go
-
首先是shardkv1/shardgrp/server.go , 对于服务器端来说涉及到一个检测所属组的问题, 因为配置可能发生变化, 还有就是检测的时机问题(应该在raft达成一致后,在Server真正执行之前去检测,避免不同节点出现不一致的情况), 为了便于检测所属组,也为了同步最新的映射关系, 我在Server端也保存了shardToGid结构(在变更配置时, 通过组客户端的RPC参数传给Server), 相关代码如下:
type KVServer struct {me intdead int32 // 由 Kill() 设置rsm *rsm.RSMgid tester.Tgid // 分片组ID// 您的代码在这里mu sync.MutexkvMap map[string]*ValueVersionclientPutResults map[int64]ClientPutResult // clientId -> ClientPutReq// 存储当前负责的分片// ownedShards map[shardcfg.Tshid]boolshardToGid map[shardcfg.Tshid]tester.Tgid...... }// isThisGroup 检查给定键的分片是否属于该服务器的组 func (kv *KVServer) isThisGroupUseKey(key string) bool {kv.mu.Lock()defer kv.mu.Unlock()shard := shardcfg.Key2Shard(key)return kv.shardToGid[shard] == kv.gid }func (kv *KVServer) DoOp(req any) any {// Your code hereswitch req.(type) {case *rpc.GetArgs:// 在请求通过raft同步后,实际执行前, 再次检查分片归属,// 以防在请求提交到RSM和应用到状态机之间配置发生了变化, 导致server组应用了不再拥有的分片if !kv.isThisGroupUseKey(req.(*rpc.GetArgs).Key) {// // // // log.Printf("---server: %v---DoOp---分片不属于本组, 返回ErrWrongGroup\n", kv.me)return &rpc.GetReply{Err: rpc.ErrWrongGroup}}return kv.DoGet(req.(*rpc.GetArgs))case *rpc.PutArgs:// 在请求通过raft同步后,实际执行前, 再次检查分片归属,// 以防在请求提交到RSM和应用到状态机之间配置发生了变化, 导致server组应用了不再拥有的分片if !kv.isThisGroupUseKey(req.(*rpc.PutArgs).Key) {// // // // log.Printf("---server: %v---DoOp---分片不属于本组, 返回ErrWrongGroup\n", kv.me)return &rpc.PutReply{Err: rpc.ErrWrongGroup}}return kv.DoPut(req.(*rpc.PutArgs))case *shardrpc.FreezeShardArgs:return kv.doFreezeShard(req.(*shardrpc.FreezeShardArgs))case *shardrpc.InstallShardArgs:return kv.doInstallShard(req.(*shardrpc.InstallShardArgs))case *shardrpc.DeleteShardArgs:return kv.doDeleteShard(req.(*shardrpc.DeleteShardArgs))}return nil }
-
然后是shardkv1/shardgrp/client.go和shardkv1/client.go, 这里要注意的是由于它并不是真正的client, 因此我把clientId和RequestId的生成和更新放在了shardkv1/client.go, 在shardkv1/client.go中传给shardkv1/shardgrp/client.go即可, 还有为了避免在shardkv1/shardgrp/client.go中一直在配置变更后,在旧配置中一直尝试(一直收到rpc.ErrWrongGroup), 设置一个尝试次数上限, 达到上限就返回, 在shardkv1/client.go中重新获取更新后的配置在调用shardkv1/shardgrp/client.go的Get/Put, 相关代码如下:
-
shardkv1/client.go代码如下:
// 创建一个分片组客户端,用于与负责给定键的分片组通信 func (ck *Clerk) makeSharedClerk(key string) (shardcfg.Tnum, *shardgrp.Clerk) {// 获取负责该键的分片shareId := shardcfg.Key2Shard(key)// 读取当前配置cfg := ck.sck.Query()// 查找负责该键的组中的服务器_, servers, ok := cfg.GidServers(shareId)if !ok {// 该分片没有分配给任何组return -1, nil}return cfg.Num, shardgrp.MakeClerk(ck.clnt, servers) }// 从分片组获取一个键的值 // 您可以使用 shardcfg.Key2Shard(key) 找到负责该键的分片 // 并使用 ck.sck.Query() 读取当前配置并查找负责该键的组中的服务器 // 您可以通过调用 shardgrp.MakeClerk(ck.clnt, servers) 来创建该组的客户端 func (ck *Clerk) Get(key string) (string, rpc.Tversion, rpc.Err) {// 您将需要修改这个函数for {cfgNum, sharedClerk := ck.makeSharedClerk(key)if cfgNum == -1 || sharedClerk == nil {// 该分片没有分配给任何组,等待并重试// // log.Printf("makeSharedClerk: shard not assigned, retrying...")continue}value, version, err := sharedClerk.Get(key)if err == rpc.OK {// log.Printf("Clerk: %v---Get---成功获取值: <key:%s, value:%s, version:%d>\n", ck.me, key, value, version)return value, version, rpc.OK} else if err == rpc.ErrWrongGroup {continue} else {return "", 0, err}} }// 将一个键值对放入分片组 func (ck *Clerk) Put(key string, value string, version rpc.Tversion) rpc.Err {// 您将需要修改这个函数ck.mu.Lock()ck.requestId++clientId := ck.mereqId := ck.requestIdck.mu.Unlock()for {cfgNum, sharedClerk := ck.makeSharedClerk(key)if cfgNum == -1 || sharedClerk == nil {// 该分片没有分配给任何组,等待并重试continue}err := sharedClerk.Put(key, value, version, clientId, reqId)if err == rpc.OK {return rpc.OK} else if err == rpc.ErrWrongGroup {continue} else {return err}} }
-
shardkv1/shardgrp/client.go代码如下:
func (ck *Clerk) Get(key string) (string, rpc.Tversion, rpc.Err) {// Your code hereargs := rpc.GetArgs{Key: key}retryTime := len(ck.servers) * RetryTimefor {ck.mu.Lock()leader := ck.leaderck.mu.Unlock()// 先尝试leaderreply := rpc.GetReply{}ok := ck.clnt.Call(ck.servers[leader], "KVServer.Get", &args, &reply)if ok && reply.Err != rpc.ErrWrongLeader {if reply.Err != rpc.OK {return "", 0, reply.Err}return reply.Value, reply.Version, rpc.OK}ck.mu.Lock()ck.leader = (leader + 1) % len(ck.servers)ck.mu.Unlock()if !ok {retryTime--if retryTime <= 0 {return "", 0, rpc.ErrWrongGroup}}// 如果RPC失败,等待9ms后重试time.Sleep(9 * time.Millisecond)} }func (ck *Clerk) Put(key string, value string, version rpc.Tversion, clientId int64, reqId int64) rpc.Err {// Your code hereargs := rpc.PutArgs{Key: key, Value: value, Version: version}// 是否是第一次请求isFirstRequest := true// ck.mu.Lock()// ck.requestId++// args.ClientId = ck.clientId// args.ReqId = ck.requestIdargs.ClientId = clientIdargs.ReqId = reqId// ck.mu.Unlock()retryTime := len(ck.servers) * RetryTime// // //// log.Printf("***********Clerk 向 server: %v 发起Put请求: <key:%s, value:%s, version:%d>\n", ck.leader, args.Key, args.Value, args.Version)for {ck.mu.Lock()leader := ck.leaderck.mu.Unlock()reply := rpc.PutReply{}// log.Printf("***********Clerk 向 server: %v 发起Put请求: <key:%s, value:%s, version:%d>\n", ck.leader, args.Key, args.Value, args.Version)ok := ck.clnt.Call(ck.servers[leader], "KVServer.Put", &args, &reply)if ok && reply.Err != rpc.ErrWrongLeader {if reply.Err == rpc.OK {// log.Printf("***Clerk 从server: %v 收到Put结果: <key:%s, value:%s, version:%d>\n", ck.leader, args.Key, args.Value, args.Version)return rpc.OK}if reply.Err == rpc.ErrVersion && isFirstRequest {return rpc.ErrVersion}if reply.Err == rpc.ErrVersion && !isFirstRequest {return rpc.ErrMaybe}return reply.Err // 有可能返回rpc.ErrNoKey和ErrWrongGroup}ck.mu.Lock()ck.leader = (leader + 1) % len(ck.servers)ck.mu.Unlock()isFirstRequest = falseif !ok {retryTime--if retryTime <= 0 {return rpc.ErrWrongGroup}}//如果RPC失败,等待9ms后重试Stime.Sleep(9 * time.Millisecond)} }
-
然后就是实现ChangeConfigTo
这一块主要涉及:
-
shardkv1/shardcfg/shardcfg.go中ChangeConfigTo 的逻辑实现:
-
冻结旧组的kv数据并返回
-
将这部分数据安装到新组
-
在旧组中删除这部分数据
-
最后发布新配置(就是把新配置put到kvsrv的server)
-
相关代码如下:
// 由测试程序调用,要求控制器将配置从当前配置更改为新配置
// 在控制器更改配置的过程中,它可能会被另一个控制器取代
func (sck *ShardCtrler) ChangeConfigTo(new *shardcfg.ShardConfig) {// 获取当前配置oldConfig := sck.Query()if oldConfig == nil {// // log.Printf("ChangeConfigTo: failed to get current config")return}// // log.Printf("ChangeConfigTo: changing from config %d to config %d", oldConfig.Num, new.Num)// 创建一个结构来跟踪需要迁移的分片type migration struct {shard intoldGid tester.TgidnewGid tester.Tgidstate []byte}var migrations []migration// 新的shardToGid映射, 在冻结和安装的时候使用shardToGid := make(map[shardcfg.Tshid]tester.Tgid)for shard, gid := range new.Shards {shardToGid[shardcfg.Tshid(shard)] = gid}// 1. 冻结旧配置中的所有分片for shard, oldGid := range oldConfig.Shards {newGid := new.Shards[shard]if oldGid == newGid {continue // 分片未移动}// 创建一个分片组客户端client := shardgrp.MakeClerk(sck.clnt, oldConfig.Groups[oldGid])// 冻结分片kvState, _ := client.FreezeShard(shardcfg.Tshid(shard), new.Num, &shardToGid)migrations = append(migrations, migration{shard: shard,oldGid: oldGid,newGid: newGid,state: kvState,})// 添加日志:记录冻结分片的大小// log.Printf("ChangeConfigTo: frozen shard %d from group %d, state size: %d",shard, oldGid, len(kvState))}// 2. 安装新配置中的所有分片for _, m := range migrations {// 创建一个分片组客户端client := shardgrp.MakeClerk(sck.clnt, new.Groups[m.newGid])if client == nil {// // log.Printf("ChangeConfigTo: failed to create client for group %d", m.newGid)continue}// 安装分片client.InstallShard(shardcfg.Tshid(m.shard), m.state, new.Num, &shardToGid)// 添加日志:记录安装分片的大小// log.Printf("ChangeConfigTo: installed shard %d to group %d, state size: %d",m.shard, m.newGid, len(m.state))}// 3. 删除旧配置中的所有分片for _, m := range migrations {// 创建一个分片组客户端client := shardgrp.MakeClerk(sck.clnt, oldConfig.Groups[m.oldGid])if client == nil {// // log.Printf("ChangeConfigTo: failed to create client for group %d", m.oldGid)continue}// 删除分片client.DeleteShard(shardcfg.Tshid(m.shard), new.Num, &shardToGid)// // log.Printf("ChangeConfigTo: deleted shard %d from group %d", m.shard, m.oldGid)}// 4. 将新配置存储在 kvsrv 中configValue := new.String() // 将配置转换为字符串sck.IKVClerk.Put(configKey, configValue, rpc.Tversion(new.Num)-1)// // log.Printf("ChangeConfigTo: successfully published new config %d", new.Num)
}
-
在shardkv1/shardgrp/client.go中实现FreezeShard,InstallShard和DeleteShard, 相关代码如下:
func (ck *Clerk) FreezeShard(s shardcfg.Tshid, num shardcfg.Tnum, shardToGid *map[shardcfg.Tshid]tester.Tgid) ([]byte, rpc.Err) {// Your code here// log.Printf("======== Clerk.FreezeShard 开始: shard=%d, num=%d ========", s, num)args := shardrpc.FreezeShardArgs{Shard: s, Num: num, ShardToGid: *shardToGid}for {ck.mu.Lock()leader := ck.leaderck.mu.Unlock()reply := shardrpc.FreezeShardReply{}ok := ck.clnt.Call(ck.servers[leader], "KVServer.FreezeShard", &args, &reply)if ok && reply.Err != rpc.ErrWrongLeader {// log.Printf("======== Clerk.FreezeShard 结束: shard=%d, num=%d, stateLen=%d, err=%v ========", s, num, len(reply.State), reply.Err)return reply.State, reply.Err}ck.mu.Lock()ck.leader = (leader + 1) % len(ck.servers)ck.mu.Unlock()//如果RPC失败,等待9ms后重试time.Sleep(9 * time.Millisecond)} }func (ck *Clerk) InstallShard(s shardcfg.Tshid, state []byte, num shardcfg.Tnum, shardToGid *map[shardcfg.Tshid]tester.Tgid) rpc.Err {// Your code here// log.Printf("======== Clerk.InstallShard 开始: shard=%d, num=%d, stateLen=%d ========", s, num, len(state))args := shardrpc.InstallShardArgs{Shard: s, State: state, Num: num, ShardToGid: *shardToGid}for {ck.mu.Lock()leader := ck.leaderck.mu.Unlock()reply := shardrpc.InstallShardReply{}ok := ck.clnt.Call(ck.servers[leader], "KVServer.InstallShard", &args, &reply)if ok && reply.Err != rpc.ErrWrongLeader {// log.Printf("======== Clerk.InstallShard 结束: shard=%d, num=%d, err=%v ========", s, num, reply.Err)return reply.Err}ck.mu.Lock()ck.leader = (leader + 1) % len(ck.servers)ck.mu.Unlock()//如果RPC失败,等待9ms后重试time.Sleep(9 * time.Millisecond)} }func (ck *Clerk) DeleteShard(s shardcfg.Tshid, num shardcfg.Tnum, shardToGid *map[shardcfg.Tshid]tester.Tgid) rpc.Err {// Your code here// log.Printf("======== Clerk.DeleteShard 开始: shard=%d, num=%d ========", s, num)args := shardrpc.DeleteShardArgs{Shard: s, Num: num, ShardToGid: *shardToGid}for {ck.mu.Lock()leader := ck.leaderck.mu.Unlock()reply := shardrpc.DeleteShardReply{}ok := ck.clnt.Call(ck.servers[leader], "KVServer.DeleteShard", &args, &reply)if ok && reply.Err != rpc.ErrWrongLeader {// log.Printf("======== Clerk.DeleteShard 结束: shard=%d, num=%d, err=%v ========", s, num, reply.Err)return reply.Err}ck.mu.Lock()ck.leader = (leader + 1) % len(ck.servers)ck.mu.Unlock()//如果RPC失败,等待9ms后重试time.Sleep(9 * time.Millisecond)} }
-
在shardkv1/shardgrp/server.go中实现FreezeShard, InstallShard, DeleteShard, 相关代码如下:
func (kv *KVServer) doFreezeShard(args *shardrpc.FreezeShardArgs) (reply *shardrpc.FreezeShardReply) {kv.mu.Lock()defer kv.mu.Unlock()reply = &shardrpc.FreezeShardReply{}// 过时请求if lastConfigNum, ok := kv.lastFreezeConfigNum[args.Shard]; ok && args.Num <= lastConfigNum {if args.Num == lastConfigNum {// 返回上次冻结的状态reply.State = kv.lastFreezeState[args.Shard]reply.Num = args.Numreply.Err = rpc.OK// log.Printf("---server: %v---doFreezeShard---重复冻结分片: %v, 配置编号: %v, 返回上次冻结的状态: %v\n", kv.me, args.Shard, args.Num, reply.State)return}// // log.Printf("---server: %v---doFreezeShard---冻结分片: %v, 配置编号: %v, 请求过时, 返回ErrOutdated\n", kv.me, args.Shard, args.Num)// returnreturn &shardrpc.FreezeShardReply{Err: rpc.OK, State: []byte{}, Num: shardcfg.Tnum(lastConfigNum)}}// 更新shardToGid映射kv.shardToGid = args.ShardToGid// 冻结分片kvMap := make(map[string]ValueVersion)// 确保kvMap不为nilif kv.kvMap == nil {kv.kvMap = make(map[string]*ValueVersion)}count := 0// 收集该分片的所有键值对for k, v := range kv.kvMap {if shardcfg.Key2Shard(k) == args.Shard {kvMap[k] = ValueVersion{Value: v.Value,Version: v.Version,}count++}}// 序列化w := new(bytes.Buffer)e := labgob.NewEncoder(w)e.Encode(kvMap)reply.State = w.Bytes()// 成功reply.Num = args.Numreply.Err = rpc.OK// 记录冻结的配置编号和状态kv.lastFreezeConfigNum[args.Shard] = args.Numkv.lastFreezeState[args.Shard] = reply.State// 添加日志:记录冻结的键值对数量// log.Printf("---server: %v (gid:%v)---doFreezeShard---成功, 冻结分片: %v, 配置编号: %v, 冻结键值对数量: %d\n", kv.me, kv.gid, args.Shard, args.Num, count)return }func (kv *KVServer) doInstallShard(args *shardrpc.InstallShardArgs) (reply *shardrpc.InstallShardReply) {kv.mu.Lock()defer kv.mu.Unlock()reply = &shardrpc.InstallShardReply{}// 过时请求if lastConfigNum, ok := kv.lastInstalledConfigNum[args.Shard]; ok && args.Num <= lastConfigNum {reply.Err = rpc.OK// log.Printf("---server: %v---doInstallShard---安装分片: %v, 配置编号: %v, 配置过时\n", kv.me, args.Shard, args.Num)return}// 更新shardToGid映射kv.shardToGid = args.ShardToGid// 反序列化if len(args.State) > 0 {r := bytes.NewBuffer(args.State)d := labgob.NewDecoder(r)var kvMap map[string]ValueVersionif d.Decode(&kvMap) != nil {// 解码错误log.Fatal("Failed to decode shard state")reply.Err = rpc.ErrWrongGroupreturn}// 安装分片数据// 确保kvMap不为nilif kv.kvMap == nil {kv.kvMap = make(map[string]*ValueVersion)}// 安装分片数据count := 0for k, v := range kvMap {kv.kvMap[k] = &ValueVersion{Value: v.Value,Version: v.Version,}count++}// 添加日志:记录安装的键值对数量// log.Printf("---server: %v (gid:%v)---doInstallShard---成功, 安装分片: %v, 配置编号: %v, 安装键值对数量: %d\n", kv.me, kv.gid, args.Shard, args.Num, count)}kv.lastInstalledConfigNum[args.Shard] = args.Numreply.Err = rpc.OK// // log.Printf("---server: %v---doInstallShard---成功, 安装分片: %v, 配置编号: %v, 安装键值对: %v\n", kv.me, args.Shard, args.Num, kvMap)return }func (kv *KVServer) doDeleteShard(args *shardrpc.DeleteShardArgs) (reply *shardrpc.DeleteShardReply) {kv.mu.Lock()defer kv.mu.Unlock()reply = &shardrpc.DeleteShardReply{}// 过时请求if lastConfigNum, ok := kv.lastDeletedConfigNum[args.Shard]; ok && args.Num <= lastConfigNum {reply.Err = rpc.OK// log.Printf("---server: %v---doDeleteShard---删除分片: %v, 配置编号: %v, 配置过时\n", kv.me, args.Shard, args.Num)return}// 更新shardToGid映射kv.shardToGid = args.ShardToGid// 删除分片数据// 确保kvMap不为nilif kv.kvMap == nil {kv.kvMap = make(map[string]*ValueVersion)}// 删除分片数据count := 0keysToDelete := []string{}for k := range kv.kvMap {if shardcfg.Key2Shard(k) == args.Shard {keysToDelete = append(keysToDelete, k)}}for _, k := range keysToDelete {delete(kv.kvMap, k)count++}kv.lastDeletedConfigNum[args.Shard] = args.Numreply.Err = rpc.OK// 如果删除了键值对,则通知 RSM 需要快照if count > 0 {kv.rsm.SnapshotIfNeeded()}// 添加日志:记录删除的键值对数量// log.Printf("---server: %v (gid:%v)---doDeleteShard---成功, 删除分片: %v, 配置编号: %v, 删除键值对数量: %d, 剩余键值对数量: %d\n", kv.me, kv.gid, args.Shard, args.Num, count, len(kv.kvMap))return }// 冻结指定的分片(即拒绝未来对该分片的 Get/Put 操作) // 并返回该分片中存储的键值对 func (kv *KVServer) FreezeShard(args *shardrpc.FreezeShardArgs, reply *shardrpc.FreezeShardReply) {// 您的代码在这里err, result := kv.rsm.Submit(args)if err == rpc.ErrWrongLeader {reply.Err = err// // log.Printf("---server: %v---FreezeShard---请求被拒绝, 领导者错误\n", kv.me)return}// rpc.ErrWrongGroup 或 rpc.ErrOutdated 在result中返回// 类型断言newReply := result.(*shardrpc.FreezeShardReply)reply.State = newReply.Statereply.Num = newReply.Numreply.Err = newReply.Err// // log.Printf("---server: %v---FreezeShard---返回结果, 配置编号: %v, Err: %v\n", kv.me, reply.Num, reply.Err) }// 为指定的分片安装提供的状态 func (kv *KVServer) InstallShard(args *shardrpc.InstallShardArgs, reply *shardrpc.InstallShardReply) {// 您的代码在这里err, result := kv.rsm.Submit(args)if err == rpc.ErrWrongLeader {reply.Err = err// // log.Printf("---server: %v---InstallShard---请求被拒绝, 领导者错误\n", kv.me)return}// rpc.ErrWrongGroup 或 rpc.ErrOutdated 在result中返回// 类型断言newReply := result.(*shardrpc.InstallShardReply)reply.Err = newReply.Err// // log.Printf("---server: %v---InstallShard---返回结果, Err: %v\n", kv.me, reply.Err) }// 删除指定的分片 func (kv *KVServer) DeleteShard(args *shardrpc.DeleteShardArgs, reply *shardrpc.DeleteShardReply) {// 您的代码在这里err, result := kv.rsm.Submit(args)if err == rpc.ErrWrongLeader {reply.Err = err// // log.Printf("---server: %v---DeleteShard---请求被拒绝, 领导者错误\n", kv.me)return}// rpc.ErrWrongGroup 或 rpc.ErrOutdated 在result中返回// 类型断言newReply := result.(*shardrpc.DeleteShardReply)reply.Err = newReply.Err// // log.Printf("---server: %v---DeleteShard---返回结果, Err: %v\n", kv.me, reply.Err) }
测试结果
=== RUN TestInitQuery5A
Test (5A): Init and Query ... (reliable network)...... Passed -- time 0.0s #peers 1 #RPCs 2 #Ops 0
--- PASS: TestInitQuery5A (0.00s)
=== RUN TestStaticOneShardGroup5A
Test (5A): one shard group ... (reliable network)...... Passed -- time 5.8s #peers 1 #RPCs 1577 #Ops 180
--- PASS: TestStaticOneShardGroup5A (5.75s)
=== RUN TestJoinBasic5A
Test (5A): a group joins... (reliable network)...... Passed -- time 8.8s #peers 1 #RPCs 8948 #Ops 180
--- PASS: TestJoinBasic5A (8.82s)
=== RUN TestDeleteBasic5A
Test (5A): delete ... (reliable network)...... Passed -- time 6.2s #peers 1 #RPCs 2510 #Ops 360
--- PASS: TestDeleteBasic5A (6.19s)
=== RUN TestJoinLeaveBasic5A
Test (5A): basic groups join/leave ... (reliable network)...... Passed -- time 9.9s #peers 1 #RPCs 7782 #Ops 240
--- PASS: TestJoinLeaveBasic5A (9.94s)
=== RUN TestManyJoinLeaveReliable5A
Test (5A): many groups join/leave ... (reliable network)...... Passed -- time 15.8s #peers 1 #RPCs 5094 #Ops 180
--- PASS: TestManyJoinLeaveReliable5A (15.76s)
=== RUN TestManyJoinLeaveUnreliable5A
Test (5A): many groups join/leave ... (unreliable network)...... Passed -- time 46.2s #peers 1 #RPCs 6846 #Ops 180
--- PASS: TestManyJoinLeaveUnreliable5A (46.15s)
=== RUN TestShutdown5A
Test (5A): shutdown ... (reliable network)...... Passed -- time 9.0s #peers 1 #RPCs 4191 #Ops 180
--- PASS: TestShutdown5A (9.00s)
=== RUN TestProgressShutdown5A
Test (5A): progress ... (reliable network)...... Passed -- time 4.8s #peers 1 #RPCs 1368 #Ops 82
--- PASS: TestProgressShutdown5A (4.83s)
=== RUN TestProgressJoin5A
Test (5A): progress ... (reliable network)...... Passed -- time 11.2s #peers 1 #RPCs 4472 #Ops 456
--- PASS: TestProgressJoin5A (11.23s)
=== RUN TestOneConcurrentClerkReliable5A
Test (5A): one concurrent clerk reliable... (reliable network)...... Passed -- time 20.0s #peers 1 #RPCs 20055 #Ops 4222
--- PASS: TestOneConcurrentClerkReliable5A (20.02s)
=== RUN TestManyConcurrentClerkReliable5A
Test (5A): many concurrent clerks reliable... (reliable network)...... Passed -- time 20.9s #peers 1 #RPCs 27396 #Ops 5372
--- PASS: TestManyConcurrentClerkReliable5A (20.87s)
=== RUN TestOneConcurrentClerkUnreliable5A
Test (5A): one concurrent clerk unreliable ... (unreliable network)...... Passed -- time 20.1s #peers 1 #RPCs 3915 #Ops 176
--- PASS: TestOneConcurrentClerkUnreliable5A (20.06s)
=== RUN TestManyConcurrentClerkUnreliable5A
Test (5A): many concurrent clerks unreliable... (unreliable network)...... Passed -- time 20.6s #peers 1 #RPCs 17661 #Ops 2290
--- PASS: TestManyConcurrentClerkUnreliable5A (20.62s)
PASS
ok 6.5840/shardkv1 199.246s......===== 测试统计摘要 =====
总测试次数: 30
成功次数: 27
失败次数: 3
成功率: 90%
总耗时: 6019秒
平均每次测试耗时: 200秒
===== 测试结束 =====
比官方的快40多秒
不足之处是在进行多次测试时, 仍有约1/10的概率不能通过TestOneConcurrentClerkReliable5A和TestOneConcurrentClerkUnreliable5A, 原因是不满足线性一致性… 暂时就这样了, 以后有时间再改进
lab5B
允许控制器完成前一个控制器开始的重新配置的一个好方法是保持两个配置:一个当前配置和一个下一个配置,都存储在控制器的kvsrv中。当控制器开始重新配置时,它存储下一个配置。一旦控制器完成重新配置,它使下一个配置成为当前配置。修改InitController以首先检查是否有存储的下一个配置,其配置号比当前配置高,如果有,完成重新配置到下一个配置所需的分片移动。
这个lab比较简单
实现如下:
- 修改ChangeConfigTo: 在修改配置的时候,先设置nextConfig,在修改完成之后再将当前的Config设置为新配置,然后把nextConfig设为空。
- 实现initController: 先去查找当前配置和nextConfig,如果nextConfig的num更新,就重新调用一下changeConfigTo。
// 在 A 部分,此方法不需要执行任何操作
// 在 B 和 C 部分,此方法实现恢复功能
func (sck *ShardCtrler) InitController() {old := sck.Query()next := sck.QueryNext()// log.Println("InitController: old =", old, "next =", next)if next == nil || old == nil || old.Num >= next.Num {return}sck.ChangeConfigTo(next)
}func (sck *ShardCtrler) PutNextConfig(info string) {_, version, err := sck.IKVClerk.Get(nextConfigKey)if err == rpc.ErrNoKey {sck.IKVClerk.Put(nextConfigKey, info, 0)} else {sck.IKVClerk.Put(nextConfigKey, info, version)}
}
测试结果:
=== RUN TestJoinLeave5B
Test (5B): Join/leave while a shardgrp is down... (reliable network)...... Passed -- time 7.7s #peers 1 #RPCs 1586 #Ops 120
--- PASS: TestJoinLeave5B (7.73s)
=== RUN TestRecoverCtrler5B
Test (5B): recover controller ... (reliable network)...... Passed -- time 25.3s #peers 1 #RPCs 6658 #Ops 360
--- PASS: TestRecoverCtrler5B (25.30s)
PASS
ok 6.5840/shardkv1 33.028s
未完待续…