前两篇分别介绍了事务在MGR中的执行流程以及如何进行事务认证，本篇我们详细介绍事务如何进入Paxos以及在Paxos中具体执行过程，最后描述完成Paxos协议后如何返回给MGR。

Paxos实现简介

几个关键协程

在MGR中，Paxos协议通过一系列协程实现，其中与MySQL事务相关的协程主要由下面几个：

proposer_task

该协程用于实现paxos协议的propose服务，其从propose队列prop_input_queue获取等待propose的消息，首先进行batch，然后选择一个paxos实例（pax_machine）开始propose，消息走paxos协议期间，该协程睡眠等待协议完成。每个节点的paxos服务默认有10个propose协程；

acceptor_learner_task

该协程处理本节点或其他节点发送给本节点paxos服务的消息，该消息可以是客户端消息，也可以是Paxos协议中间过程消息。客户端消息即为通过send_message()发送的MySQL事务，也就是说事务是先经过acceptor_learner_task然后才到proposer_task进行propose。每个节点默认有至少5+N个acceptor_learner_task，具体数量与集群中节点数相关，N即为节点数；

sender_task

该协程用于将已经过本节点paxos服务处理的消息发送给其他节点，如A节点通过sender_task给其他节点发送proposal消息，或者B节点通过sender_task给A节点发送proposal回复消息；

local_sender_task

该协程与sender_task相似，但处理的是发送给本节点的消息。从上面的描述可以发现，sender_task和local_sender_task发送的消息由目标节点的acceptor_learner_task处理。

executor_task

消息借助上述协程走完paxos协议后，交由该协程执行。最终会调用MGR上层注册的回调函数进行后续的事务认证和回放等处理。

几个关键对象

如果把paxos协议看成一个关系图，那么上述协程可以认为是边（edge）或者通道，下面介绍作为顶点（vertex）的几个对象。

site_def

该对象描述了一个有时效性的MGR/Paxos集群配置，包含了当前集群中节点集，每个节点的server对象，该配置产生和生效的时间，每个节点的通信状态，本节点的全局视图和本地视图信息等。与正常的paxos协议执行相关的字段包括servers和delivered_msg。前者维护了本节点到集群中其他节点的连接，后者表现各个节点的消息完成状态。

每个节点维护了一个site_def对象，nodeno就是本节点在Paxos集群中的序号，该序号是Paxos消息synode_no的组成部分，synode_no的另一部分为消息序号，结合在一起synode_no表现为{X, N}，其中X为消息序号，N为节点序号。在此，不展开讲解MGR所用的Paxos变种Mencius，可通过链接做进一步了解。

在进行prepare、propose、accept和learn等Paxos操作时，需要依赖site_def中的server对象发送消息。

server

上图就是server的定义，我们不对每个字段做一一解读，仅关心其中跟消息发送相关的字段。srv是server指向节点的名称，con维护了本节点到server指向节点的长连接，outgoing队列用于缓存本节点需发送到server指向节点的消息，sender是在server初始化时注册的sender_task协程，用来负责从outgoing队列中读取消息发送到server指向节点。在Paxos正常运行期间，server对象是集群中节点间消息发送的媒介，通过con建立2个节点间的联系，发送端即为sender字段对应的sender_task协程，接收端就是server对应节点上的acceptor_learner_task协程。需要注意的是，本节点也会为自己创建一个server对象，此时sener字段即为local_sender_task。

pax_msg

如果说server维护了Paxos集群各节点间的通信管道，那么pax_msg就是管道中传输的消息。其定义如下：

相比前2个对象，本对象各字段更抽象，在此仅挑选几个字段进行介绍。显然from和to字段标识了消息的源节点和目标节点，to字段可能为null，表示发往所有节点；op标识消息的操作类型，比如是prepare还是propose消息；synode表示该消息的序号，类型为synode_no，消息组成详见site_def小节；msg_type表示消息类型，是普通消息还是不携带数据的noop消息等；receivers表示有多少节点已经接收到了该消息；a是消息数据，比如对于事务消息，该字段就包含了一系列的log_event。reply_to和proposal是2个ballot，ballot由一个节点序号nodeno和计数器cnt组成，表示某次提案编号，显然proposal字段表示本次消息提案的编号，reply_to表示本次消息是对reply_to对应的提案编号的回复。同个pax_msg的proposal和reply_to字段对应的ballot可能不同，举个例子，A节点向B节点发送prepare消息，B节点收到后，发现本节点已经接受了对应synode的消息，那么B节点在回复A消息时，会将消息中的propsal字段设置为本节点的提案编号并替换掉消息体。

ballot是Paxos算法强相关的字段，对于某个确定的synode消息，在prepare阶段，一个节点只能继续处理比其之前收到的所有提案编号更大的提案，在acceptor阶段，一个节点只能处理不小于其之前收到的所有提案编号的提案。

pax_machine

在节点上，pax_msg保存在pax_machine中，在Paxos算法中，pax_machine就是一个Paxos实例，保存了一条Paxos日志，一系列有序的Paxos日志组成了Paxos状态机。每个节点维护了一个5万个pax_machine对象的缓存（paxos cache），如果节点上所有pax_machine的数据大小超过了1GB，那么会开始清理无用的pax_machine。淘汰算法采用lru机制，不能清理还未走完Paxos协议的pax_machine。pax_machine定义如下：

synode标识了该pax_machine处理的消息序号。在MGR中，一个消息/提案是以异步的方式走完Paxos协议，发起投票的proposer_task会在rv字段上等待并周期性被唤醒，直到该提案完成。op表示消息操作类型。proposer、acceptor和learner分别保存了消息序号为synode的提案从本节点视角看到的不同阶段的状态。

proposer字段是提案的发起者维护的字段，accetor和learner不会操作proposer字段。MGR中的Paxos实现支持2阶段和3阶段协议，相比2阶段，3阶段协议多了一个prepare的过程。MGR对于正常的事务消息采用2阶段的方式。所以proposer字段包括了prepare和propose 2个阶段，其中bal表示目前的提案编号，prep_nodeset和prop_nodeset分别表示已经回复prepare和propose消息的节点集；sent_prop和sent_learn分别在收到大多数节点回复prepare和propose消息时设置，设置为bal；msg字段在proposer_task进行pax_machine初始化时被置为等待投票的客户端消息，如果有其他节点回复的prepare消息中携带的proposal大于proposer节点的ballot，那么msg会被替换为回复消息对应的pax_msg对象，即投票的value发生改变。

acceptor字段由作为提案接受者的节点维护，保存了本节点至今已收到的消息编号为synode的最大ballot，msg字段为对应ballot携带的pax_msg。learner字段表示编号为synode的消息最终被学习的消息体/value。显然，该字段不为空表示对应synode已经完成了Paxos协议。

Paxos协议准备阶段

在简单介绍MGR中Paxos实现所依赖的主要协程和对象后，下面按照协议流程进行详细讲解。

事务从客户端进入Paxos

Xcom在初始化时注册tcp_server协程来监听socket服务器端连接，每当有新的连接进来，就会创建一个acceptor_learner_task协程来处理该连接的后续消息，在第一篇已经提及在进行MGR初始化时，本节点会向本地Xcom申请创建6个连接/socket客户端，也就是会有6个acceptor_learner_task来分别响应本节点的客户端请求。

事务通过send_message()封装进入到Xcom模块，send_message()会基于round robin的方式选择一个socket客户端并将事务数据写入到该socket中，事务演变成一个Paxos消息（PAX_MSG），类型为client_msg。socket客户端对应的acceptor_learner_task会读取paxos消息并调用dispatch_op处理。

dispatch_op根据消息的类型分别对各种消息进行处理，如果是client_msg，会进一步读取消息中的c_t字段（cargo_type类型），对于事务消息会调用handle_client_msg，交由channel_put插入到prop_input_queue队列中。

Proposer处理逻辑

获取事务并batch

事务被插入prop_input_queue队列后，进入Proposer处理流程，对应的协程为proposer_task。该协程负责让事务开始走Paxos协议，等待Paxos流程执行完成，并进行最后处理。具体操作如下。

调用channel_get从prop_input_queue获取请求进行处理，开始走paxos协议，首先会进行batch操作，通过一次性处理多条请求来提高效率，前提是请求不能是config或者view类型。完成batch后，本节点会判断当前paxos的繁忙情况，如果正在处理的请求数量超过设置的event_horizon(默认为10)参数，则选择等待并不断跟event_horizon参数比较，否则马上走paxos协议。

异步执行paxos

开始paxos协议时，先会增加synode(X,N)中的X即msgno，获取当前的节点集并将其设置为proposer节点集-proposer_site，将增加后的synode作为节点当前请求current_message。从节点pax_machine的cache中找到一个可用的pax_machine对象并初始化，该pax_machine对象就跟该请求后续的paxos协议流程相绑定，需要走协议的请求在节点间进行传输是会被封装为pax_msg，即paxos消息（为了便以区分，我们将需要达成协议的操作叫做请求，将请求在节点间进行传输的媒介叫做paxos消息），pax_machine对象包括proposer，acceptor和learn等多个字段，分别用于处理paxos协议的不同阶段；

在上面的基础上，进一步进行了pax_machine和pax_msg的初始化，之后，开始paxos协议的proposer，此处有个参数threephase，表示是否需要在propose之前先走prepare流程，在Xcom中，该参数为false，即直接进行propose，无需prepare，当然，例外的情况是如果这个请求设置了force_delivery。如果需要prepare，调用push_msg_3p，否则调用push_msg_2p走paxos协议。走协议的过程为异步的过程。在后续描述中，我们假设走push_msg_3p。

等待事务完成Paxos处理

在此期间，proposer_task会在pax_machine的rv队列睡眠，周期性醒来，或者每当有新请求被learn后也会醒来，判断该pax_machine是否已经走完learn阶段。若等待一段时间后，还未到learn阶段，则会重新调用push_msg_3p，此时synode还是原来的，一个消息使用相同的synode prepare多次并不会有问题，如果之前prepare已经成功，那么会返回之前prepare的值，不影响协议的正确性。

若对应synode对应的请求已经完成learn阶段，proposer_task会调用match_my_msg判断完成learn的消息的内容是否与从客户端发过来的请求内容相同，如果相同，则表示完成了本请求的paxos协议流程，统计本次请求发了多少时间并删除消息队列中该客户端请求（pax_machine中仍然会保留一段时间，由sweeper_task负责回收），后面会介绍，达成协议的客户端请求会通过learn流程返回给客户端。

Paxos协议实现

下面分析如何走paxos协议，这一过程需要非常关心pax_machine对象的状态变化。本文将Paxos协议细分为prepare、propose和learn三个阶段进行描述。网上有幅图较好得表现了大致的执行流程，可供参考：

prepare阶段

proposer_task调用的push_msg_3p，在push_msg_3p()函数中会prepare_push_3p来初始化pax_machine的proposer子对象和pax_msg内容，引入ballot来进行投票操作。之后重置proposer子对象的prep_nodeset字段，该字段用于表示prepare阶段已经回复的节点位图。prepare_push_3p()会调用prepare_msg()来进行设置pax_msg的操作类型为prepare_op并最终调用send_to_acceptors()来将prepare消息发送给各个acceptor节点。

前面已经介绍过每个节点对应一个server对象，其中有个outgoing队列，缓存本节点发往server对应节点的消息。如果server对象是节点自身，则会初始化一个local_sender_task()协程，它处理outgoing消息的方式是直接调用dispatch_op()函数。如果server对象是集群中其他paxos节点，outgoing消息通过sender_task()发送，对应paxos节点的服务器端通过acceptor_learner_task()接收，最终也是调用dispatch_op()函数处理消息携带的请求。

prepare_op处理

对于prepare_op，dispatch_op()先会从本节点的pax_machine cache中找到一个跟消息的synode对应的pax_machine对象，如果没有，则基于lru来重新初始化一个。最后会调用handle_prepare来处理prepare_op，该函数本质上调用handle_simple_prepare，分为2种情况：

如果需要处理的synode对应的pax_machine对象在本节点已经处于learn的状态，则调用teach_ignorant_node()初始化一个pax_msg消息reply，将本节点pax_machine中learner子对象对应的pax_msg赋予reply，给源节点回复一个类型为learn_op，内容为reply的消息。
如果对应synode在本节点的pax_machine对象不是learn状态，则先使用gt_ballot判断消息中的ballot投票次数是否大于pax_machine对象promise中的ballot，即需要满足paxos协议消息prepare阶段的接收承诺。或者，即使不满足promise，但该pax_msg是noop，且本节点pax_machine对象的状态是已经accept了一个noop消息，则进入下一步处理。若不满足前面2中消息，则不回应（这里不回复不会有问题，其实在proposer_task()那边会通过不断知道最新达成协议的消息，所以会进行相应操作）。

handle_simple_prepare()处理满足上述任一条件的场景，若满足promise，则更新promise对应的ballot，判断本节点pax_machine是否已经accept了一个值（哪怕是noop），则将pax_machine的acceptor中的pax_msg赋予用于回复的pax_msg对象reply，将类型设置为ack_prepare_op。否则，表示没有之前没有接收过synode对应的prepare消息。单纯返回ack_prepare_empty_op。

ack_prepare_op和ack_prepare_empty_op处理

分别调用handle_ack_prepare和handle_ack_prepare_empty。两则的区别在于，对于handle_ack_prepare，表示远端节点已经accept了一个value（判断标志为回复消息的投票器大于本节点缓存的pax_machine的proproser.msg中的投票器），那么本节点需要将自己propose的值替换为远端节点返回的值继续走propose，并更新投票器为回复消息中的投票器。所以，如果不同节点ack_prepare回复了不同的值，那么本节点是选取投票器最大的节点的值作为propose的值。

handle_ack_prepare/handle_ack_prepare_empty会调用handle_simple_ack_prepare()来将pax_machine对象中proposer.prep_nodeset位图中源消息对应字段置位，表示已经收到了其prepare回复消息，之后进一步调用check_propose()。

propose阶段

check_propose()先调用prep_majority()判断是否大多数节点都已经回复本次prepare。若是，开始为propose操作进行初始化，包括更新pax_machine对象proposer中pax_msg的proposal投票器，synode。将proposer.prop_nodeset清零。调用propose_msg()来将消息初始化为accept_op，更新reply_to投票器为消息发送投票器，调用brand_app_data()更新请求中app_key部分的内容（与prepare不同）。调用send_to_acceptors()来将propose消息发送给各个acceptor节点，check_propose()在prep_majority()分支最后将propose消息发送的sent_prop投票器设置为接收到的消息中的最大投票器，这样就不会二次就入到check_propose()处理流程中，因为在进入前会判断消息的投票器是否大于sent_prop。

accept_op处理

dispatch_op()调用handle_accept来处理accept_op，最终调用handle_simple_accept()，该函数与handle_simple_prepare()类似，但也有所不同：如果对应synode在本节点的pax_machine对象不是learn状态，则先使用gt_ballot判断消息中的ballot投票次数是否大于等于pax_machine对象promise中的ballot，即需要满足paxos协议消息propose阶段的接收承诺。或者，即使不满足promise，但该pax_msg是noop，且本节点pax_machine对象的状态是已经accept了一个noop消息，则进入下一步处理。若不满足前面2中消息，则不回应。

handle_simple_accep()t处理满足上述任一条件的accept请求，调用replace_pax_msg()将本节点pax_machine中的acceptor.msg替换为accept消息中的pax_msg。发送回复消息，此时消息类型为ack_accept_op。

ack_accept_op处理

dispatch_op()调用handle_ack_accept()来处理ack_accept_op，该函数先进行有效性判断，并确认回复消息的投票器跟pax_machine对象投票器相同（回复本节点的accept请求），将pax_machine对象proposer.prop_nodeset位图对应bit置位。若之前还未达成大多数，则调用check_learn()，与prepare类似，调用prop_majority()来判断是否满足大多数。若是，则更新proposer.msg->receivers位图信息，并根据系统设置的no_duplicate_payload参数来调用tiny_learn_msg()或learn_msg()，之后更新proposer.sent_learn防止二次进入majority处理流程。

消息处理时的流控

在进行dispatch_op()时，会判断当前节点状态是否too_far()，如果是，则不响应ack_prepare_op、ack_prepare_empty_op 和ack_accept_op。这三个op分别是其他节点对本节点的prepare和accept的回复操作。也就是如果当前消息的synode_no msgno大于本节点executed_msg+事件时间窗（too_far()），那么就不响应其他节点回复的ack消息？个人感觉这应该是考虑到了executed_msg递增逻辑。如果当前消息比较靠前，此时回复了ack，那该消息majority了后，会继续走paxos，但就算走完paxos协议，也无法递增executed_msg，所以，这么做其实是让比它synode_no小的本节点消息先完成，体现顺序性。

learn阶段

Xcom设置no_duplicate_payload为true，所以走tiny_learn_msg()流程。tiny_learn_msg初始化类型为tiny_learn_op的消息，并调用send_to_all_site()。send_to_all_site()的执行逻辑与send_to_acceptors()类似。

tiny_learn_op和read_op处理

dispatch_op()处理tiny_learn_op的方式比较复杂，首先需要判断本节点acceptor.msg是否存在，若不存在或者acceptor.msg->proposal不等于tiny_learn_op消息携带的投票器，则表示没有参与之前的accept流程或者当前accept值不是经过大多数同意的值，这两种情况需要调用send_read处理。

send_read()会调用pax_msg_new()创建一个新的pax_msg消息，调用create_read将消息初始化为read_op，调用send_to_someone()或send_to_others()消息发送给其他所有节点或者采用round robin方式发送给一个其他节点。消息接收节点的dispatch_op调用handle_read()来处理read_op，处理方式为向源节点回复learn_op消息。本节点接收到恢复的learn_op消息后，交由dispatch_op()进行处理。

learn_op处理

如果本节点参与了之前大多数接受的accept流程，那么走简洁流程，将acceptor.msg->op设置为learn_op，调用update_max_synode更新节点集信息。调用activate_sweeper激活sweeper_task协程进行pax_machine cache清理工作，具体实现暂时不介绍。最后交由handle_learn()来进行最后处理。

handle_learn()首先调用finished()判断该本节点pax_machine是否已经进行了learn_op，避免重复学习该请求。若否，执行do_learn()，交由executor_task协程来学习请求。

do_learn先将pax_msg对象中的a->chosen设置为TRUE，并将其赋予本节点pax_machine的acceptor.msg和learner.msg，最终调用add_cache_size增加pax_machine缓存大小，并调用shrink_cache来尝试回收部分缓存空间。

如果是通过send_read()发起的learn_op（即该节点未参与accept流程）而不是直接由tiny_learn_op转换而来，则是由dispatch_op处理learn_op，方式略有不同：在更新本节点pax_machine对象时，简洁流程是将对象的acceptor.msg赋予acceptor.msg和learner.msg，而learn_op时是将从其他节点收到的pax_msg赋予本节点pax_machine对象的acceptor.msg和learner.msg。

execute阶段

接下来看看executor_task()是如何将已经处于learned状态的pax_machine对象（或者说是已经达成Paxos协议的请求）告知客户端程序的。具体实现中，executor_task()协程分为2个阶段，FETCH和EXECUTE，分别对应请求获取和执行。

FETCH

FETCH用于获取和感知已经处于learn的消息，这通过调用get_xcom_message()实现，此时会引入重要的全局变量executed_msg表示已经走完Paxos全流程但还未被executor_task()处理的最小的synode请求。get_xcom_message()会调用TIMED_TASK_WAIT()来在executed_msg对应的pax_machine对象中rv字段进行等待，直到finished()返回true。然后通过set_executed_msg()来进行本节点的全局性状态更新，包括设置新的current_message，调用task_wakeup()唤醒在等待executed_msg更新的其他协程/线程（比如proposer_task在propose新请求时会判断节点的繁忙情况，需要在exec_wait等待状态变化），之后对executed_msg增一，设置最新的executor_site。

EXECUTE

在set_executed_msg()之前，会将协程的状态设置为EXECUTE，这样，协程执行下个循环时候就能够进入EXECUTE流程。该状态有个重要的变量为delivered_msg，表示还未传送给客户端的最小synode请求，正常情况下跟executed_msg相等，当有新的请求在FETCH阶段被处理后，executed_msg增一，但delivered_msg未增加，那么就需要在EXECUTE流程中将其重新变为相等，这也是将状态从FETCH变为EXECUTE的原因。

在EXECUTE中，首先判断delivered_msg请求对应的pax_machine中的请求类型是否为no-op，若是，无需交还客户端处理，直接调用incr_synode将delivered_msg增一。否则，需先调用execute_msg()处理该请求，然后在增一。最后根据情况调用garbage_collect_servers进行全局性的清理工作。

execute_msg()根据不同的请求类型进行处理，对于app_type，则调用deliver_to_app()，最终会执行MGR上层注册到Xcom的回调处理函数xcom_receive_data进行处理。后续的处理流程已经在前两篇做了详细的描述。

总结

本篇对MGR/Xcom中的Paxos实现进行了详细分析，结合本系列的前面两篇，基本上对MGR中事务的处理全过程进行了梳理。与前面2篇一样，本篇的分析过程仅限于正常的事务处理，对于成员变更、心跳等流程在Paxos中的处理并未涉及。后续有时间再专门开出一个系列进行分析。

从文中，可以发现，MGR在Paxos层维护了一个较大的paxos cache，并设置为超过1G后才进行清理，在第二篇中我们也提到事务认证模块的冲突检测数据库也会占据较大的内存。所以，相比非MGR的MySQL业务场景，MGR场景会占用更多的内存空间，不能在按照传统的MySQL内存占用规律来规划系统内存空间，否则容易导致mysqld OOM现场发生。

ITPub博客

MySQL事务在MGR中的漫游记 - Paxos协议