分布式系统入门笔记(三):从PhxPaxos中再看Paxos协议工程实现

  Phxpaxos已经开源了,且他们号称开源和内部使用的是同一套代码,那么作为小厂的程序员可有福了,可以一睹研究一下生产环境下大规模分布式系统是怎样练成的。初看代码可能会比较犯晕,果真生产环境的实现跟《Paxos made code》的复杂度不是一个数量级的,可是复杂归复杂,但毕竟代码中没有用到复杂的Moden C++特性、大量的模板元编程和晦涩难懂的编码技巧,所以只要功夫下到位相信还是肯定能搞明白其内部流程的。
  本文就是通过Phxpaxos中所附带的简单例子,摸索了解Phxpaxos中对Paxos算法的实现,算是验证一下前面对Paxos算法的学习吧。当然之前也说过,Phxpaxos同Lamport老爷爷的原版Multi-Paxos相比已经修改了很多,毕竟老爷爷的文章比较的偏理论化,所以理论上不修改的Multi-Paxos是不可能满足线上分布式系统的可用性和可靠性需求的,具体对于遇到的修改再行另表吧。

一、预备操作

  默认情况下Phxpaxos的存储模块使用的是glog,但不知道怎么回事,在我MacOS下VMware Fusion虚拟机Ubuntu-1604的环境下,严格按照《编译安装手册》还是会报无法创建log文件的错误,不知道是不是因为目录使用NFS挂载的原因,因为在实体机上面本地存储没有发现这个问题,此处也不详究了。其实Phxpaxos的实现中,很多地方都有详细的且分好等级的日志信息,查看代码发现只要设置了pLogFunc函数指针,就可以在所有log记录之前执行这个hook函数,于是乎可以在sample中设置这个option,那么整个系统的运行路径和运行状态也就一览无余了。

1
2
3
4
5
6
7
#include <stdarg.h>
void custLog(const int iLogLevel, const char * pcFormat, va_list args) {
char sBuf[1024] = {0};
vsnprintf(sBuf, sizeof(sBuf), pcFormat, args);
std::cerr << sBuf << std::endl;
}
oOptions.pLogFunc = custLog;

  Phxpaxos的代码虽然是多,但是当除掉存储模块、网络模块、CheckPoint模块、Benchmark和单元测试等部分代码后,核心代码其实也是十分有限的,而且和Paxos算法相关的部分都单独放在algorithm目录中了,查看这个文件夹中的文件名,赫然醒目的acceptor、proposer、learner、instance,就让我们估计道知道他们是什么角色履行的职责了。

二、附带phxelection选主例程解析

  Phxpaxos附带的两个sample详细构建过程都在《README》中描述清楚了,本来是想phxelection和phxecho两个例子都一起跟踪的,但是后面看着看着发现,phxelection流程走完基本就定型了,phxecho和前者的差异主要就是传递了自定义的StateMachine,所以在Paxos算法Chosen Value之后,会额外的执行客户提供的状态机转移函数Execute(),这其实和phxelection在原理上没有本质的差异,因为后者算是对于选主的特殊情况,预先定义了MasterStateMachine状态机而已。不过phxecho的日志量要小一些,可以两个例程结合起来看。

2.1 Phxpaxos中的Master Node

  在Phxpaxos的设计中弱化了Multi-Paxos中提到的Leader角色,而是使用了Master的概念,在后面PhxSQL项目的文档中也强调了:Master是唯一具有外部写权限的Node,所以可以保证这个Node的数据肯定是最新的副本,Client的所有写请求和强一致性的读请求都需要直接或者由其他Node代理转发到这个Master Node上面,而对数据一致性要求不高的普通读请求,其读请求才可以在非Master Node上面执行。
  由此可见,在稳定的情况下Master Node起到了单点协调者的作用。但是当系统新启动,或者在Master Node挂掉的时候,就需要有一个选主的操作。《Paxos理论介绍(3): Master选举》已经将选主的原理详细说明了。概而言之,就是在没有Master的时候,大家都可以认为自己是Master并发出TryBeMaster请求,这就等于大家实现一个Basic-Paxos的流程,根据Paxos的算法原理可以保证,最后只有唯一一个节点的提案被Chosen,这时候被Chosen提案的Proposer就被选为Master Node。在系统运行的过程中,各个节点会事先约定一个租赁周期,在这个周期到达之前Master Node可以提起续约的请求,其他节点当发现约定的检查周期到达之后并且Master租赁有效期间之前还能够请求到Master Node,就表示Master Node续约成功,此时就不再发起TryBeMaster的请求。这样在稳定的情况下Master Node可以一直稳定的运行下去,而当Master Node异常或者普通Node和Master Node通信失常情况下的处理方式,会在后文中作进一步的解释。

2.2 phxelection选主的实现流程

  Phxpaxos中的两个项目默认是使用的三个节点来实验的,根据Paxos中Majority的需求,其节点数至少是三个,而且尽可能的是奇数数目。每一个运行的进程对外被抽象成Node节点,在内部使用带序列号的Instance作为执行单元,且每个Instance内部都含有完整的Acceptor、Proposer、Leader的角色,而且同一个Instance中的三个角色是存在于同一个进程中的,相互之间的结果是立即可见的。

2.2.1 Node初始化操作

  Phxelection历程启动很简单,当代码创建PhxElection对象oElection后,直接调用其成员函数PhxElection::RunPaxos()启动。在这个成员函数中,可以设置相关参数oOptions(注意这个选项很大,而且后续会深入透传到整个Phxpaxos内部),这个sample中,最主要的是要打开选主开关bIsUseMaster,然后调用Node::RunNode(),也只在这个函数中才进行PNode实体的创建和初始化工作,并将得到对象的地址保存返回给poNode指针,这样用户程序也可以操作Node(而非PNode)规定的其他可用接口来访问或操作底层对象了。
  PNode对象的创建采用典型的两段式构造,当然下面很多对象都是两段式构造,主要工作在PNode::Init()中实现,这个初始化涉及的内容比较的多,几乎贯穿了整个系统的启动过程:
  (1). InitLogStorage: 好像底层是基于啥LevelDB搞的吧,先不管它了;
  (2). InitNetWork: 默认的网络配置是根据命令行参数解析到的节点网络信息IP:Port,对于UDP协议创建接收套接字和一个发送套接字,对于TCP则是通常的bind-listen服务端初始化。TCP类型的初始化工作,还包括使用epoll_create创建异步事件侦听,同时创建封装了pipe匿名管道的Notify对象用于对接收到的消息进行传递,并设置管道NONBLOCK且将管道的读端添加EPOLLIN事件侦听;设置成员变量poNetWork;
  (3). MasterList: 由于我们只用到一个Group,所以只创建一个MasterDamon对象,可见默认的Master Lease是10s,可以通过SetLeaseTime自由自定义设置这个租赁周期但是必须大于1s;同时还要创建一个MasterStateMachine对象,首先尝试从之前保存的数据加载历史信息,如果读取失败进行默认初始化,同时设置version的值为(uint64_t)-1;
  (4). Group: 和上面一样也只有一个,主要把上面已经初始化的信息保存到这个容器里面;这个里面,让我们最感兴趣的是创建了一个Instance类型的m_oInstance对象,大概看了一下Instance类的声明就让人感到十分兴奋,因为大部分的消息回调OnRecieveXXX都在这个类的声明里面,同时还包含了Acceptor、Proposer、Learner成员,表明我们离Paxos已经是很接近了;
  (5). ProposeBatch: 批量提交,算是Phxpaxos相比Multi-Paxos的一个新操作,待到我后面研究到这个深度的时候再深入;
  (6). InitStateMachine: 参数只有一个——Options,一个超大巨形体参数,但是在这个sample中我们并没有创建StateMachine,所以这里的AddStateMachine啥也没做,当然对于另外一个Phxecho需要使用用户定义的StateMachine,此处就会进行添加;然后当我们开启了选主的功能后,各个Node都会启动运行RunMaster(),这个操作会创建一个线程运行MasterDamon::run(),这个线程就是Master Node选取和维护的主要实现部分,此处跳过,后面会单独展开详述;
  (7). Group Init: 将启动时候添加的所有Node列表信息添加到SystemVSM中,同时添加GetSystemVSM()、GetMasterSM()两个StateMachine。其实看到这里,让人感觉到StateMachine实际等价于某个带状态的函数,当所谓的StateMachine发生转移的时候,实际就是其对应的函数被执行。然后执行Instance::Init(),由于Instance的初始化有较多操作,且是跟Paxos密切相关的,下面单独展开介绍。

2.2.2 Instance初始化操作

  在Instance初始化的过程中,主要是对Paxos中的几个角色的初始化,其角色大多都是继承自Thread类,所以意味着各自以线程的方式独立运行。我们主要关心的角色有:
  (1). Learner
  最终映射成LearnerSender::run()线程,做的事情就是:等待条件变量m_bIsComfirmed被设置,然后调用SendLearnedValue(m_llBeginInstanceID, m_iSendToNodeID);从llBeginInstanceID开始一直发送到当前的m_llInstanceID。对在每一个instance发送的过程中,都需要从log中查取这个instance的相关信息(比如提案节点、提案号、Chosen Value等)后打包发送,具体的信息请参看Learner::SendLearnValue()中的打包过程,然后可以选择之前在Node初始化中网络部分的TCP还是UDP方式发送,具体任务交由网络模块负责了。
  Learner会等待记录ACK信息,当前被确认的ID保存在m_llAckInstanceID变量中,确认超时后错误返回,错误时候记录当前已经发送的llSendInstanceID变量不被更新。
  (2). Acceptor
  初始化较为的简单,因为我们没有历史记录,所以加载数据发现为空,就将m_llInstanceID设置为0就可以了;
  (3). CheckpointMgr
  这个暂时也不展开说了。作为一个状态机如果要重新加载状态,最直观的就是从最开始零状态一直进行历史记录回放,但是这种方式效率低,而且随着状态机的运行记录所有的历史信息也是不现实的,所以为了增加效能实现的CheckPoint功能就是将历史记录某个时间点的状态做完整快照,加载状态可以从那个CheckPoint开始重放到当前日志记录的状态(这个过程叫做CatchUp),具体的信息可以查看《状态机Checkpoint详解》
  本sample都是从头开始的,这些东西基本都是0状态的。
  (4). IOLoop
  这也是开辟一个线程,主要是在一个消息队列m_oMessageQueue中不断的完成取出消息和处理消息(主要就是通过一个带互斥锁、条件变量封装的std::deque,腾讯说好的无锁队列呢?)的工作,正常情况下会取出消息,然后发配到OnReceive(*psMessage)处理,而OnReceive的处理流程会对数据包拆包检查,然后根据消息类型分别发配给Learner、Acceptor、Proposer对应角色去处理,消息的类型定义在了comm\commdef.h的enum PaxosMsgType中,看上去Proposer和Acceptor比较明确,而Learner处理的消息比较多啊。
  通过把这个OnReceive的处理大概逛了一下,基本就是标准的Paxos协议的内容了,比如Proposer接收到Acceptor的表决信息后,会先调用m_oMsgCounter.AddReceive(),然后检查m_oMsgCounter.IsPassedOnThisRound()看看是不是已经满足超过半数决议了,如果是就进行接下来的例程Accept()或者向Learner广播消息ProposerSendSuccess(),而如果投票失败(否决票过多或者超时)则等待几十毫秒后重新发起Prepare请求。
  上面提及的只是消费消息,消息的生产者在哪里呢?之前说道PNode在初始化网络层的时候有过TCP和UDP两种通信方式,TCP通过使用event异步事件,而UDP在一个线程中不断的接收消息,这些渠道接收到的信息,最后都会放到这个队列中去的。
  此外Instance还保留了一个m_oRetryQueue重试队列,用于处理Paxos相关消息,具体什么原理暂时不详。
  上面的这些线程的循环,都在m_bIsEnd=true的时候会退出来,在IOLoop::Stop()会设置这个变量,检索后在Instance对象被析构的时候会调用之。

2.2.3 MasterDamon主线程工作

  唉,代码中有好多错误的单词,不知道可不可以作为一个槽点……
  根据上面说到的选主原理,这个线程的工作内容也十分简单,实质的工作内容就是TryBeMaster():首先在Master Node稳定正常的情况下自身会自动续约,那么在m_llAbsExpireTime之前就会返回m_iMasterNodeID和版本号,否则返回nullnode;当Node发现返回nullnode的时候,无论是Master Node的原因还是自己和其通信的原因,都会发起一个包含llMasterVersion信息的Proposer请求,那么:
  (1). 系统刚开始的时候,所有的Node都认为自己是Master Node,然后发起Propose()请求,经过Basic-Paxos的正常流程,会最终有一个Node被Chosen,由于大家都在同一个Instance中,被Chosen Value所对应同一进程中的Learner已经知道Chosen Value了,然后就会立即发送ProposerSendSuccess()广播给所有的Learner(也可以配置该节点自身需不需要进行学习),所有Learner接收到该消息后都通过OnProposerSendSuccess()进行学习,并设置m_bIsLearned=true状态;Learner的后续执行流程检查到这个状态后,就会执行状态机MasterStateMachine状态转移Execute()即LearnMaster(),设置所有节点的m_iMasterNodeID为被Chosen Value的节点。系统启动Master Node新选取成功时候,m_llAbsExpireTime的值为0,而后面续约的时候更新为提起Propose时刻+Master租赁时长-100ms长度,自此选主成功。
  (2). 所有节点都会在Lease Timer之前发起(但不一定会实质执行)TryBeMaster(),为了减少异常情况下通过Basic-Paxos新选取Master时候可能的“活锁”冲突,大家发起Prepare的时间点会有个微小的elf差异,当发现Master Node在超时时间之前仍然有效的时候,非Master节点都直接返回而不执行BeMaster(),Master节点会发起一个续约的正常Propose请求,最终状态机转移的时候更新时间等信息,然后所有节点学习重新调整即可;
  (3). 当Master Node本身挂掉的时候,大家都会发起Propose请求,那么此时就退化成初始状态多个Node采用Basic-Paxos的方法选主的情况了;
  (4). 当某个非Master Node因为通信或者其他的原因,错误的发起了BeMaster()的时候,wiki也说了这里是个“乐观锁”的解决思路:这个节点将自己观测到的最大version打包到Propose参数里面去,后续通过正常的Paxos流程会被Chosen,待到最后执行状态转移LearnMaster()的时候,会解包对参数进行校验,发现请求的version版本和当前最新版本不符,那么放弃这次状态机的实质切换操作而直接返回了。我们乐观的预估,此时改节点和系统的通信正常了,在真正的MasterNode下次续约的时候,改节点将会正确学习到Master Node的信息并正常工作。

三、小结

  排除代码中很多写错的英文单词,虽然涉及到的模块、对象众多,但是整个项目的设计和实现还是挺清晰的,配合详细的wiki和文档以及详细的日志信息,初学者花时间跟踪项目流程还不是很困难,同时也算是了解到了腾讯内部C++的开发风格,而在代码的关键点添加BreakPoint()回调接口还是挺有新意的。
  还有,项目作者号称需要C++11标准的支持,但是发现除了for表达式、nullptr关键字外,C++的新特性基本可以说都没用到,不过也好,这样的代码更容易跟踪理解。项目中使用了大量的定时器和延时操作,就像当初通信核心网代码一样,这也是复杂系统所不能避免的。此外成员变量十分的多,但是基本都没有提供注释,这点读起来比较的费力。
  本篇文章算是走了个流水,但是核心重要的CheckPoint、成员动态变更和他们创造的BatchPropose都被略过了,有时间再研究吧。

参考