概述
lab2中实现了raft协议,本lab将在raft之上实现一个可容错的k/v存储服务,第一部分是实现一个不带日志压缩的版本,第二部分是实现日志压缩。时间原因我只完成了第一部分。
设计思路
如上图,lab2实现了raft协议,本lab将实现kvserver。每个raft都关联一个kvserver,Clerks发送Put(), Append(), Get() RPC给leader服务器中的kvserver,kvserver收到请求后将操作打包成Log Entry提交给raft,然后阻塞等待raft将这个Entry拷贝到其它server,当Log Entry被拷贝到大部分的server后,leader 的raft会通知kvserver(raft往管道中塞comitted Entry,kvserver通过读这个管道获取通知),kvserver执行命令,然后响应Clerk。
Clerk
客户端通过Clerk发送请求,来看下Clerk代码:
type Clerk struct {
servers []*labrpc.ClientEnd
// You will have to modify this struct.
lastLeader int
cid int64
seq int
}
func (ck *Clerk) Get(key string) string {
// You will have to modify this function.
// 参数: 要读的key, 当前clerk的id, 请求序列号
getArgs := GetArgs{Key: key, Cid:ck.cid, Seq:ck.seq}
reply := GetReply{}
for {
doneCh := make(chan bool, 1)
go func() {
//发送Get() RPC
ok := ck.servers[ck.lastLeader].Call("KVServer.Get", &getArgs, &reply)
doneCh <- ok
}()
select {
case <-time.After(600 * time.Millisecond):
DPrintf("clerk(%d) retry PutAppend after timeout\n", ck.cid)
continue
case ok := <- doneCh:
//收到响应后,并且是leader返回的,那么说明这个命令已经执行了
if ok && reply.WrongLeader != WrongLeader {
//请求序列号加1
ck.seq++
return reply.Value
}
}
//换一个server重试
ck.lastLeader++
ck.lastLeader %= len(ck.servers)
}
return ""
}
这里只给出了Get()的代码,Put()和Append()类似,发送KVServer.Get给一个server,如果这个server不是leader,换一个server重试。直到发给真正的leader,并且leader将这个命令拷贝到大部分其它server后,然后成功执行该命令,Clerk.Get()才会返回。
KVServer
再来看下服务端的代码,KVServer处理Clerk的RPC请求:
type KVServer struct {
mu sync.Mutex
me int
rf *raft.Raft
applyCh chan raft.ApplyMsg
maxraftstate int // snapshot if log grows this big
// Your definitions here.
// 保存键值对
db map[string]string
latestReplies map[int64]*LatestReply
notify map[int]chan struct{}
}
func (kv *KVServer) Get(args *GetArgs, reply *GetReply) {
// Your code here.
if _, isLeader := kv.rf.GetState(); !isLeader {
reply.WrongLeader = WrongLeader
reply.Err = ""
return
}
// 防止重复请求
kv.mu.Lock()
if latestReply, ok := kv.latestReplies[args.Cid]; ok && args.Seq <= latestReply.Seq {
reply.WrongLeader = IsLeader
reply.Value = latestReply.Reply.Value
reply.Err = latestReply.Reply.Err
kv.mu.Unlock()
return
}
kv.mu.Unlock()
command := Op{Operation:"Get", Key:args.Key, Cid:args.Cid, Seq:args.Seq}
index, term, _ := kv.rf.Start(command)
// 阻塞等待结果
kv.mu.Lock()
ch := make(chan struct{})
kv.notify[index] = ch
kv.mu.Unlock()
select {
case <-ch:
curTerm, isLeader := kv.rf.GetState()
DPrintf("%v got notify at index %v, isLeader = %v\n", kv.me, index, isLeader)
if !isLeader || curTerm != term {
reply.WrongLeader = WrongLeader
reply.Err = ""
} else {
reply.WrongLeader = IsLeader
kv.mu.Lock()
if value, ok := kv.db[args.Key]; ok {
reply.Value = value
reply.Err = OK
} else {
reply.Err = ErrNoKey
}
kv.mu.Unlock()
}
}
}
KVServer.db用于保存键值对。
KVServer.Get()首先判断自己是不是leader,如果不是leader,直接返回,这样Clerk好重试其它server。如果是leader,先在缓存中找,看这个请求是否已经执行过了。
因为可能出现这么一种情况:如果leader commit一个Entry后立即奔溃了,那么Clerk就收不到响应,那么Clerk会将这个请求发给新的leader,新的leader收到请求后如果不做任何措施,将会二次commit该Log Entry,对于Put()和Append()请求执行两次是不正确的,所以需要一个办法防止一个请求执行两次。
可以这么做:每个Clerk都分配一个唯一的cid,每个请求分配一个唯一的序列号seq,每成功一个请求,该序列号加一。服务端记录每个客户端cid最近一次apply的请