网站建设公司前台住房与城乡建设网上办事大厅

实验目标：基于raft日志复制算法实现的线性一致性kv存储引擎。

线性一致性：

1. 所有的读操作都能够读取到最近一次写操作的结果。
2. 所有节点（或者进程）在同一时刻，看到的数据都是相同的。

简而言之，线性一致性保证所有的操作在系统中是按照其提交的顺序进行执行的，因此，在任何时刻，对于任意一个节点或进程所提出的读操作，都应该返回最新提交的写操作的结果。

当一个节点或进程发出一个写操作之后，其他节点或进程必须立即且无条件地能够看到这个写操作的结果，以保证线性一致性的实现。

需要注意的是，线性一致性通常会带来更高的延迟和更大的开销，因为系统需要通过各种机制来保证不同节点或进程之间的数据同步。

想法：raft保障的日志都是一致性的，所以如果我们把操作通过raft来传播的话，通过applyChan我们就可以得到2）所有节点（或者进程）在同一时刻，看到的数据都是相同的。然后如果我们两个连续的读要读到最近一次写的内容，如果没有别的client并发操作的话。那么在日志看来就是

前缀日志 + 读操作 + 读操作那么根据raft日志的顺序，我们可以发现所有的读操作都能够读取到最近一次写操作的结果。但是这个东西有一个限制，就是我们对于append和put操作只能执行一次，为啥呢。

有人可能认为，只要写请求是幂等的，那重复执行多次也是可以满足线性一致性的，实际上则不然。考虑这样一个例子：对于一个仅支持 put 和 get 接口的 raftKV 系统，其每个请求都具有幂等性。设 x 的初始值为 0，此时有两个并发客户端，客户端 1 执行 put(x,1)，客户端 2 执行 get(x) 再执行 put(x,2)，问（客户端 2 读到的值，x 的最终值）是多少。对于线性一致的系统，答案可以是 (0,1)，(0,2) 或 (1,2)。然而，如果客户端 1 执行 put 请求时发生了上段描述的情况，然后客户端 2 读到 x 的值为 1 并将 x 置为了 2，最后客户端 1 超时重试且再次将 x 置为 1。对于这种场景，答案是 (1,1)，这就违背了线性一致性。归根究底还是由于幂等的 put(x,1) 请求在状态机上执行了两次，有两个 LZ 点。因此，即使写请求的业务语义能够保证幂等，不进行额外的处理让其重复执行多次也会破坏线性一致性。

所以我们要线性一致性的话，我们就是要保障put/append操作可以commit的多次，但是只能应用在kv层一次，所以我们可以对每个客户端记录下每个客户端最后的commitIdx.

lab3

client端

我们要找到server集群的leader，要用一个commitIdx来区分请求，使得请求有一个序列，因为client没有并发，所以这个最大值就是最后一个请求，我们需要把这个对于每个客户端记录一下

func (ck *Clerk) PutAppend(key string, value string, op string) {// You will have to modify this function.args := PutAppendArgs{Key:          key,Value:        value,Op:           op,LeaderId:     ck.LeaderId,CommandId_PA: ck.commandId,ClientId:     ck.clientId,}DPrintf("Node{%v} start appendput{%v} value{%v} op{%v}", ck.clientId, key, args.Value, op)for {var reply PutAppendReplyif !ck.servers[ck.LeaderId].Call("KVServer.PutAppend", &args, &reply) || reply.Err == ErrWrongLeader || reply.Err == ErrTimeOut {ck.LeaderId = (ck.LeaderId + 1) % int64(len(ck.servers))// log.Println(1)continue}ck.commandId++DPrintf("Node{%v} appendput sucess", ck.clientId)return}
}

server端

因为start在raft底层已经上了锁，足够保障并发安全，不用加kv层的锁，不如可能会死锁。然后raft层存的是log of operations。同时read也要记录，因为我们根据的是日志的顺序来决定执行的顺序的。最后那个delete为了减少内存占用。实际上不用担心ch有东西被delete掉了，因为kv有重试机制，同时更重要的是kv applier chan传递时上了锁，delete也上了锁。同时一个apply协程也保障了顺序性。也只有一个在channel等待，同时我们做了一个buffer的channel，根据前面的描述，buffer的大小不影响正确性，我们可以用，因为最多一个在buffer里（因为我们用的是一个aplier协程），但我们不用buffer的话，可能在start后，得到notify_chan的锁就获得不到，因为我们start比较快就commit阻塞在applier了。

func (kv *KVServer) PutAppend(args *PutAppendArgs, reply *PutAppendReply) {// Your code here.var op_type intif args.Op == "Append" {op_type = APPEND} else {op_type = PUT}make_op := Op{ClientId:  args.ClientId,OpType:    op_type,Key:       args.Key,Value:     args.Value,CommandId: args.CommandId_PA,}DPrintf("Node{%v} may be ........", kv.me)kv.mu.Lock()if kv.isdupicate(make_op.ClientId, make_op.CommandId) {lastreply := kv.clientsInformation[make_op.ClientId].Last_repyreply.Err = lastreply.Errkv.mu.Unlock()return}kv.mu.Unlock()DPrintf("Node{%v} stall here", kv.me)index, _, is_leader := kv.rf.Start(make_op)if !is_leader {reply.Err = ErrWrongLeaderreturn}kv.mu.Lock()ch := kv.newChannel(index)kv.mu.Unlock()select {case rpc := <-ch:reply.Err = rpc.Errif rpc.Err == OK {// log.Printf("reply will Node{%v} the return reply optype{%v} commidex{%v}, lastapplied{%v} key{%v} value{%v}", kv.me, make_op.OpType, make_op.CommandId, kv.lastApplied, make_op.Key, make_op.Value)}case <-time.After(500 * time.Millisecond):reply.Err = ErrTimeOut}DPrintf("Node{%v} here will reply{%v}", kv.me, reply)kv.mu.Lock()kv.Delete(index)kv.mu.Unlock()}

snapshot要存的是kv storage和clientInformation。先recover snapshot在kv storage和clientinforamtion ,然后raft重放log，接着作用于clientinformation和storage上，不需要持久化kv_storage和client_information。因为下层的raft的log和snapshot已经保障了，重放时会恢复。

func (kv *KVServer) applier() {for !kv.killed() {DPrintf("Node{%v} here applier", kv.me)select {case message := <-kv.applyCh:{DPrintf("Node{%v} try to applymessage{%v}", kv.me, message)if message.CommandValid {kv.mu.Lock()if message.CommandIndex <= kv.lastApplied {DPrintf("Node{%v} command{%v} is less than kv lastapply{%v}", kv.me, message.CommandIndex, kv.lastApplied)kv.mu.Unlock()continue}kv.lastApplied = message.CommandIndexreply := new(CommandReply)make_op := message.Command.(Op)DPrintf("Node{%v} makeop.........{%v}", kv.me, make_op)if make_op.OpType != GET && kv.isdupicate(make_op.ClientId, make_op.CommandId) {DPrintf("Node{%v} is duplicate", kv.me)reply.Err = kv.clientsInformation[make_op.ClientId].Last_repy.Err} else {reply = kv.applylogtoState(make_op)if make_op.OpType != GET {kv.clientsInformation[make_op.ClientId] = ClientInfo{Last_commandId: make_op.CommandId, Last_repy: *reply}}}// kv.rf.Persist(kv.clientsInformation, kv.lastApplied)current_term, is_Leader := kv.rf.GetState()if is_Leader && message.CommandTerm == current_term {DPrintf("Node{%v} get state", kv.me)notify_chan := kv.newChannel(message.CommandIndex)notify_chan <- reply// if kv.maxraftstate != -1 {// 	kv.rf.Persist(kv.clientsInformation, kv.storage)// }DPrintf("reply to notify chan{%v}", reply)} else {DPrintf("Node{%v} is not leader", kv.me)}// 3Bif kv.rf.ShouldSnap(kv.maxraftstate, message.SnapshotIndex) {DPrintf("Node{%v} start to snapshot index{%v}", kv.me, message.CommandIndex)kv.snapMake(message.CommandIndex)}kv.mu.Unlock()} else if message.SnapshotValid {kv.mu.Lock()kv.storeSnapshot(message.Snapshot)kv.lastApplied = message.SnapshotIndexkv.mu.Unlock()} else {panic(fmt.Sprintf("Valid message{%v}", message))}}}}
}