MIT 6.824 lab2 启动流程－－多阻塞输入驱动的状态机模型设计－－raft算法实现

标签（空格分隔）： 分布式系统

运行

go test -run Election

首先go编译器会执行

TestInitialElection(t *testing.T) 
TestReElection(t *testing.T)

两个函数

1.make_config

TestInitialElection函数会执行
cfg := make_config(t, servers, false)
初始化整个网络

make_config函数会填充config结构体,其中

cfg.net = labrpc.MakeNetwork()

会调用labrpc.MakeNetwork()函数创建一个网络,这个函数原型如下:

func MakeNetwork() *Network {
    //fmt.Printf("MakeNetwork开始了n")
    rn := &Network{}
    rn.reliable = true
    rn.ends = map[interface{}]*ClientEnd{}
    rn.enabled = map[interface{}]bool{}
    rn.servers = map[interface{}]*Server{}
    rn.connections = map[interface{}](interface{}){}
    rn.endCh = make(chan reqMsg)

    // single goroutine to handle all ClientEnd.Call()s
    go func() {
        for xreq := range rn.endCh {
            go rn.ProcessReq(xreq)
        }
    }()

    return rn
}

从上面代码可以看出,这个函数只是初始化了Network结构体。关键在于这个结构体里有一个channel，rn.endCh.注意到目前为止的代码运行在主线程里,然后该函数创建了一个线程 thread1，监听该管道

2. cfg.start1

主线程main继续执行

for i := 0; i < cfg.n; i++ {
        cfg.logs[i] = map[int]int{}
        cfg.start1(i)
    }

cfg.n指服务器的个数
这段代码通过for循环初始化每一个服务器，也就是网络中的每一个节点,这个函数会进一步完善cfg结构体的内容
cfg.start1函数原型如下:

func (cfg *config) start1(i int) {
    cfg.crash1(i)

    // a fresh set of outgoing ClientEnd names.
    // so that old crashed instance's ClientEnds can't send.
    cfg.endnames[i] = make([]string, cfg.n)
    for j := 0; j < cfg.n; j++ {
        cfg.endnames[i][j] = randstring(20)
    }

    // a fresh set of ClientEnds.
    ends := make([]*labrpc.ClientEnd, cfg.n)
    for j := 0; j < cfg.n; j++ {
        ends[j] = cfg.net.MakeEnd(cfg.endnames[i][j])
        cfg.net.Connect(cfg.endnames[i][j], j)
    }

    cfg.mu.Lock()

    // a fresh persister, so old instance doesn't overwrite
    // new instance's persisted state.
    // but copy old persister's content so that we always
    // pass Make() the last persisted state.
    if cfg.saved[i] != nil {
        cfg.saved[i] = cfg.saved[i].Copy()
    } else {
        cfg.saved[i] = MakePersister()
    }

    cfg.mu.Unlock()

    // listen to messages from Raft indicating newly committed messages.
    applyCh := make(chan ApplyMsg)
    go func() {
        for m := range applyCh {
            err_msg := ""
            if m.UseSnapshot {
                // ignore the snapshot
            } else if v, ok := (m.Command).(int); ok {
                cfg.mu.Lock()
                for j := 0; j < len(cfg.logs); j++ {
                    if old, oldok := cfg.logs[j][m.Index]; oldok && old != v {
                        // some server has already committed a different value for this entry!
                        err_msg = fmt.Sprintf("commit index=%v server=%v %v != server=%v %v",
                            m.Index, i, m.Command, j, old)
                    }
                }
                _, prevok := cfg.logs[i][m.Index-1]
                cfg.logs[i][m.Index] = v
                cfg.mu.Unlock()

                if m.Index > 1 && prevok == false {
                    err_msg = fmt.Sprintf("server %v apply out of order %v", i, m.Index)
                }
            } else {
                err_msg = fmt.Sprintf("committed command %v is not an int", m.Command)
            }

            if err_msg != "" {
                log.Fatalf("apply error: %vn", err_msg)
                cfg.applyErr[i] = err_msg
                // keep reading after error so that Raft doesn't block
                // holding locks...
            }
        }
    }()

    rf := Make(ends, i, cfg.saved[i], applyCh)

    cfg.mu.Lock()
    cfg.rafts[i] = rf
    cfg.mu.Unlock()

    svc := labrpc.MakeService(rf)
    srv := labrpc.MakeServer()
    srv.AddService(svc)
    cfg.net.AddServer(i, srv)
}

这段函数有几个关键代码:
1.

ends := make([]*labrpc.ClientEnd, cfg.n)
...中间省略若干代码
rf := Make(ends, i, cfg.saved[i], applyCh)
cfg.mu.Lock()
cfg.rafts[i] = rf
cfg.mu.Unlock()

这一段是调用ClientEnd函数创建服务器节点,然后调用raft中的Make函数(lab2中需要完成的函数，经过上面的分析可以看出这个函数运行在main主线程里)初始化该节点的rf结构体，然后把该节点的rf结构体加入到网络配置信息中。cfg.rafts管理每一个节点的raft结构体信息

ClientEnd函数原型如下:

func (rn *Network) MakeEnd(endname interface{}) *ClientEnd {
    rn.mu.Lock()
    defer rn.mu.Unlock()

    if _, ok := rn.ends[endname]; ok {
        log.Fatalf("MakeEnd: %v already existsn", endname)
    }

    e := &ClientEnd{}
    e.endname = endname
    e.ch = rn.endCh
    rn.ends[endname] = e
    rn.enabled[endname] = false
    rn.connections[endname] = nil

    return e
}

这段代码中有个关键语句,

e.ch=rn.endch

这个语句说明，每个节点的管道实际上都指向了Network所创建的管道。而由之前的分析，这个管道实际上被线程thread1阻塞监听处理
2.

applyCh := make(chan ApplyMsg)
    go func()

这一段代码所创建的管道的用处还不清楚,假设该处创建的线程为thread2
3.

svc := labrpc.MakeService(rf)
    srv := labrpc.MakeServer()
    srv.AddService(svc)
    cfg.net.AddServer(i, srv)

这段代码尚未清楚

3. rpc调用的实现

rpc调用的实现在labrpc里
函数为:

func (e *ClientEnd) Call(svcMeth string, args interface{}, reply interface{}) bool {
    req := reqMsg{}
    req.endname = e.endname
    req.svcMeth = svcMeth
    req.argsType = reflect.TypeOf(args)
    req.replyCh = make(chan replyMsg)

    qb := new(bytes.Buffer)
    qe := gob.NewEncoder(qb)
    qe.Encode(args)
    req.args = qb.Bytes()

    e.ch <- req

    rep := <-req.replyCh
    if rep.ok {
        rb := bytes.NewBuffer(rep.reply)
        rd := gob.NewDecoder(rb)
        if err := rd.Decode(reply); err != nil {
            log.Fatalf("ClientEnd.Call(): decode reply: %vn", err)
        }
        return true
    } else {
        return false
    }
}

这个函数是ClientEnd结构里的方法
svcMeth string是rpc调用的方法名
args和reply分别是输入和结果参数
该函数首先封装req结构体
然后用序列化框架序列化req.
然后通过channel将req传入e.ch管道。这里是最关键的。正如同之前讨论的每个节点的管道实际上都指向了Network所创建的管道,所以e.ch实际上是Network结构体里的rn.endch管道，这个管道由线程thead1监听处理.这是在labrpc.MakeNetwork()函数中已经处理好的.
thead1线程代码如下:

    go func() {
        for xreq := range rn.endCh {
            go rn.ProcessReq(xreq)
        }
    }()

所以rpc的核心就是rn.ProcessReq函数
这个函数就不分析了
依我的理解，labrpc中并没有为每个节点创建线程，实现真正意义上线程之间的通信。而是将网络拓扑组织成数据结构的形式存储在主线程中(即config结构体),然后通过线程thread1以及channel来实现模拟意义上的rpc。
除此之外，框架代码还实现如下功能：
对于raft.go中所创建的rpc调用函数Raft.AppendEntries，框架代码实现了实验实现者并不需要将func注册到rpc框架里,直接调用

ok:=rf.peers[server].Call("Raft.AppendEntries",args,reply)

就可以远程调用该函数。
但问题在于，这个函数是实验完成者在框架代码里添加的函数，并不是实验框架设计者预先定义的函数。那么该函数怎么被实验框架中的labrpc识别呢？
这是因为go语言支持反射机制,go的反射机制库为reflect。反射机制允许程序根据传入的字符串”Raft.AppendEntries”,动态调用Raft.AppendEntries方法。

多阻塞输入驱动的状态机模型设计

假设
1. 阻塞输入为:listen1,listen2,listen3..
2. 状态state为:s1,s2,s3,s4…
3. do(state)指只要状态机处于state状态就必须要周期执行的动作
4. do(listen,state)指状态机处于state状态,在接收到listen的输入后所要执行的动作
5. switch语句指选择语句
6. select语句指select系统调用语句(在c语言中),在go语言中指select关键字
7. thread指线程
8. change(state)语句指将state状态变换到change(state)状态
9. cancel(do(state))语句指取消state的动作do(state)所在的线程
10. go (do(change(state)))语句指用线程方式运行动作do(change(state))
则设计模式有以下几种模式:
1. 集中外包式

  for{
    switch{
        case state==s1:


            select{
                case listen1:
                    get(input)
                    do(listen1,state)
                    change(state)
                case listen2:
                    get(input)
                    do(listen2,state)
                    change(state)
                case listen3:
                    get(input)
                    do(listen3,state)
                    change(state)
                default:
                    do(state)
            }
        case state==s2:

            select{
                case listen1:
                    get(input)
                    do(listen1,state)
                    change(state)
                case listen2:
                    get(input)
                    do(listen2,state)
                    change(state)
                case listen3:
                    get(input)
                    do(listen3,state)
                    change(state)
                default:
                    do(state)
            }
        case state==s3:

            select{
                case listen1:
                    get(input)
                    do(listen1,state)
                    change(state)
                case listen2:
                    get(input)
                    do(listen2,state)
                    change(state)
                case listen3:
                    get(input)
                    do(listen3,state)
                    change(state)
                default:
                    do(state)
            }
    }
  }

1. 集中内包式

for{
    select{
        case listen1:
            get(input)
            switch{
                case state==s1:

                    do(listen1,state)
                    change(state)
                    cancel(do(state))
                    go do(change(state))
                case state==s2:
                    do(listen1,state)
                    change(state)
                    cancel(do(state))
                    go do(change(state))
                case state==s3:
                    do(listen1,state)
                    change(state)
                    cancel(do(state))
                    go do(change(state))
            }
        case listen2:
            get(input)
            switch{
                case state==s1:
                    do(listen2,state)
                    change(state)
                    cancel(do(state))
                    go do(change(state))
                case state==s2:
                    do(listen2,state)
                    change(state)
                    cancel(do(state))
                    go do(change(state))
                case state==s3:
                    do(listen2,state)
                    change(state)
                    cancel(do(state))
                    go do(change(state))
            }
        case listen3:
            get(input)
            switch{
                case state==s1:
                    do(listen3,state)
                    change(state)
                    cancel(do(state))
                    go do(change(state))
                case state==s2:
                    do(listen3,state)
                    change(state)
                    cancel(do(state))
                    go do(change(state))
                case state==s3:
                    do(listen3,state)
                    change(state)
                    cancel(do(state))
                    go do(change(state))
            }
    }
}

3.分布内包式

thread1:
func listen1(){
    get(input)
    switch{
            case state==s1:
                do(listen1,state)
                change(state)
                cancel(do(state))
                go do(change(state))
            case state==s2:
                do(listen1,state)
                change(state)
                cancel(do(state))
                go do(change(state))
            case state==s3:
                do(listen1,state)
                change(state)
                cancel(do(state))
                go do(change(state))
        }
}
thread2:
func listen2(){
    switch{
        case state==s1:
            do(listen2,state)
            change(state)
            cancel(do(state))
            go do(change(state))
        case state==s2:
            do(listen2,state)
            change(state)
            cancel(do(state))
            go do(change(state))
        case state==s3:
            do(listen2,state)
            change(state)
            cancel(do(state))
            go do(change(state))
    }
}
thread3:
func listen3(){
    switch{
        case state==s1:
            do(listen3,state)
            change(state)
            cancel(do(state))
            go do(change(state))
        case state==s2:
            do(listen3,state)
            change(state)
            cancel(do(state))
            go do(change(state))
        case state==s3:
            do(listen3,state)
            change(state)
            cancel(do(state))
            go do(change(state))
    }
}

这三种设计各有优劣
一般来讲的话，状态机在state状态下收到输入listen后变换成状态state2中,需要执行收到输入listen后的动作do(listen,state2).除此之外,在状态state2中有可能周期性执行某种动作do(state2).这种情况是很常见的.
举例说:

集中外包式的方法有一种解决do(state2)的天然优势，比较上面的代码就可以清楚的看到.但是也可以看的出来
而以集中外包式和集中内包式的方法运行，会发现只要有一个线程就可以解决多阻塞监听的问题。这是因为运用了select套接字
而分布内包式的设计，有多少个阻塞监听就有多少个线程。从这个角度来看，确实比上面两种设计方式复杂
除此之外
在解决某一状态周期性动作时,集中外包式仍然只需要在一个线程里就可以解决.而集中内包式以及分布内包式需要反复创建线程以及取消线程来实现

但是集中外包式也有自己的问题.
有些阻塞监听是天然的多线程.
即每个阻塞监听只能以单个线程的方式接收，不能使用select套接字.此时分布内包式的设计更自然一些。
但是这种情况并非不能通过集中外包式设计.用main()线程实现集中外包式状态机。listen1,listen2,listen3三个阻塞监听线程通过三个管道chan1,chan2,chan3与main线程通信。而main线程监听这三个管道chan1,chan2,chan3.如果listen1监听到数据，则向管道chan1中写数据。这样通过增加一层管道通信就可以实现这种要求.
但是这种设计也有相关问题。比如，listen1监听到数据value1,而状态机的状态改变需要用到value1,那么value1如何从listen1线程传入main线程需要额外的设计
下面给出设计代码

thread1 listen1(chan1 chan int){
    get(input)
    chan1<-1
}
thread2 listen2(chan2 chan int){
    get(input)
    chan2<-1
}
thread3 listen3(chan3 chan int){
    get(input)
    chan2<-1
}
thread_main main(){
    var chan1 chan int//管道１
    var chan2 chan int//管道２
    var chan3 chan int//管道３
    go listen1(chan1)//创建listen1线程
    go listen2(chan2)//创建listen2线程
    go listen3(chan3)//创建listen3线程
    //实现一个集中外包式的状态机
    for{
    switch{
        case state==s1:


            select{
                case listen1:
                    get(input)
                    do(listen1,state)
                    change(state)
                case listen2:
                    get(input)
                    do(listen2,state)
                    change(state)
                case listen3:
                    get(input)
                    do(listen3,state)
                    change(state)
                default:
                    do(state)
            }
        case state==s2:

            select{
                case listen1:
                    get(input)
                    do(listen1,state)
                    change(state)
                case listen2:
                    get(input)
                    do(listen2,state)
                    change(state)
                case listen3:
                    get(input)
                    do(listen3,state)
                    change(state)
                default:
                    do(state)
            }
        case state==s3:

            select{
                case listen1:
                    get(input)
                    do(listen1,state)
                    change(state)
                case listen2:
                    get(input)
                    do(listen2,state)
                    change(state)
                case listen3:
                    get(input)
                    do(listen3,state)
                    change(state)
                default:
                    do(state)
            }
    }
}

raft算法描述

在任何时刻，每一个服务器节点都处于这三个状态之一:领导人，跟随者或者候选人

下面是关于raft算法中几个最重要的地方，分别加以说明

1. 原文:
   任期在Raft算法中充当逻辑时钟的作用，这会允许服务器节点查明一些过期的信息比如陈旧的领导者。当服务器之间通信的时候会交换当前任期号；如果一个服务器的当前任期号比其他人小，那么他会更新自己的编号到较大的编号值。如果一个候选人或者领导者发现自己的任期号过期了，那么他会立即恢复成跟随者状态。如果一个节点接收到一个包含过期的任期号的请求，那么他会直接拒绝这个请求。
   说明:
   任期号的比较，在状态机进行状态转换时非常重要。尤其是在解决陈旧的领导重新连上后的问题上。

2. 原文:
   Raft算法中服务器节点之间通信使用远程过程调用(RPCs),并且基本的一致性算法只需要两种类型的RPCs。请求投票(RequestVote)　RPCs由候选人在选举期间发起，然后附加条目(AppendEntries)RPCs由领导人发起，用来复制日志和提供一种心跳机制
   说明:
   实验中只需要实现两种RPC调用.所以实际上每个服务器只有四种驱动状态机的输入情况(还有一个选举计时器，但是这是节点内部的输入驱动)

   • 接收到其它服务器发出的RequestVote的输入
   • 自己发出RequestVote后接收到的返回结果
   • 接收到其它服务器发出的AppendEntries的输入
   • 自己发出AppendEntries后接收到的返回结果

3. 原文:
   跟随者:如果在超过选举超时时间的情况之前都没有收到领导人的心跳，或者是候选人请求投票的，就自己变成候选人.
   说明:
   有两种输入情况都会造成选举计时器的重置

   • 收到心跳包
   • 收到候选人请求投票
     所以，如果一个follower收到了请求投票后，会重置选举计时器。

4. 原文:
   每一个服务器最多会对一个任期号投出一张选票
   说明:
   一个任期内的所有的选票的数量是一定的

5. 
   原文:
   一旦候选人赢得选举，他就立即成为领导人。然后他会向其他的服务器发送心跳消息来建立自己的权威并且阻止新的领导人的产生
   说明:
   阐释了leader状态应该做的动作
6. 
   原文:
   在转变成候选人后就立即开始选举过程
   
   • 自增当前的任期号
   • 给自己投票
   • 重置选举超时计时器
   • 发送请求投票的RPC给其他所有服务器
     如果接收到大多数服务器的选票，那么就变成领导人
     如果接收到来自新的领导人的附加日志RPC,转变成跟随者
     如果选举过程超时，再次发起一轮选举
     说明:
     在候选人状态受到以上三种输入的驱动，需要处理这三种情况
7. 
   原文:
   跟随者:如果在超过选举超时时间的情况之前都没有收到领导人的心跳，或者是候选人请求投票的，就自己变成候选人.
   说明:
   给出了跟随者变成候选人的条件

下面给出状态机的伪代码描述.状态机作为一个线程在Make函数里创建

repeat:
    如果当前状态是follower状态:
        如果选举计时器超时:
            变成candidate
            自增当前的任期号
            给自己投票
            重置选举超时计时器
            发送请求投票的RPC给其他所有服务器
    如果当前状态是candidate状态:
        select{//select关键字
            case 接收到大多数服务器的选票：
                变成leader
                重置votedFor
                重置选举计时器
                发送心跳包
            case 接收到来自新的领导人的附加日志RPC:
                变成follower
                重置votedFor
                重置选举计时器
            case 选举过程超时:
                //再一次选举
                变成candidate
                自增当前的任期号
                给自己投票
                重置选举超时计时器
                发送请求投票的RPC给其他所有服务器
        }
    如果当前状态是leader状态:
            发送心跳包
            睡眠50ms

下面给出四种输入驱动的伪代码描述:

1. 接收到其它服务器发出的RequestVote的输入

       func RequestVote:
        如果(发送方的term<自己的term):
            则不投票给它
            return 
        如果(发送方的term>自己的term):
            重置votedFor
            重置选举计时器
        如果(发送方的term>=自己的term 并且(自己的votedFor为空　或者　votedFor==发送方的身份编号)):
            就投票给他
        否则:
            不投票给他

1. 自己发出RequestVote后接收到的返回结果

       func send_RequestVote:
        向其他服务器发送requestvote
        收到其它服务器的返回值value
        如果其它服务器给自己投票了:
            则自己的票数+1
        如果自己的票数大于一半:
            则向状态机通过管道发送消息说"接收到大多数服务器的选票"

1. 接收到其它服务器发出的AppendEntries的输入

       func AppendEntries:
        如果(发送方的term<自己的term):
            说明这是旧领导发送给自己的心跳包
            什么也不处理，返回自己的term给旧领导，让旧领导知道自己的term过期了
            return
        否则:
            如果自己现在的身份是candidate:
                说明新领导产生了，向状态机通过管道发送消息"接收到来自新的领导人的附加日志RPC"
            如果自己是follower:
                正常处理领导发来的日志
                重置选举计时器

1. 自己发出AppendEntries后接收到的返回结果

       func send_AppendEntries:
        发送日志
        收到其它服务器的返回值
        如果对方返回的任期>自己的任期:
            说明自己是旧leader
            更新自己的任期
            将自己变成follower
        否则:
            正常处理follower的返回值

多线程编程所产生的问题

1.多阻塞输入驱动的状态变化模型
2.多线程通信
一个管道，多线程写，一线程读
产生的
1. 写阻塞问题
即另一个线程向管道里写数据，读线程没有取出数据，所造成的另一个写线程无法往管道里写
2. 写超时问题
如果写超时了，则写超时的线程如何停止写.如果没处理这种情况，1.线程会无限阻塞，2.有的写线程有需求，即如果写超时了，则该数据data要取消向管道里写。如果不处理写超时，在超时后，如果管道里的数据被取出了，则data等到了向管道里写的机会。但是data是超时的数据，不应该往里写，但是被读线程读出来了.
一个管道,一线程写，多线程读
产生的读竞争问题
一个管道，多线程写，多线程读

最后

以上就是阔达歌曲为你收集整理的MIT 6.824 lab2 启动流程以及raft算法实现MIT 6.824 lab2 启动流程－－多阻塞输入驱动的状态机模型设计－－raft算法实现的全部内容，希望文章能够帮你解决MIT 6.824 lab2 启动流程以及raft算法实现MIT 6.824 lab2 启动流程－－多阻塞输入驱动的状态机模型设计－－raft算法实现所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。