Lab2： Key/Value Server
找不到实习，投简历全部在初筛，要么就是被挂，回来狠狠沉淀。
KV Server很久之前就写完了，当时不是很理解，看了黑马点评关于Redis的部分后，现在有了更多的理解，但是对于go的写法，熟练度不够，不过现在应该不影响我对代码结构的设计了。
Key/value server
这个实现， 就是在可靠或不可靠网络下实现一个简单的KV server。这里我降Reliable Network和Dropped Messages合并了，因为后者其实就是因为网络原因需要不断的进行重试，当时具体完成的细节和步骤不太记得了，将就看吧。
首先在客户端实现Get、Put方法，有几个事情需要考虑，一个数据的版本，一个是返回值的问题。
Client
目前的创建客户端如下，其实就是返回一个调用的对象罢了。
func MakeClerk(clnt *tester.Clnt, server string) kvtest.IKVClerk {ck := &Clerk{clnt: clnt, server: server}return ck
}
下面是一些参数的定义，就是核心就是KEY、VALUE、VERSION以及返回值
type PutArgs struct {Key stringValue stringVersion Tversion
}type PutReply struct {Err Err
}type GetArgs struct {Key string
}type GetReply struct {Value stringVersion TversionErr Err
}
Client中Get方法的实现。加入超时的判断，在规定实践内可以不断的尝试，超时直接返回NoKEY。逻辑简单易懂
func (ck *Clerk) Get(key string) (string, rpc.Tversion, rpc.Err) {// You will have to modify this function.args := rpc.GetArgs{Key: key}reply := rpc.GetReply{}const timeoutLimit = 5 * time.SecondstartTime := time.Now()ok := ck.clnt.Call(ck.server, "KVServer.Get", &args, &reply)for !ok {if time.Since(startTime) > timeoutLimit {return "", 0, rpc.ErrNoKey}time.Sleep(100 * time.Millisecond)ok = ck.clnt.Call(ck.server, "KVServer.Get", &args, &reply)}return reply.Value, reply.Version, reply.Err
}
Client中Put方法的实现。Put方法相比与GET方法要复杂一些，涉及到一个Value的Version的判断，有时候版本号错误可能是因为第一次成功，结果消息没有发送到客户端，客户端尝试之前，另外一个客户端进行修改，结果导致ErrVersion的出现，所以这里给了一个ErrMaybe，很好理解。
func (ck *Clerk) Put(key, value string, version rpc.Tversion) rpc.Err {args := rpc.PutArgs{Key: key,Value: value,Version: version,}reply := rpc.PutReply{}ok := ck.clnt.Call(ck.server, "KVServer.Put", &args, &reply)if ok {return reply.Err}const timeoutLimit = 5 * time.SecondstartTime := time.Now()for !ok {if time.Since(startTime) > timeoutLimit {return rpc.ErrNoKey}time.Sleep(100 * time.Millisecond)ok = ck.clnt.Call(ck.server, "KVServer.Put", &args, &reply)}if reply.Err == rpc.ErrVersion {return rpc.ErrMaybe} else {return reply.Err}
Server
基础的定义，以及创建一个Server的代码如下，使用两个map，来记录Value，和Version。
type KVServer struct {mu sync.Mutexkaval map[string]stringkaver map[string]rpc.Tversion
}func MakeKVServer() *KVServer {kv := &KVServer{}kv.kaval = make(map[string]string, 0)kv.kaver = make(map[string]rpc.Tversion, 0)return kv
}
Get的实现如下，先把kv锁住，防止其他进程修改，返回对应的数据即可。
func (kv *KVServer) Get(args *rpc.GetArgs, reply *rpc.GetReply) {kv.mu.Lock()defer kv.mu.Unlock()if version, ok := kv.kaver[args.Key]; ok {reply.Version = versionreply.Value = kv.kaval[args.Key]reply.Err = rpc.OKreturn}reply.Err = rpc.ErrNoKey
}
Put的实现如下。
逻辑主要是，根据参数，判断版本号是否正确，不正确返回不正确版本号，找不到这个key的话，说明数据库里并没有这个KEY，传过来的Version如果不是0，直接返回ErrNoKey。最后就是正常的情况，进行设置数据，有锁保证一致性。
func (kv *KVServer) Put(args *rpc.PutArgs, reply *rpc.PutReply) {kv.mu.Lock()defer kv.mu.Unlock()if version, ok := kv.kaver[args.Key]; ok {if version != args.Version {reply.Err = rpc.ErrVersionreturn}} else {if args.Version != 0 {reply.Err = rpc.ErrNoKeyreturn}}kv.kaval[args.Key] = args.Valuekv.kaver[args.Key]++reply.Err = rpc.OK
}
Implementing a lock using key/value clerk
在redis中有一个SETNX来当作分布式锁使用，这里也是要求实现一个锁。
原理大概就是，在获取锁的时候，保证数据Version正确，释放锁也必须是获取锁的客户端，然后如果有多个锁，每个锁需要有不同的名字来区分。后续可能会出现lease这个东西，就是锁过期时间，防止客户端挂掉，其他客户端永远获取不到这个锁。
抽象一下，Version的正确性是靠KV server来保证的。锁的结构如下：
type Lock struct {// IKVClerk is a go interface for k/v clerks: the interface hides// the specific Clerk type of ck but promises that ck supports// Put and Get. The tester passes the clerk in when calling// MakeLock().ck kvtest.IKVClerklockKeyName stringclientId string
}
下面是获取一个锁对象，没什么好说的，给一个ID，一个Name即可，这了的RandomID还是太随意了，目前够用。
func MakeLock(ck kvtest.IKVClerk, l string) *Lock {lk := &Lock{ck: ck}lk.lockKeyName = llk.clientId = kvtest.RandValue(8)return lk
}
有了锁之后，自然要进行锁的获取。正如实验中说的，代码并不多。这里需要解释一下，这里的Version，并不是直接获取的，需要测试代码获取，然后传入，防止网络不行或者说是被其他Client获取锁了，这里一直循环，得不到直接让进程睡眠，用了循环。代码如下。
func (lk *Lock) Acquire() {for {value, version, err := lk.ck.Get(lk.lockKeyName)if err == rpc.ErrNoKey || (err == rpc.OK && value == "") {ok := lk.ck.Put(lk.lockKeyName, lk.clientId, version)if ok == rpc.OK {break} else if ok == rpc.ErrMaybe {value, _, _ = lk.ck.Get(lk.lockKeyName)if value == lk.clientId {break}}}time.Sleep(10 * time.Millisecond)}
}
下面是一个释放锁的过程。
ErrMaybe的解释
- 第一次 Put（清空）可能已经成功（服务器 value=""，版本+1），但回复丢了。
- 重试 Put 用旧 version → 服务器 ErrVersion → Clerk 返回 ErrMaybe。
- 这时再 Get 检查，如果已经空了，就放心退出（否则继续重试）。
func (lk *Lock) Release() {for {value, version, _ := lk.ck.Get(lk.lockKeyName)if value == lk.clientId {ok := lk.ck.Put(lk.lockKeyName, "", version)if ok == rpc.OK {return} else if ok == rpc.ErrMaybe {value, _, _ = lk.ck.Get(lk.lockKeyName)if value == "" {break}}} else {return}time.Sleep(10 * time.Millisecond)}
}