网站首页 返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

raft理论与实践[6]-lab3a-基于raft构建分布式容错kv服务

2020-06-10 02:37:37 admin 664

## 准备工作

* 阅读[raft论文](https://links.jianshu.com/go?to=https%3A%2F%2Flink.zhihu.com%2F%3Ftarget%3Dhttp%253A%2F%2Fnil.csail.mit.edu%2F6.824%2F2017%2Fpapers%2Fraft-extended.pdf)

* 阅读[raft理论与实践[1]-理论篇](https://links.jianshu.com/go?to=https%3A%2F%2Fzhuanlan.zhihu.com%2Fp%2F102023809)

* 阅读[raft理论与实践[2]-lab2a](https://links.jianshu.com/go?to=https%3A%2F%2Fzhuanlan.zhihu.com%2Fp%2F102948740)

* 阅读[raft理论与实践[3]-lab2a讲解](https://links.jianshu.com/go?to=https%3A%2F%2Fzhuanlan.zhihu.com%2Fp%2F103223270)

* 阅读[raft理论与实践[4]-lab2b日志复制](https://links.jianshu.com/go?to=https%3A%2F%2Fzhuanlan.zhihu.com%2Fp%2F103386687)

* 阅读[raft理论与实践[5]-lab2c日志复制](https://links.jianshu.com/go?to=https%3A%2F%2Fzhuanlan.zhihu.com%2Fp%2F103473049)

* 阅读[模拟RPC远程过程调用](https://links.jianshu.com/go?to=https%3A%2F%2Flink.zhihu.com%2F%3Ftarget%3Dhttps%253A%2F%2Fdreamerjonson.com%2F2019%2F12%2F25%2Fgolang-109-lab-simulate-rpc%2F)

## 前言

* 在之前的文章中，我们实现了raft算法的基本框架

* 在本实验中，我们将基于raft算法实现分布式容错的kv服务器

* 客户端用于交互raft服务器

* kvraft/client.go文件用于书写我们的客户端代码，调用Clerk的Get/Put/Append方法为系统提供强一致性的保证

* 这里的强一致性指的是，如果我们一个一个的调用（而不是并发）Clerk的Get/Put/Append方法，那么我们的系统就好像是只有一个raft服务器存在一样，并且调用是序列的，即后面的调用比前面的调用后执行

* 对于并发调用，最终状态可能难以预料，但是必须与这些方法按某种顺序序列化后执行一次的结果相同

* 如果调用在时间上重叠，则这些调用是并发的。例如，如果客户端X调用Clerk.Put(),同时客户端Y调用Clerk.Append()

* 同时，后面的方法在执行之前，必须保证已经观察到前面所有方法执行后的状态（技术上叫做线性化（linearizability））

* 强一致性保证对应用程序很方便，因为这意味着所有客户端都看到相同的最新状态

* 对于单个服务器，强一致性相对简单。多台的副本服务器却相对困难，因为所有服务器必须为并发请求选择相同的执行顺序，并且必须避免使用最新状态来回复客户端

## 本服务实现的功能

* 本服务支持3种基本的操作，`Put(key, value)`, `Append(key, arg)`, and `Get(key)`

* 维护着一个简单的键/值对数据库

* `Put(key, value)`将数据库中特定key的值绑定为value

* `Append(key, arg)`添加，将arg与key对应。如果key的值不存在，则其行为类似于Put

* `Get(key)` 获取当前key的值

* 在本实验中，我们将实现服务具体的功能，而不必担心Raft log日志会无限增长

## 实验思路

* 对lab2中的raft服务器架构进行封装，封装上一些数据库、数据库快照、并会处理log的具体执行逻辑。

* 对于数据库执行的Get/Put/Append方法都对其进行序列化并放入到lab2 raft的体系中，这样就能保证这些方法的一致性

## 获取源代码

* 假设读者已经阅读了准备工作中的一系列文章

* 在此基础上我们增加了本实验的基本框架kvraft文件以及linearizability文件

* 读者需要在kvraft文件夹中，实验本实验的具体功能

* 获取实验代码如下

## 客户端

* Clerk结构体存储了所有raft服务器的客户端`servers []*labrpc.ClientEnd`，因此我们可以通过Clerk结构体与所有raft服务器通信

* 我们需要为Clerk结构体实现`Put(key, value)`, `Append(key, arg)`, `Get(key)`方法

* Clerk结构体是我们连接raft服务器的桥梁

* 注意Clerk必须将方法发送到当前的leader节点中，由于其可能并不会知道哪一个节点为leader，因此需要重试。但是记住保存上一个leader的id会加快这一过程，因为leader在稳定的系统里面是不会变的。

* 客户端必须要等到此操作不仅为commit，而且已经被完全应用后，才能够返回，这才能够保证下次get操作能够得到最新的

* 需要注意的是，如果raft服务器出现了分区，可能会陷入一直等待，直到分区消失

### 补充Clerk

* leader记录最后一个leader的序号

* seq 记录rpc的序号

* id记录客户端的唯一id

### 补充Get方法

* Get方法会遍历访问每一个raft服务，直到找到leader

* 调用时会陷入堵塞，等待rpc方法返回

* 设置有超时时间，一旦超时，会重新发送

* 为了保证Get方法到的数据是准确最新的，也必须要将其加入到raft算法中

* 客户端必须要等到此操作不仅为commit，而且已经被完全应用后，才能够返回，这才能够保证下次get操作能够得到最新的。

ck.leader %= cnt
        done := make(chan bool, 1)
        go func() {
            ok := ck.servers[ck.leader].Call("KVServer.Get", args, reply)
            done <- ok
        }()
        select {
        case <-time.After(200 * time.Millisecond): // rpc timeout: 200ms
            ck.leader++
            continue
        case ok := <-done:
            if ok && !reply.WrongLeader {
                ck.seq++
                if reply.Err == OK {
                    return reply.Value
                }
                return ""
            }
            ck.leader++
        }
    }
    
    return ""

}</pre>

### 补充Append和Put方法

* 调用同一个PutAppend方法，但是最后一个参数用于标识具体的操作

* 和Get方法相似，遍历访问每一个raft服务，直到找到leader

* 调用时会陷入堵塞，等待rpc方法返回

* 设置有超时时间，一旦超时，会重新发送

* 客户端必须要等到此操作不仅为commit，而且已经被完全应用后，才能够返回，这才能够保证下次get操作能够得到最新的。

ck.leader %= cnt
        done := make(chan bool, 1)
        go func() {
            ok := ck.servers[ck.leader].Call("KVServer.PutAppend", args, reply)
            done <- ok
        }()
        select {
        case <-time.After(200 * time.Millisecond): // rpc timeout: 200ms
            ck.leader++
            continue
        case ok := <-done:
            if ok && !reply.WrongLeader && reply.Err == OK {
                ck.seq++
                return
            }
            ck.leader++
        }
    }

}</pre>

### Server

* kvraft/server.go文件用于书写我们的客户端代码

* KVServer结构是对于之前书写的raft架构的封装

* `applyCh chan raft.ApplyMsg` 用于状态虚拟机应用coommit log，执行操作

* `db map[string]string` 是模拟的一个数据库

* `notifyChs map[int]chan struct{}` commandID => notify chan 状态虚拟机应用此command后，会通知此通道

* `duplicate map[int64]*LatestReply` 检测重复请求

## 完成PutAppend、Get方法

* 下面以PutAppend为例，Get方法类似

* 检测当前是否leader状态

* 检测是否重复请求

* 将此command通过`rf.Start(cmd)` 放入raft中

* `select`等待直到ch被激活，即command index被此kv服务器应用

* ch被激活后,需要再次检测当前节点是否为leader

* 如果不是,说明leader更换,立即返回错误，这时由于如果不再是leader，那么虽然此kv服务器应用了此command index，但不一定是相同的command

* 这个时候会堵塞直到序号为commandIndex的命令被应用，但是，如果leader更换，此commandIndex的命令不一定就是我们的当前的命令

* 但是完全有可能新的leader已经应用了此状态，我们这时候虽然仍然返回错误，希望客户端重试，这是由于操作是幂等的并且重复操作无影响。

* 优化方案是为command指定一个唯一的标识，这样就能够明确此特定操作是否被应用

DPrintf ("[%d]: leader %d receive rpc: PutAppend(%q => (%q,%q), (%d-%d).\n", kv.me, kv.me,
        args.Op, args.Key, args.Value, args.ClientID, args.SeqNo)
    
    kv.mu.Lock()
    // duplicate put/append request
    if dup, ok := kv.duplicate[args.ClientID]; ok {
        // filter duplicate
        if args.SeqNo <= dup.Seq {
            kv.mu.Unlock()
            reply.WrongLeader = false
            reply.Err = OK
            return
        }
    }
    
    // new request
    cmd := Op{Key: args.Key, Value: args.Value, Op: args.Op, ClientID: args.ClientID, SeqNo: args.SeqNo}
    index, term, _ := kv.rf.Start(cmd)
    ch := make(chan struct{})
    kv.notifyChs[index] = ch
    kv.mu.Unlock()
    
    reply.WrongLeader = false
    reply.Err = OK
    
    // wait for Raft to complete agreement
    select {
    case <-ch:
        // lose leadership
        curTerm, isLeader := kv.rf.GetState()
        if !isLeader || term != curTerm {
            reply.WrongLeader = true
            reply.Err = ""
            return
        }
    case <-kv.shutdownCh:
        return
    }

}</pre>

## 完成对于log的应用操作

* `&lt;-kv.applyCh` 是当log成为commit状态时，状态机对于log的应用操作

* 本系列构建的为kv-raft服务，根据不同的服务其应用操作的方式不同

* 下面的操作是简单的操作内存map数据库

* 同时，将最后一个操作记录下来，避免同一个log应用了两次。

<pre style="font-family: Consolas, "Liberation Mono", Menlo, Courier,
monospace; -webkit-font-smoothing: antialiased; margin: 0px; padding:
0.88889em; max-width: 100%; font-size: 0.9em; word-break: break-all; white-
space: pre-line; font-style: normal; font-variant: normal; font-weight:
normal; font-stretch: normal; line-height: 1.45; color: rgb(86, 116, 130);
word-wrap: normal; background: rgb(246, 246, 246); overflow: auto; letter-
spacing: normal; orphans: 2; text-align: left; text-indent: 0px; text-
transform: none; widows: 2; word-spacing: 0px; -webkit-text-stroke-width: 0px;
text-decoration-style: initial; text-decoration-color: initial;">func (kv
*KVServer) applyDaemon() {  
for {  
select {  
case msg, ok := <-kv.applyCh:  
if ok {  
// have client's request? must filter duplicate command  
if msg.Command != nil {  
cmd := msg.Command.(Op)  
kv.mu.Lock()  
if dup, ok := kv.duplicate[cmd.ClientID]; !ok || dup.Seq < cmd.SeqNo {  
switch cmd.Op {  
case "Get":  
kv.duplicate[cmd.ClientID] = &LatestReply{Seq: cmd.SeqNo,  
Reply: GetReply{Value: kv.db[cmd.Key],}}  
case "Put":  
kv.db[cmd.Key] = cmd.Value  
kv.duplicate[cmd.ClientID] = &LatestReply{Seq: cmd.SeqNo,}  
case "Append":  
kv.db[cmd.Key] += cmd.Value  
kv.duplicate[cmd.ClientID] = &LatestReply{Seq: cmd.SeqNo,}  
default:  
DPrintf("[%d]: server %d receive invalid cmd: %v\n", kv.me, kv.me, cmd)  
panic("invalid command operation")  
}  
if ok {  
DPrintf("[%d]: server %d apply index: %d, cmd: %v (client: %d, dup seq: %d <
%d)\n",  
kv.me, kv.me, msg.CommandIndex, cmd, cmd.ClientID, dup.Seq, cmd.SeqNo)  
}  
}  
// notify channel  
if notifyCh, ok := kv.notifyChs[msg.CommandIndex]; ok && notifyCh != nil {  
close(notifyCh)  
delete(kv.notifyChs, msg.CommandIndex)  
}  
kv.mu.Unlock()  
}  
}  
}  
}  
}</pre>

## 测试

* 注意，如果上面的测试出现错误也不一定是程序本身的问题，可能是单个进程运行多个测试程序带来的影响

* 同时，我们可以运行多次避免偶然的影响

* 因此，如果出现了这种情况，我们可以为单个测试程序独立的运行n次，保证正确性，下面是每10个测试程序独立运行，运行n次的脚本

sleep 40
    
    if grep -nr "FAIL.*raft.*" res; then
        echo "fail"
    fi

done</pre>

## 总结

* 在本实验中，我们封装了lab2a raft框架实现了容错的kv服务

* 如果出现了问题，需要仔细查看log，思考问题出现的原因

* 下一个实验中，我们将实现日志的压缩

## 参考资料

* [项目链接](https://links.jianshu.com/go?to=https%3A%2F%2Flink.zhihu.com%2F%3Ftarget%3Dhttps%253A%2F%2Fgithub.com%2Fdreamerjackson%2Fgolang-deep-distributed-lab)

* [lab3实验介绍][http://nil.csail.mit.edu/6.824/2020/labs/lab-kvraft.html](https://links.jianshu.com/go?to=https%3A%2F%2Flink.zhihu.com%2F%3Ftarget%3Dhttp%253A%2F%2Fnil.csail.mit.edu%2F6.824%2F2020%2Flabs%2Flab-kvraft.html))

* 阅读[raft论文](https://links.jianshu.com/go?to=https%3A%2F%2Flink.zhihu.com%2F%3Ftarget%3Dhttp%253A%2F%2Fnil.csail.mit.edu%2F6.824%2F2017%2Fpapers%2Fraft-extended.pdf)

* 阅读[raft理论与实践[1]-理论篇](https://links.jianshu.com/go?to=https%3A%2F%2Fzhuanlan.zhihu.com%2Fp%2F102023809)

* 阅读[raft理论与实践[2]-lab2a](https://links.jianshu.com/go?to=https%3A%2F%2Fzhuanlan.zhihu.com%2Fp%2F102948740)

* 阅读[raft理论与实践[3]-lab2a讲解](https://links.jianshu.com/go?to=https%3A%2F%2Fzhuanlan.zhihu.com%2Fp%2F103223270)

* 阅读[raft理论与实践[4]-lab2b日志复制](https://links.jianshu.com/go?to=https%3A%2F%2Fzhuanlan.zhihu.com%2Fp%2F103386687)

* 阅读[raft理论与实践[5]-lab2c日志复制](https://links.jianshu.com/go?to=https%3A%2F%2Fzhuanlan.zhihu.com%2Fp%2F103473049)

* 阅读[模拟RPC远程过程调用](https://links.jianshu.com/go?to=https%3A%2F%2Flink.zhihu.com%2F%3Ftarget%3Dhttps%253A%2F%2Fdreamerjonson.com%2F2019%2F12%2F25%2Fgolang-109-lab-simulate-rpc%2F)

转载文章，原文链接： raft理论与实践[6]-lab3a-基于raft构建分布式容错kv服务

关键字词raft laba

分享到：

上一篇：golang快速入门[5.1]-go语言是如何运行的-链接器

下一篇：golang-raft算法理论与实践

如需留言，请登录，没有账号？请注册

0 条评论 0 人参与

网站首页 返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

raft理论与实践[6]-lab3a-基于raft构建分布式容错kv服务

zhou 2年前

zhou 2年前

zhou 2年前

zhou 2年前

武云 2年前

网站首页 返回列表 像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

raft理论与实践[6]-lab3a-基于raft构建分布式容错kv服务

zhou 2年前

zhou 2年前

zhou 2年前

zhou 2年前

武云 2年前

网站首页返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。