分布式

Go, KV数据库, Raft

2023-01-15

Source

Edit

History

6.824 lab3a

简介

实现踩坑过程

首先按照自己的思路进行了基本实现，踩了许多坑。

错误1：

Test: one client (3A)

2022/12/06 17:12:56 2 server: Get RPC opf.opIndex==3 but args.OpIndex==5

原因：对于opIndex == maxOpIndex的情况，这条指令也已经在raft层中Start一次commit了，也需要读取applyCh，而我忘记了。

错误2：

Test: progress in majority (3A)

2022/12/06 18:22:20 2 server: Get RPC opf.opIndex==1 but args.OpIndex==2, opf=={1 Put 1 14}

发现我忘了一件事情：就算不是leader，也要apply来自raft的日志（已实现，用一个goroutine来定期apply）

错误3：

— FAIL: TestUnreliable3A (5.24s)
test_test.go:293: get wrong value, key 3, wanted:
x 3 0 yx 3 1 yx 3 2 yx 3 3 yx 3 4 yx 3 5 yx 3 6 yx 3 7 y
, got
x 3 0 yx 3 1 yx 3 2 yx 3 3 yx 3 4 yx 3 5 yx 3 6 yx 3 6 yx 3 7 y
test_test.go:293: get wrong value, key 1, wanted:
x 1 0 yx 1 1 yx 1 2 yx 1 3 yx 1 4 yx 1 5 yx 1 6 yx 1 7 yx 1 8 yx 1 9 yx 1 10 y
, got
x 1 0 yx 1 1 yx 1 2 yx 1 3 yx 1 4 yx 1 5 yx 1 6 yx 1 7 yx 1 8 yx 1 8 yx 1 9 yx 1 9 yx 1 10 y
test_test.go:126: failure
test_test.go:148: duplicate element x 1 8 y in Append result

当网络不稳定时，有一些指令被重复执行了。发现在server的RPC处理当中，我应该把判断OpIndex的逻辑写到Start前面，不然由于applyCh是单独的goroutine读取，在Start之后还是会重复执行提交但未回复的指令。

错误4：

Test: completion after heal (3A) …

— FAIL: TestOnePartition3A (6.21s)
1
test_test.go:539: Put did not complete

发现没有实现超时机制。设计了一个map，保存每个client对应的已applied的opf，由RPC handler尝试读取，直至超时。实现超时后仍通不过，调试发现kv.maxOpIndexs应该写在applyLogs()协程里面。修改后通过该测试样例及之前的所有样例。

错误5：

— FAIL: TestManyPartitionsOneClient3A (14.86s)
test_test.go:293: get wrong value, key 0, wanted:
x 0 0 yx 0 1 yx 0 2 yx 0 3 yx 0 4 yx 0 5 yx 0 6 yx 0 7 yx 0 8 yx 0 9 yx 0 10 yx 0 11 yx 0 12 yx 0 13 yx 0 14 yx 0 15 yx 0 16 yx 0 17 yx 0 18 yx 0 19 yx 0 20 yx 0 21 yx 0 22 yx 0 23 yx 0 24 yx 0 25 y
, got
x 0 0 yx 0 1 yx 0 2 yx 0 3 yx 0 4 yx 0 5 yx 0 6 yx 0 7 yx 0 8 yx 0 9 yx 0 10 yx 0 11 yx 0 12 yx 0 13 yx 0 14 yx 0 15 yx 0 16 yx 0 17 yx 0 18 yx 0 19 yx 0 20 yx 0 21 yx 0 22 yx 0 23 yx 0 24 yx 0 25 yx 0 25 y
test_test.go:126: failure
test_test.go:148: duplicate element x 0 25 y in Append result

在Partition测试中又出现重复写入。在applyLogs中去重后又发现：client让某个partition A apply了一个操作，当这个partition A和另一个partition B合并时，领导者变成了partition B的成员，但partition A的maxOpIndex更新，导致partition A无法执行新命令。

发现applyLogs的sleep应该删掉，否则apply太慢了。但是错误仍未解决。

继续跟踪Raft发现，旧leader向大多数节点append一个指令对应的日志成功后，在本地状态机进行了commit，并进行了apply，使得K/V层的指令执行成功。但commit之后还来不及通过心跳或appendEntries让其他节点接收到leaderCommit，就变更领导者了，由于新领导者认为这个命令还未apply，所以无法执行后续指令。这应该就是实验指南和论文中提到的领导者变更问题。为了确保我的Raft实现没问题，用网上的lab3代码跑了一下3A，确定能够通过测试，从而将问题焦点落在server的实现上。

将代码进行了进一步检查和修改，又倒腾了一下午debug，终于通过所有算例。完整实现思路以下一章为准。

完整设计思路

Client

Client向任意服务器发送请求，请求失败则换服务器重试，直到请求成功为止。每个Client以创建时的时间戳作为其ID，请求序号从1递增。

Server

server数据结构：

type KVServer struct {
    mu      sync.Mutex
    me      int
    rf      *raft.Raft
    applyCh chan raft.ApplyMsg
    dead    int32 // set by Kill()

    maxraftstate int // snapshot if log grows this big

    // Your definitions here.
    table map[string]string // the database
    // appliedLogs   string // for test
    maxOpIndexs   map[int]int // the latest operation that server has applied for each client
    clientApplied map[int]chan OpFields
}

table为存储K/V对的哈希表，maxOpIndexs保存每个客户端已经applied的最大请求编号，clientApplied用于在server成功apply一个请求的时候通知Get/PutAppend handler。

用于apply从Raft端提交的日志的go routine如下：

func (kv *KVServer) applyLogs() {
    for kv.killed() == false {
        applyMsg := <-kv.applyCh

        op, ok := applyMsg.Command.(Op)
        if !ok {
            log.Fatal(errors.New("applyMsg type error"))
        }
        opf := op.DecodeOp()

        DPrintf("%v server: applyCh receive logIdx: %v, OpIndex: %v, Key: %v, value: %v, ClientId: %v\n", kv.me, applyMsg.CommandIndex, opf.opIndex, opf.key, opf.value, opf.ckId)
        kv.mu.Lock()
        if opf.opIndex > kv.maxOpIndexs[opf.ckId] {
            kv.OperatePutAppend(opf)
            kv.maxOpIndexs[opf.ckId] = opf.opIndex
        }
        kv.mu.Unlock()

        select {
        case kv.clientApplied[opf.ckId] <- opf:
        default:
        }
    }
}

当Raft提交上来的日志编号大于当前应用的最大编号时，server才将其执行，以避免重复。然后，server借助kv.clientApplied中的channel告知Get/PutAppend Handler请求已被提交。

这里的一个问题是为什么满足opf.opIndex > kv.maxOpIndexs[opf.ckId]的请求就可以执行，而不需要opf.opIndex == kv.maxOpIndexs[opf.ckId]+1。这是我之前调试时一直犯的错误。我举个例子，假设当前对于某个客户端，MaxOpIndex为1，而opf.opIndex为3，那么此时有可能2还没有执行完，正在处于重试阶段，而用户就又输入了第三条指令。因此上述语句只需要保证不重复执行就行，而无需按照OpIndex顺序执行。

Get/PutAppend Handler执行过程如下：

调用Start
通过Start的返回值判断是否为leader，若不为leader直接返回
判断是否args.OpIndex == kv.maxOpIndexs[args.ClientId]，若满足，则是论文中所提到的这条指令已经应用但未回复客户端的情况。此时直接回复即可。
在clientApplied中创建一个channel用于唤醒
使用select语句从channel中读取。若在规定时间内读取成功，需满足opf.opIndex >= args.OpIndex才能回复成功；若读取超时，则回复Wrong Leader。
删除这个channel

关于加锁

Client中的锁只作用与序列号，而没有作用于整个过程。这代表一个指令执行过程中，还可以交叉执行其他指令。指令最终在K/V数据库中作用的顺序以指令开始被请求的顺序，即加锁确定序列号的前后顺序为准。

Server同理。

简介

相关资料

实现踩坑过程

完整设计思路

Client

Server

关于加锁