完成笔记
本次作业分成三个清晰的步骤, 分别是构建一个 KV Server, 然后使用这个 KV Server 实现一个分布式锁, 最后是在网络不稳定, 存在丢包与抱错的场景下, 如何保证这个分布式锁的可用性.
KV 服务器的搭建
- 每一个客户端使用
Clerk与服务端通信, 使用 RPC 的方式通信. - 客户端会通过 RPC 向服务端发送两种操作,
Put(key, value, version), 与Get(key). - 服务端中会在内存中使用 MAP 为每一个 Key 记录一个
(Value, Version)的 tuple. - RPC 中的
Get(key)只会返回rpc.ErrNoKey这一种报错, 但是Put(key, value, version)会返回rpc.ErrVersion或者rpc.ErrMaybe.
服务端的实现
在一个 KV 服务器中, 服务端是用来存储数据的, 服务端主要有三部分需要实现, 如何存储数据, 客户端 RPC 调用服务端的 Get() 函数, 服务端如何返回数据, 客户端 PRC 调用服务端的 Put() 函数, 服务端如何执行.
- 服务端的结构体如下:
type KVServer struct {mu sync.Mutex// Your definitions here.// data structure to store key-value pairs and their versionsversions map[string]rpc.Tversiondata map[string]string
}
- 当服务端收到客户端的
Get()请求时, 会执行下面的Get()函数.
// Get returns the value and version for args.Key, if args.Key
// exists. Otherwise, Get returns ErrNoKey.
func (kv *KVServer) Get(args *rpc.GetArgs, reply *rpc.GetReply) {// Your code here.kv.mu.Lock()defer kv.mu.Unlock()value, exists := kv.data[args.Key]// if key does not exist, return ErrNoKeyif !exists {reply.Err = rpc.ErrNoKeyreturn}// key exists, return value and versionreply.Value = valuereply.Version = kv.versions[args.Key]reply.Err = rpc.OK
}
- 当服务端收到客户端的
Put()请求时. 需要执行下面的Put()函数.
func (kv *KVServer) Put(args *rpc.PutArgs, reply *rpc.PutReply) {// Your code here.kv.mu.Lock()defer kv.mu.Unlock()currentVersion, exists := kv.versions[args.Key]// key does not existif !exists {// if version is 0, add the key and value, version becomes 1if args.Version == 0 {// install the valuekv.data[args.Key] = args.Valuekv.versions[args.Key] = 1 // initial version is 1reply.Err = rpc.OK} else {// key doesn't exist and version is not 0reply.Err = rpc.ErrNoKey}return}// key exists, check versionif args.Version != currentVersion {reply.Err = rpc.ErrVersionreturn}// update value and versionkv.data[args.Key] = args.Valuekv.versions[args.Key] = currentVersion + 1reply.Err = rpc.OK
}
客户端的实现
客户端的本质是
一些疑问
- 为什么在
oneClient新建一个客户端的时候, 使用的lk := MakeLock(ck, "l")是相同的呢, 也就是说在 Server 中使用相同的 Key 表示使用的 Value, 那么是不是多个 Clients 使用的都是一个分布式锁呢?
一些想法
- 分布式锁测试的步骤是, 首先初始化一个 TestKV 的测试框架, 主要是读取测试环境的一些参数, 然后是批量生成大量的 Clients 客户端.
- 启动与创建多个 Clients 客户端的方式是通过创建一个
runClient协程来实现的, 如果是多个并发, 会异步的创建与启动多个runClient协程. 每一个协程会调用函数MakeClerk()来创建一个客户端, 也就是下面的函数
func (ts *TestKV) MakeClerk() kvtest.IKVClerk {clnt := ts.Config.MakeClient()ck := MakeClerk(clnt, tester.ServerName(tester.GRP0, 0))return &kvtest.TestClerk{ck, clnt}
}
这个函数的服务端 Server 是相同的, TestClerk 绑定了客户端与
- 创建客户端之后, 每一个协程都会调用函数
oneClient并且使用这个客户端. 在调用函数oneClient的时候, 会使用函数lk := MakeLock(ck, "l")来创建一个锁, 也就是说每个协程对应一个客户端, 使用这个客户端在 Server 上创建一个锁, 但是!!! 所有的协程在创建锁的时候, key 都是相同的, 也就是 "l", 这也就是说明, 所有的客户端都使用相同的分布式锁. - 由于并发与网络连接不稳定导致分布式锁重复获取锁的问题: 当 Clinet_A 与 Client_B 在
Acquire()阶段, 发现Get(lk.key)的应答都为空, 也就是没有人持有锁, 两个都想获取锁, 此时, Clinet_A 和 Client_B 都会执行 Put 操作, 如果 Client_A 在 Put 时, 请求丢失, 或者 Put 失败, 此时 Cilent_B 发起了 Put, 并且成功, 然后 Client_A 重试, 此时由于 Client_B 已经写入了 Server, 因此 Client_A 继续失败(由于重复写入), 但是返回值是rpc.ErrMaybe, 此时, 我们再通过 Get 进行一次检查, 发现应答成功, 并且已经上锁, Client_A 误认为是自己上锁了, 其实是 Client_B 上的锁. 这种并发与网络不稳定导致分布式锁的获取错误.
分布式锁的搭建
错误示例
func (lk *Lock) Acquire() {// Your code herefor {// try to acquire the lock by checking if the lock is currently unlockedvalue, latest_version, err := lk.ck.Get(lk.l)if err == rpc.ErrNoKey || (value == "unlocked" && latest_version == lk.lock_version) {err = lk.ck.Put(lk.l, "locked", lk.lock_version)if err == rpc.OK {lk.lock_version += 1return} else {log.Printf("failed to acquire lock: %v", err)}}// log.Printf("lock is currently locked, retrying..., value: %v, err: %v", value, err)}
}
正确答案如下:
func (lk *Lock) Acquire() {// Your code herefor {// try to acquire the lock by checking if the lock is currently unlockedvalue, latest_version, err := lk.ck.Get(lk.l)if err == rpc.ErrNoKey {value = "unlocked"latest_version = 0}if value == "unlocked" {err = lk.ck.Put(lk.l, "locked", latest_version)if err == rpc.OK {lk.lock_version = latest_version + 1return}}}
}
func (lk *Lock) Acquire() {// Your code herefor {val, ver, err := lk.ck.Get(lk.l)if err == rpc.ErrNoKey {val = ""ver = 0}// If we already hold the lock (previous Put succeeded but response was lost)if val == lk.id {return}// If the lock is free, try to acquire itif val == "" {err = lk.ck.Put(lk.l, lk.id, ver)if err == rpc.OK {return}// ErrMaybe: might have succeeded, loop back and check with Get// ErrVersion: someone else got it, loop back}time.Sleep(100 * time.Millisecond)}
}func (lk *Lock) Release() {// Your code herefor {val, ver, err := lk.ck.Get(lk.l)if err == rpc.OK && val != lk.id {// Lock is not held by us - our release already succeededreturn}if err == rpc.OK && val == lk.id {// We still hold the lock, try to release iterr = lk.ck.Put(lk.l, "", ver)if err == rpc.OK {return}// ErrMaybe: might have succeeded, loop back and check with Get}time.Sleep(100 * time.Millisecond)}
}