专栏首页CNCFTiKV 源码解析系列文章(十八)Raft Propose 的 Commit 和 Apply 情景分析

TiKV 源码解析系列文章(十八)Raft Propose 的 Commit 和 Apply 情景分析

在学习了 前面的文章 之后,相信大家已经对 TiKV 使用的 Raft 核心库 raft-rs 有了基本的了解。raft-rs 实现了 Raft Leader election 和 Log replication 等核心功能,而消息的发送、接收、应用到状态机等操作则需要使用者自行实现,本文将要介绍的就是 TiKV 中这些部分的处理过程。

Raft Ready

在开始正题之前,我们先简单回顾一下 raft-rs 与外部代码的交互接口: Ready。Ready 结构的 定义 如下:

pub struct Ready {
    /// The current volatile state of a Node.
    /// SoftState will be nil if there is no update.
    /// It is not required to consume or store SoftState.
    ss: Option<SoftState>,

    /// The current state of a Node to be saved to stable storage BEFORE
    /// Messages are sent.
    /// HardState will be equal to empty state if there is no update.
    hs: Option<HardState>,

    /// States can be used for node to serve linearizable read requests locally
    /// when its applied index is greater than the index in ReadState.
    /// Note that the read_state will be returned when raft receives MsgReadIndex.
    /// The returned is only valid for the request that requested to read.
    read_states: Vec<ReadState>,

    /// Entries specifies entries to be saved to stable storage BEFORE
    /// Messages are sent.
    entries: Vec<Entry>,

    /// Snapshot specifies the snapshot to be saved to stable storage.
    snapshot: Snapshot,

    /// CommittedEntries specifies entries to be committed to a
    /// store/state-machine. These have previously been committed to stable
    /// store.
    pub committed_entries: Option<Vec<Entry>>,

    /// Messages specifies outbound messages to be sent AFTER Entries are
    /// committed to stable storage.
    /// If it contains a MsgSnap message, the application MUST report back to raft
    /// when the snapshot has been received or has failed by calling ReportSnapshot.
    pub messages: Vec<Message>,

    must_sync: bool,
}

Ready 结构包括了一些系列 Raft 状态的更新,在本文中我们需要关注的是:

  • hs: Raft 相关的元信息更新,如当前的 term,投票结果,committed index 等等。
  • committed_entries: 最新被 commit 的日志,需要应用到状态机中。
  • messages: 需要发送给其他 peer 的日志。
  • entries: 需要保存的日志。

Proposal 的接收和在 Raft 中的复制

TiKV 3.0 中引入了类似 Actor 的并发模型,Actor 被视为并发运算的基本单元:当一个 Actor 接收到一则消息,它可以做出一些决策、创建更多的 Actor、发送更多的消息、决定要如何回答接下来的消息。每个 TiKV 上的 Raft Peer 都对应两个 Actor,我们把它们分别称为 PeerFsm 和 ApplyFsm。PeerFsm 用于接收和处理其他 Raft Peer 发送过来的 Raft 消息,而 ApplyFsm 用于将已提交日志应用到状态机。

TiKV 中实现的 Actor System 被称为 BatchSystem,它使用几个 Poll 线程从多个 Mailbox 上拉取一个 Batch 的消息,再分别交由各个 Actor 来执行。为了保证 线性一致性 ,一个 Actor 同时只会在一个 Poll 线程上接收消息并顺序执行。由于篇幅所限,这一部分的实现在这里不做详述,感兴趣的同学可以在 raftstore/fsm/batch.rs 查看详细代码。

上面谈到,PeerFsm 用于接收和处理 Raft 消息。它接收的消息为 PeerMsg,根据消息类型的不同会有不同的处理:

/// Message that can be sent to a peer.
pub enum PeerMsg {
    /// Raft message is the message sent between raft nodes in the same
    /// raft group. Messages need to be redirected to raftstore if target
    /// peer doesn't exist.
    RaftMessage(RaftMessage),
    /// Raft command is the command that is expected to be proposed by the
    /// leader of the target raft group. If it's failed to be sent, callback
    /// usually needs to be called before dropping in case of resource leak.
    RaftCommand(RaftCommand),
    /// Result of applying committed entries. The message can't be lost.
    ApplyRes { res: ApplyTaskRes },
    ...
}

...

impl PeerFsmDelegate {
    pub fn handle_msgs(&mut self, msgs: &mut Vec<PeerMsg>) {
        for m in msgs.drain(..) {
            match m {
                PeerMsg::RaftMessage(msg) => {
                    self.on_raft_message(msg);
                }
                PeerMsg::RaftCommand(cmd) => {
                    self.propose_raft_command(cmd.request, cmd.callback)
                }
                PeerMsg::ApplyRes { res } => {
                    self.on_apply_res(res);
                }
                ...
            }
        }
    }
}

这里只列出了我们需要关注的几种消息类型:

  • RaftMessage: 其他 Peer 发送过来 Raft 消息,包括心跳、日志、投票消息等。
  • RaftCommand: 上层提出的 proposal,其中包含了需要通过 Raft 同步的操作,以及操作成功之后需要调用的 callback 函数。
  • ApplyRes: ApplyFsm 在将日志应用到状态机之后发送给 PeerFsm 的消息,用于在进行操作之后更新某些内存状态。

我们主要关注的是 PeerFsm 如何处理 Proposal,也就是 RaftCommand 的处理过程。在进入到 PeerFsmDelegate::propose_raft_command 后,首先会调用 PeerFsmDelegate::pre_propose_raft_command 对 peer ID, peer term, region epoch (region 的版本,region split、merge 和 add / delete peer 等操作会改变 region epoch) 是否匹配、 peer 是否 leader 等条件进行一系列检查,并根据请求的类型(是读请求还是写请求),选择不同的 Propose 策略见( Peer::inspect):

let policy = self.inspect(&req);
let res = match policy {
    Ok(RequestPolicy::ReadIndex) => return self.read_index(ctx, req, err_resp, cb),
    Ok(RequestPolicy::ProposeNormal) => self.propose_normal(ctx, req),
    ...
};

对于读请求,我们只需要确认此时 leader 是否真的是 leader 即可,一个较为轻量的方法是发送一次心跳,再检查是否收到了过半的响应,这在 raft-rs 中被称为 ReadIndex (关于 ReadIndex 的介绍可以参考 这篇文章 )。对于写请求,则需要 propose 一条 Raft log,这是在 propose_normal 函数中调用 Raft::propose 接口完成的。在 propose 了一条 log 之后,Peer 会将 proposal 保存在一个名为 apply_proposals 的 Vec 中。随后一个 Batch (包含了多个 Peer)内的 proposal 会被 Poll 线程统一收集起来,放入一个名为 pending_proposals 的 Vec 中待后续处理。

在一个 Batch 的消息都经 PeerDelegate::handle_msgs 处理完毕之后,Poll 对 Batch 内的每一个 Peer 调用 Peer::handle_raft_ready_append:

  1. 用记录的 last_applied_index 获取一个 Ready。
  2. 在得到一个 Ready 之后,PeerFsm 就会像我们前面所描述的那样,调用 PeerStorage::handle_raft_ready 更新状态(term,last log index 等)和日志。
  3. 这里的状态更新分为持久化状态和内存状态,持久化状态的更新被写入到一个 WriteBatch 中,内存状态的更新则会构造一个 InvokeContext,这些更新都会被一个 PollContext 暂存起来。

于是我们得到了 Batch 内所有 Peer 的状态更新,以及最近提出的 proposal,随后 Poll 线程会做以下几件事情:

  1. 将 Proposal 发送给 ApplyFsm 暂存,以便在 Proposal 写入成功之后调用 Callback 返回响应。
  2. 将之前从各个 Ready 中得到的需要发送的日志发送给 gRPC 线程,随后发送给其他 TiKV 节点。
  3. 持久化已保存在 WriteBatch 中需要更新的状态。
  4. 根据 InvokeContext 更新 PeerFsm 中的内存状态。
  5. 将已提交日志发送给 ApplyFsm 进行应用(见 Peer::handle_raft_ready_apply)。

Proposal 在 Raft 中的确认

上面我们阐述了 Region 的 Leader 在收到 proposal 之后,是调用了哪些接口将 proposal 放到 Raft 状态机中的。在这之后,这个 proposal 虽然被发往了 ApplyFsm 中暂存,但是 ApplyFsm 目前还不能 apply 它并调用关联的 callback 函数,因为这个 proposal 还没被 Raft 中的过半节点确认。那么,Leader 节点上的 raftstore 模块是如何处理收到的其他副本的 Raft 消息,并完成日志的确认的呢?

答案就在 PeerFsmDelegate::on_raft_message 函数中。在一个 Peer 收到 Raft 消息之后,会进入这个函数中进行处理,内部调用 Raft::step 函数更新 Raft 状态机的内存状态。之后,调用 RawNode::ready 函数获取 committed_entries,最终作为 ApplyMsg::Apply 任务发送给 ApplyFsm,由 ApplyFsm 执行指令,如果 proposal 是由本节点发出,还会调用 callback 函数(之前通过 ApplyMsg::Proposal 任务暂存在 ApplyFsm 中)以向客户端返回响应。

Proposal 的应用

在上一部分我们提到,PeerFsm 会将 Proposal 以及已提交日志发送给对应的 ApplyFsm,它们对应的消息类型分别是 ApplyMsg::Proposal 和 ApplyMsg::Apply,下面将会讲述 ApplyFsm 是如何处理这两种类型的消息的。

对于 ApplyMsg::Proposal 的处理非常简单(见 ApplyFsm::handle_proposal),ApplyFsm 会把 Proposal 放入 ApplyDelegate::pending_cmds 中暂存起来,后续在应用对应的日志时会从这里找出相应的 Callback 进行调用。

而 ApplyMsg:Apply 中包含的是实际需要应用的日志,ApplyFsm 会针对这些日志进行(见 ApplyFsm::handle_apply):

  1. 修改内存状态,将变更的状态(last applied index 等)、数据持久化。
  2. 调用 Proposal 对应的 Callback 返回响应。
  3. 向 PeerFsm 发送 ApplyRes,其中包含了 applied_term、applied_index 等状态(用于更新 PeerFsm 中的内存状态)。

这里存在一个特殊情况,就是所谓的“空日志”。在 raft-rs 的实现中,当选举出新的 Leader 时,新 Leader 会广播一条“空日志”,以提交前面 term 中的日志(详情请见 Raft 论文)。此时,可能还有一些在前面 term 中提出的 proposal 仍然处于 pending 阶段,而因为有新 Leader 产生,这些 proposal 永远不可能被确认了,因此我们需要对它们进行清理,以免关联的 callback 无法调用导致一些资源无法释放。清理的逻辑参照 ApplyFsm::handle_entries_normal 函数。

总结

这里用一个流程图总结一下 TiKV 处理 Proposal 的大致流程,如下:

简言之,TiKV 使用了两个线程池来处理 Proposal,并且将一个 Raft Peer 分成了两部分:PeerFsm 和 ApplyFsm。在处理 Proposal 的过程中,首先由 PeerFsm 获取日志并驱动 Raft 内部的状态机,由 ApplyFsm 根据已提交日志修改对应数据的状态机(region 信息和用户数据)。

由于这部分代码涉及到各种 corner case 的处理,因此逻辑较为复杂,希望感兴趣的读者可以进一步从源码中获取更多细节。

本文分享自微信公众号 - CNCF(lf_cncf)

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2020-03-24

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 用Jaeger做数据分析|跟踪告诉我们更多!

    我将直接切入主题,Jaeger目前只可视化收集来自测仪应用程序的数据。它不执行任何后处理(除了服务依赖关系图)或任何计算,以从它收集的跟踪中获得其他有趣的指标或...

    CNCF
  • PlanetScale迁移开源Vitess测试套件|从Python到Go

    作者:Deepthi Sigireddi。嘉宾博客文章最初在PlanetScale的博客发表。

    CNCF
  • 你知道你的Docker镜像里有什么吗?Buildpacks知道。

    你不能修补一个你不知道你有的漏洞。这就是为什么了解Docker镜像中的内容是确保其安全性的第一步。幸运的是,任何使用Cloud Native Buildpack...

    CNCF
  • TiKV 源码解析系列文章(十八)Raft Propose 的 Commit 和 Apply 情景分析

    在学习了 前面的文章 之后,相信大家已经对 TiKV 使用的 Raft 核心库 raft-rs 有了基本的了解。raft-rs 实现了 Raft Leader ...

    PingCAP
  • Golang蠕虫泛滥?让我们揪出其始作俑者

    近日,国外安全网站SECURITYWEEK披露,一款Go语言恶意软件正大量感染Linux服务器,其使用了多达6种传播感染方式,包含4个远程执行漏洞(ThinkP...

    FB客服
  • Log4j2日志框架

    log4j2是一个比较新的日志框架,作为log4j的升级版本,修复了它的锁竞争问题提升了性能,提供了丰富的组件支持以及良好的语义配置。

    李鸿坤
  • 空间转录组学(Spatial Transcriptomics)

    前面给大家简单的科普了空间转录组,可能还是有些人对背后的技术原理有些疑惑,那么我们深入的探讨一下技术原理。文末有视频会更直观。

    生信交流平台
  • 科学计算库Numpy——随机模块

    py3study
  • Spark Core源码精读计划 22 | BlockInfoManager与其实现的块锁机制

    在上一篇文章中,我们对与块相关的BlockId、BlockData和BlockInfo有了比较全面的理解。前面已经提到过,块在读写时有锁机制,并且委托给Bloc...

    暴走大数据
  • Spring+SpringMVC+MyBatis+easyUI整合优化篇(一)Java语言中System.out.print与Log的比较

    作者:13 GitHub:https://github.com/ZHENFENG13 版权声明:本文为原创文章,未经允许不得转载。 前言 距离上一次更新博...

    我是十三

扫码关注云+社区

领取腾讯云代金券