4. PhxPaxos源码分析之Learner
目录
1. PhxPaxos源码分析之关于PhxPaxos
2. PhxPaxos分析之网络基础部件
3. PhxPaxos源码分析之Proposer、Acceptor
4. PhxPaxos源码分析之Learner
5. PhxPaxos源码分析之状态机
6. PhxPaxos源码分析之归档机制
7. PhxPaxos源码分析之整体架构
4.1 基本概念
在Paxos中Learner角色负责向其他节点学习选中的提案值。具体包括如下两种场景：
- Learner所在节点参与了提案选举，Learner需要知道其接受(accept)的提案值是否被选中(chosen)。
- Learner所在节点已落后于其他节点，Learner需要选择合适的策略快速完成追赶，并重新参与到提案选举当中。
4.2 选中(chosen)通知
当本节点Proposer的某个提案被选中(chosen)时，通过(MsgType_PaxosLearner_ProposerSendSuccess)消息通知到各个节点，代码如下：
void Learner::OnProposerSendSuccess(const PaxosMsg &oPaxosMsg)
{
BP->GetLearnerBP()->OnProposerSendSuccess();
PLGHead("START Msg.InstanceID %lu Now.InstanceID %lu Msg.ProposalID %lu State.AcceptedID %lu "
"State.AcceptedNodeID %lu, Msg.from_nodeid %lu",
oPaxosMsg.instanceid(), GetInstanceID(), oPaxosMsg.proposalid(),
m_poAcceptor->GetAcceptorState()->GetAcceptedBallot().m_llProposalID,
m_poAcceptor->GetAcceptorState()->GetAcceptedBallot().m_llNodeID,
oPaxosMsg.nodeid());
if (oPaxosMsg.instanceid() != GetInstanceID())
{
//Instance id not same, that means not in the same instance, ignord.
PLGDebug("InstanceID not same, skip msg");
return;
}
if (m_poAcceptor->GetAcceptorState()->GetAcceptedBallot().isnull())
{
//Not accept any yet.
BP->GetLearnerBP()->OnProposerSendSuccessNotAcceptYet();
PLGDebug("I haven't accpeted any proposal");
return;
}
BallotNumber oBallot(oPaxosMsg.proposalid(), oPaxosMsg.nodeid());
if (m_poAcceptor->GetAcceptorState()->GetAcceptedBallot() != oBallot)
{
//Proposalid not same, this accept value maybe not chosen value.
PLGDebug("ProposalBallot not same to AcceptedBallot");
BP->GetLearnerBP()->OnProposerSendSuccessBallotNotSame();
return;
}
//learn value.
m_oLearnerState.LearnValueWithoutWrite(
oPaxosMsg.instanceid(),
m_poAcceptor->GetAcceptorState()->GetAcceptedValue(),
m_poAcceptor->GetAcceptorState()->GetChecksum());
BP->GetLearnerBP()->OnProposerSendSuccessSuccessLearn();
PLGHead("END Learn value OK, value %zu", m_poAcceptor->GetAcceptorState()->GetAcceptedValue().size());
TransmitToFollower();
}
正常情况下，所有节点处于online状态，共同参与paxos选举。因此，各个节点的instance id一致。为了避免冲突，paxos建议只由主节点的proposer发起提案，这样保证接受提案和习得提案编号一致。
此时，Learn习得的提案值实际上就是本节点Accept的数据，因此learner只更新内存状态即可，无需再次落盘(acceptor已落盘)。最后，如果存在follower节点，数据同步到follower(follower节点不参与paxos算法，相当于某个paxos节点的同步备)。
4.3 提案值追赶
一旦节点处于落后状态，它无法再参与到paxos提案选举中来。这时需要由learner发起主动学习完成追赶。
PhxPaxos启动时，启动learner定时器。learner定时发送learn请求到各个节点，发送请求携带本节点的Instance ID、Node ID信息。各节点收到该请求后，处理逻辑如下：
//本节点无可学习数据，跳过本次请求
if (oPaxosMsg.instanceid() >= GetInstanceID())
{
return;
}
//本节点有可学习数据，且本节点保存了此部分的paxos log
if (oPaxosMsg.instanceid() >= m_poCheckpointMgr->GetMinChosenInstanceID())
{
//通知Learner Sender发送数据
if (!m_oLearnerSender.Prepare(oPaxosMsg.instanceid(), oPaxosMsg.nodeid()))
{
BP->GetLearnerBP()->OnAskforLearnGetLockFail();
PLGErr("LearnerSender working for others.");
//如本节点正在处理其他节点的learn请求，但发起者只少了一条记录，绕过Learner Sender发送。
if (oPaxosMsg.instanceid() == (GetInstanceID() - 1))
{
PLGImp("InstanceID only difference one, just send this value to other.");
//send one value
AcceptorStateData oState;
int ret = m_oPaxosLog.ReadState(m_poConfig->GetMyGroupIdx(), oPaxosMsg.instanceid(), oState);
if (ret == 0)
{
BallotNumber oBallot(oState.acceptedid(), oState.acceptednodeid());
SendLearnValue(oPaxosMsg.nodeid(), oPaxosMsg.instanceid(), oBallot, oState.acceptedvalue(), 0, false);
}
}
//如果Leaner Sender正在处理其他节点请求，跳过本次请求
return;
}
}
//本节点有可学习数据，并且本节点当前并未处理其他节点的Learn请求，发送本节点信息给发起者
SendNowInstanceID(oPaxosMsg.instanceid(), oPaxosMsg.nodeid());
当接收节点有可学习数据，并且当前并未处理其他节点的Learn请求，发送如下信息包括：本节点的node id、当前的instance id、min chosen instance id、请求的instance id；除此之外，当paxos log差距超过50，那么两边的主节点信息、集群信息可能不同，发送此部分数据。发起者可能依次接收到多个节点请求，需要二次确认：
//当初请求的instance id和本节点已不一致，说明已通过其他方式开始习得
if (oPaxosMsg.instanceid() != GetInstanceID())
{
PLGErr("Lag msg, skip");
return;
}
//本节点已经追赶甚至超过了接受节点的paxos log，跳过
if (oPaxosMsg.nowinstanceid() <= GetInstanceID())
{
PLGErr("Lag msg, skip");
return;
}
//本节点数据落后太多，接受节点已无完整的paxos log数据，尝试进入Checkpoint模式
if (oPaxosMsg.minchoseninstanceid() > GetInstanceID())
{
BP->GetCheckpointBP()->NeedAskforCheckpoint();
PLGHead("my instanceid %lu small than other's minchoseninstanceid %lu, other nodeid %lu",
GetInstanceID(), oPaxosMsg.minchoseninstanceid(), oPaxosMsg.nodeid());
AskforCheckpoint(oPaxosMsg.nodeid());
}
else if (!m_bIsIMLearning)
{
//接受节点有可习得的完整数据，向接受节点确认习得请求
ComfirmAskForLearn(oPaxosMsg.nodeid());
}
其实上面做了这么多，简单讲就是：如果本节点在请求这段时间内并未进行自主学习，并且也没有其他节点在此之前通知本节点开始学习，本节点发送确认请求到learn接收者，准备开始接受learn数据。learn接受者收到确认请求激活learner sender线程，发送paxos log数据。
4.4 Learner Sender
Learner Sender发送本节点数据到请求节点。为了避免learn操作造成网络拥塞，在数据发送前做了限流。
void LearnerSender :: SendLearnedValue(const uint64_t llBeginInstanceID, const nodeid_t iSendToNodeID)
{
PLGHead("BeginInstanceID %lu SendToNodeID %lu", llBeginInstanceID, iSendToNodeID);
uint64_t llSendInstanceID = llBeginInstanceID;
int ret = 0;
uint32_t iLastChecksum = 0;
//control send speed to avoid affecting the network too much.
int iSendQps = LearnerSender_SEND_QPS;
int iSleepMs = iSendQps > 1000 ? 1 : 1000 / iSendQps;
int iSendInterval = iSendQps > 1000 ? iSendQps / 1000 + 1 : 1;
PLGDebug("SendQps %d SleepMs %d SendInterval %d AckLead %d",
iSendQps, iSleepMs, iSendInterval, m_iAckLead);
int iSendCount = 0;
while (llSendInstanceID < m_poLearner->GetInstanceID())
{
ret = SendOne(llSendInstanceID, iSendToNodeID, iLastChecksum);
if (ret != 0)
{
PLGErr("SendOne fail, SendInstanceID %lu SendToNodeID %lu ret %d",
llSendInstanceID, iSendToNodeID, ret);
return;
}
if (!CheckAck(llSendInstanceID))
{
return;
}
iSendCount++;
llSendInstanceID++;
ReleshSending();
if (iSendCount >= iSendInterval)
{
iSendCount = 0;
Time::MsSleep(iSleepMs);
}
}
//succ send, reset ack lead.
m_iAckLead = LearnerSender_ACK_LEAD;
PLGImp("SendDone, SendEndInstanceID %lu", llSendInstanceID);
}
注意，这里的while循环判定逻辑为while (llSendInstanceID < m_poLearner->GetInstanceID())，即只追赶到当前instance id的前一条。这是因为learner当前所处的instance id可能尚未完成提案，但这之前的提案是已选中(chosen)的。单条paxos log发送逻辑如下：
int LearnerSender :: SendOne(const uint64_t llSendInstanceID, const nodeid_t iSendToNodeID, uint32_t & iLastChecksum)
{
BP->GetLearnerBP()->SenderSendOnePaxosLog();
AcceptorStateData oState;
int ret = m_poPaxosLog->ReadState(m_poConfig->GetMyGroupIdx(), llSendInstanceID, oState);
if (ret != 0)
{
return ret;
}
BallotNumber oBallot(oState.acceptedid(), oState.acceptednodeid());
ret = m_poLearner->SendLearnValue(iSendToNodeID, llSendInstanceID, oBallot, oState.acceptedvalue(), iLastChecksum);
iLastChecksum = oState.checksum();
return ret;
}
4.5 更快的对齐数据
下述文字截取自《微信自研生产级paxos类库PhxPaxos实现原理介绍》。
上文说到当各台机器的当前运行实例编号不一致的时候，就需要Learner介入工作来对齐数据了。Learner通过其他机器拉取到当前实例的chosen value，从而跳转到下一编号的实例，如此反复最终将自己的实例编号更新到与其他机器一致。那么这里学习一个实例的网络延时代价是一个RTT。可能这个延迟看起来还不错，但是当新的数据仍然通过一个RTT的代价不断写入的时候，而落后的机器仍然以一个RTT来进行学习，这样会出现很难追上的情况。
这里需要改进，我们可以提前获取差距，批量打包进行学习，比如A机器Learner记录当前实例编号是x，B机器是y，而x < y，那么B机器通过通信获取这个差距，将(x,y]的chosen value一起打包发送给A机器，A机器进行批量的学习。这是一个很不错的方法。
但仍然不够快，当落后的数据极大，B机器发送数据需要的网络耗时也将变大，那么发送数据的过程中，A机器处于一种空闲状态，由于paxos另外一个瓶颈在于写盘，如果不能利用这段时间来进行写盘，那性能仍然堪忧。我们参考流式传输，采用类似的方法实现Learner的边发边学，B机器源源不断的往A机器输送数据，而A机器只需要收到一个实例最小单元的包体，即可立即解开进行学习并完成写盘。
具体的实现大概是先进行一对一的协商，建立一个Session通道，在Session通道里直接采用直塞的方式无脑发送数据。当然也不是完全的无脑，Session通过心跳机制进行维护，一旦Session断开即停止发送。
“我们参考流式传输，采用类似的方法实现Learner的边发边学，B机器源源不断的往A机器输送数据，而A机器只需要收到一个实例最小单元的包体，即可立即解开进行学习并完成写盘。”。这部分实际上封装在网络层，来看如何做到A机器接收到一个最小的实例单元：
int MessageEvent :: OnRead()
{
if (m_iLeftReadLen > 0)
{
return ReadLeft();
}
int iReadLen = m_oSocket.receive(m_sReadHeadBuffer + m_iLastReadHeadPos, sizeof(int) - m_iLastReadHeadPos);
if (iReadLen == 0)
{
BP->GetNetworkBP()->TcpOnReadMessageLenError();
PLErr("read head fail, readlen %d, socket broken", iReadLen);
return -1;
}
m_iLastReadHeadPos += iReadLen;
if (m_iLastReadHeadPos < (int)sizeof(int))
{
PLImp("head read pos %d small than sizeof(int) %zu", m_iLastReadHeadPos, sizeof(int));
return 0;
}
m_iLastReadHeadPos = 0;
int niLen = 0;
int iLen = 0;
memcpy((char *)&niLen, m_sReadHeadBuffer, sizeof(int));
iLen = ntohl(niLen) - 4;
if (iLen < 0 || iLen > MAX_VALUE_SIZE)
{
PLErr("need to read len wrong %d", iLen);
return -2;
}
m_oReadCacheBuffer.Ready(iLen);
m_iLeftReadLen = iLen;
m_iLastReadPos = 0;
//second read maybe no data read, so readlen == 0 is ok.
bool bAgain = false;
iReadLen = m_oSocket.receive(m_oReadCacheBuffer.GetPtr(), iLen, &bAgain);
if (iReadLen == 0)
{
if (!bAgain)
{
PLErr("second read data fail, readlen %d, no again, socket broken", iReadLen);
return -1;
}
else
{
PLErr("second read data, readlen %d need again", iReadLen);
return 0;
}
}
if (iReadLen == iLen)
{
ReadDone(m_oReadCacheBuffer, iLen);
m_iLeftReadLen = 0;
m_iLastReadPos = 0;
}
else if (iReadLen < iLen)
{
m_iLastReadPos = iReadLen;
m_iLeftReadLen = iLen - iReadLen;
PLImp("read buflen %d small than except len %d", iReadLen, iLen);
}
else
{
PLErr("read buflen %d large than except len %d", iReadLen, iLen);
return -2;
}
return 0;
}
OnRead函数由以下两部分组成：
- 首先读取数据包大小，这个过程可能需要分多次完成。
- 读取指定数据包大小的数据，这部分也可能需要分多次完成。
当单个数据包读完，已获得完整的“最小实例单元”，通过ReadDone将数据包发往Node节点处理。
至于心跳，其实就是PhxPaxos中的Ack机制。在SendLearnedValue中，每发送一条记录需要执行一次CheckAck。如果检查失败，将终止发送。每条记录发送后，learner sender要求对端发送一个异步的ack请求，这个过程是异步的。CheckAck逻辑如下：
const bool LearnerSender :: CheckAck(const uint64_t llSendInstanceID)
{
m_oLock.Lock();
if (llSendInstanceID < m_llAckInstanceID)
{
m_iAckLead = LearnerSender_ACK_LEAD;
PLGImp("Already catch up, ack instanceid %lu now send instanceid %lu",
m_llAckInstanceID, llSendInstanceID);
m_oLock.UnLock();
return false;
}
//如果当前发送的instance id和对端已确认的instance id差距超过了设定值
while (llSendInstanceID > m_llAckInstanceID + m_iAckLead)
{
uint64_t llNowTime = Time::GetSteadyClockMS();
uint64_t llPassTime = llNowTime > m_llAbsLastAckTime ? llNowTime - m_llAbsLastAckTime : 0;
//同时，超过了设定的超时时间，提前终止
if ((int)llPassTime >= LearnerSender_ACK_TIMEOUT)
{
BP->GetLearnerBP()->SenderAckTimeout();
PLGErr("Ack timeout, last acktime %lu now send instanceid %lu",
m_llAbsLastAckTime, llSendInstanceID);
CutAckLead();
m_oLock.UnLock();
return false;
}
BP->GetLearnerBP()->SenderAckDelay();
//PLGErr("Need sleep to slow down send speed, sendinstaceid %lu ackinstanceid %lu",
//llSendInstanceID, m_llAckInstanceID);
m_oLock.WaitTime(20);
}
m_oLock.UnLock();
return true;
}
4.6 关于选中(chosen)
在整个讲解learner中，我们一直在强调选中(chosen)提案或者选中(chosen)值。这个过程提出以下疑问，并尝试解答。
问：learner如何知道某个提案是否被选中呢？
答：“4.2 选中通知”是通知各个learner值是否已被选中的一种常规方式。
问：如果某个值被选中后，提案发起节点异常，选中消息未发出会如何？
答：重新发起选举，新的被选中的提案编号不同，但提案值保持不变。
问：“4.2 选中通知”中，只更新了内存状态，在持久化的数据中，如何区分一个值的状态是accept还是chosen呢？
答： 如果当前的instance id为N，在N-1之前的所有提案值都是chosen的。当instance id为N的提案值可能是chosen状态、也可能是accept状态。
问：为何非chosen状态的数据也需要落盘
答：参见P2C的不变性，即Prepare和Accept阶段做过的承诺、接受过的值即便节点重启等异常情况下也需要保持不变。
问：我还有一个关于Instance类初始化的问题。
答：Instance类本章尚未涉及，留在第七章解答吧。
4.6 总结
经过第三、第四两章的讲解，paxos协议的算法实现已经分析完成。PhxPaxos并未对paxos做任何变种，甚至还做了一点简化。
下一节，我们将介绍paxos made simple中另外一个重要的概念：状态机。当然，这里提到的“简化”也会提及。
【转载请注明】随安居士. 4. PhxPaxos源码分析之Learner. 2017.11.15
