济南seo网站排名关键词优化,景区网站建设方案 费用,大兴网站建设价格,html网页制作模板代码免费实验内容
实现一个分片 k/v 存储系统#xff0c;分片指如所有以“a”开头的键可能是一个分片#xff0c;所有以 “b”开头的键可能是另一个分片。每个副本组仅处理几个分片的 Put、Append 操作#xff0c;实现并行操作#xff0c;系统总吞吐量#xff08;单位时间的放入和…实验内容
实现一个分片 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。
实验环境
OSWSL-Ubuntu-18.04 golanggo1.17.6 linux/amd64
Part B: Sharded Key/Value Server
Implshardkv/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.Num1 {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)}
}
