braft架构
braft分为几个板块，我根据自己的理解进行了概述：
ballot:
ballot，ballot_box
选票类，raft的选举leader步骤
cli:
cli，cli_service
客户端，cli用来修改或者获取复制状态机组节点状态，cli_service继承cli
closure：
closure_helper，closure_queue
返回时的信息和报告内容
configuration：
configuration，configuration_manager
配置信息类，peerid类，nodeid类
configuration_manager用来记录和管理configuration条目历史变更
file:
file_reader:读取文件内容
file_service:为状态机增加或者移除一个file_reader
file_system_adaptor：文件系统适配
protobuf_file：
remote_file_copier：远程文件复制，存储使用快照
fsm_caller:
复制状态机，发生事件时的操作
fsync:
参数选择内存数据落到磁盘的方式，直接调用mingw函数
lease:
作为leader节点和follower节点的状态
log:
log:日志类，log定义和LogStorage定义
log_entry:每条日志条目
log_manager：用于管理日志条目
memory_log：日志条目索引相关操作
node:
node：节点类，每一个node是一个复制状态机的节点实例
node_manager：node节点管理
raft（核心）:
raft:定义task，状态机操作，节点参数，状态机参数等
raft_meta：
raft_service
repeated_timer_task:
重复任务计时器
replicator：
leader，follower之间的信息复制
route_table：
一张通信表，存多个复制组，可获取每个复制组leader，更新组配置，等
snapshot：
snapshot：快照类，
snapshot_executor：快照执行操作
snapshot_throttle：大容量磁盘读/写
storage：
存储类，包含元数据存储，log存储，快照存储，稳定存储等
util：
放一些常用的公用方法
之后对我认为比较重要的，或者官方文档中出现的头文件进行注释翻译
configuration.h
| 变量名 | 类型 | 说明 |
|---|---|---|
| GroupId | typedef std::string |
raft组名 |
| VersionedGroupId |
typedef std::string |
GroupId加个版本号 {group_id}_{index} |
=======struct PeerId：=======
butil::EndPoint addr;//就是ip+port
int idx;//idx算是下标号，默认0
除了构造函数外，
reset()：初始化值为0
is_empty()：都为0时认为是空的
parse()：解析字符串并赋值
to_string()：将值转成字符串，ip:port:idx
<,==,!=,<<操作符被重载
=======struct NodeId：=======
GroupId group_id;//存组名
PeerId peer_id;//存一个peerid
有一个两值都传的构造函数
to_string()：将值转成字符串,group_id:ip:port:idx
<,==,!=,<<操作符被重载
=======class Configuration：=======
std::set<PeerId> _peers；//是一个peers集合
typedef std::set<PeerId>::const_iterator const_iterator；//一个循环集合的迭代器
两个构造函数，可以用vector和set来构造
重载了=符号（赋值）
void reset()：清空集合
bool empty()：是否为空
size_t size() ：返回大小
const_iterator begin() / end()：返回迭代器头部 / 迭代器尾部
void list_peers(std::set<PeerId>* peers) / list_peers(std::vector<PeerId>* peers)：使用参数替换掉当前集合
void append_peers(std::set<PeerId>* peers)：将一系列集合插入到当前集合
bool add_peer(const PeerId& peer):插入一个新节点并判断是否插入成功（是否为新peer）
bool remove_peer(const PeerId& peer):移除节点，并判断是否移除
bool contains(const PeerId& peer_id) /contains(const std::vector<PeerId>& peers)：判断是否存在这些peer
bool equals(const std::vector<PeerId>& peers) / equals(const Configuration& rhs): 判断是否和参数匹配
void diffs(const Configuration& rhs,
Configuration* included,
Configuration* excluded)：得到当前值和rhs的不同，included为this-rhs,excluded为rhs-this
int parse_from(butil::StringPiece conf):从字符串解析并赋值（peerid之间用,隔开）
重载了<<输出符号，输出peerid，用逗号隔开
raft.h
#ifndef BRAFT_RAFT_H
#define BRAFT_RAFT_H
//一堆头文件
#include <string>#include <butil/logging.h>
#include <butil/iobuf.h>
#include <butil/status.h>
#include <brpc/callback.h>
#include "braft/configuration.h"
#include "braft/enum.pb.h"
#include "braft/errno.pb.h"template <typename T> class scoped_refptr;//一个模板类scoped_refptrnamespace brpc {
class Server;
} // namespace brpcnamespace braft {
//给出这些类的声明，定义在其他c文件中
class SnapshotWriter;//快照写
class SnapshotReader;//快照读
class SnapshotHook;//
class LeaderChangeContext;//leader变更信息
class FileSystemAdaptor;//文件适配器
class SnapshotThrottle;//快照大磁盘
class LogStorage;//log存储
//给出一个默认值，会用来做初始化
const PeerId ANY_PEER(butil::EndPoint(butil::IP_ANY, 0), 0);// 特定于raft的closure(返回报告)，包含一个butil :: Status以报告操作是否成功。
class Closure : public google::protobuf::Closure {
public://两个获取status值的函数butil::Status& status() { return _st; }const butil::Status& status() const { return _st; }private:butil::Status _st;//status
};// 描述一个特定的错误
class Error {
public://构造函数Error() : _type(ERROR_TYPE_NONE) {}Error(const Error& e) : _type(e._type), _st(e._st) {}//返回值ErrorType type() const { return _type; }const butil::Status& status() const { return _st; }butil::Status& status() { return _st; }//设置error类型void set_type(ErrorType type) { _type = type; }//重载赋值号Error& operator=(const Error& rhs) {_type = rhs._type;_st = rhs._st;return *this;}
private:// 需要赋值ErrorType _type;butil::Status _st;
};
//将error类型转为字符串
inline const char* errortype2str(ErrorType t) {switch (t) {case ERROR_TYPE_NONE:return "None";case ERROR_TYPE_LOG:return "LogError";case ERROR_TYPE_STABLE:return "StableError";case ERROR_TYPE_SNAPSHOT:return "SnapshotError";case ERROR_TYPE_STATE_MACHINE:return "StateMachineError";}return "Unknown";
}
//重载输出符号
inline std::ostream& operator<<(std::ostream& os, const Error& e) {os << "{type=" << errortype2str(e.type()) << ", error_code=" << e.status().error_code()<< ", error_text=`" << e.status().error_cstr()<< "'}";return os;
}// libraft的基本消息结构
struct Task {//任务Task() : data(NULL), done(NULL), expected_term(-1) {}//构造函数// 数据应用于状态机butil::IOBuf* data;// 当数据应用于状态机或发生错误时继续。Closure* done;// 如果expected_term与当前值不匹配，则拒绝此任务// 此节点，如果值不为-1// 默认值：-1int64_t expected_term;
};class IteratorImpl;//类声明// Iterator over a batch of committed tasks
//
// Example:
// void YouStateMachine::on_apply(braft::Iterator& iter) {
// for (; iter.valid(); iter.next()) {
// brpc::ClosureGuard done_guard(iter.done());
// process(iter.data());
// }
// }
class Iterator {DISALLOW_COPY_AND_ASSIGN(Iterator);//关闭隐式调用构造函数
public:// 移至下一个任务。void next();// 返回当前的唯一且单调递增的索引// 任务:// - 唯一性保证了，在不同的peers中具有相同索引的已提交的任务始终相同且保持不变// - 单调性确保在小组中的所有peers中，对于任何索引对i，j（i < j），// 索引为| i |的任务必须在索引| j |的任务之前应用int64_t index() const;// 返回任务应用时的leader任期int64_t term() const;// 返回与在leader节点中传递给Node::apply的内容相同的数据const butil::IOBuf& data() const;// 如果done()为非空，无论此操作成功或失败，您都必须在应用此任务后// 调用done()->Run()，否则将泄漏相应的资源。//// 如果此节点在担任该组的负责人时提出此任务，并且领导位置在此之前没有改变,// 那么done()正是传递给Node::apply的内容，// 在使用给定任务更新StateMachine之后，done()可能代表某种延续（例如响应客户端）。// 否则done()必须为NULLClosure* done() const;// 如果此迭代器当前引用的是有效任务，返回true，// 否则返回false,说明迭代器到达了这批任务的结尾，或者说明发生了些错误bool valid() const;// 发生某些严重错误时调用。// 并且我们认为最后的|ntail|任务（从上一个迭代的任务开始）没有被应用//// 如果| st | 不是NULL，它将描述错误的详细信息。void set_error_and_rollback(size_t ntail = 1, const butil::Status* st = NULL);private:
friend class FSMCaller;//友元类声明Iterator(IteratorImpl* impl) : _impl(impl) {}//构造函数~Iterator() {};//析构函数// _impl的所有权属于FSMCaller；IteratorImpl* _impl;
};// | StateMachine | 是一个raft节点的所有事件的接收器。
// 为您自己的业务逻辑实现特定的StateMachine。
// 注意：不需要所有接口都是线程安全的，它们是按顺序调用的，就是说每个方法都会阻塞之后的所有方法。
class StateMachine {
public:virtual ~StateMachine();//析构函数，需要继承并实现// 使用一批可以通过| iterator |访问的任务,来更新StateMachine。//// 当传递给Node :: apply的一个或多个任务已提交到raft组时（法定个数已收到这些任务，并将其存储在后备存储中）调用。//// 一旦该函数返回给调用者，我们认为所有通过| iter |处理迭代的任务已成功应用。 // 而且，如果没有应用所有给定的任务，我们会将其视为严重错误，并报告错误类型为ERROR_TYPE_STATE_MACHINEvirtual void on_apply(::braft::Iterator& iter) = 0;//需要继承实现// 当raft节点崩溃关闭时调用一次。// 不实现默认啥都不做virtual void on_shutdown();// 用户定义的快照生成功能，此方法将阻塞on_apply。// 当fsm可以是cow（写时复制）时，用户可以使快照异步。// 快照完成后，调用done-> Run（）。// 成功返回0，失败返回error.virtual void on_snapshot_save(::braft::SnapshotWriter* writer,::braft::Closure* done);// 用户定义的快照加载功能// 获取并加载快照// 成功返回0，失败返回errno// 默认值：不加载任何内容并返回错误。virtual int on_snapshot_load(::braft::SnapshotReader* reader);// 当所属节点在| term |成为组的leader时调用// 默认值：不执行任何操作virtual void on_leader_start(int64_t term);// 当所属节点不是组的leader时调用// |status| 描述详细信息virtual void on_leader_stop(const butil::Status& status);// 当遇到严重错误时，将调用on_error，在这之后，在错误修复并重新启动该节点之前，不允许对该节点进行任何进一步的修改。virtual void on_error(const ::braft::Error& e);// 将配置提交给组后调用virtual void on_configuration_committed(const ::braft::Configuration& conf);virtual void on_configuration_committed(const ::braft::Configuration& conf, int64_t index);// 当跟随者停止跟随某个领导者并且其leader_id变为NULL时，将调用此方法，// 情况包括：// 1.处理选举超时并开始投票// 2.接收到候选者的更高term的请求，例如vote_request或来自新leader的append_entries_request// 3.从当前领导者接收timeout_now_request并开始request_vote// 参数stop_following_context提供有关跟随者之前跟随过的领导者的信息(leader_id, term and status)// 用户可以在节点停止跟随某个领导者时重置节点的信息。virtual void on_stop_following(const ::braft::LeaderChangeContext& ctx);// 当跟随者或候选人开始跟随领导者时，将调用此方法// 其leader_id（在调用方法之前应为NULL）设置为领导者的id,// 情况包括：// 1.候选人收到领导的append_entries// 2.跟随者（无领导者）从领导者那里获得append_entries// 参数start_following_context提供有关跟随者开始跟随的领导者的信息(leader_id, term and status)// 用户可以在开始跟随某个领导者时重置节点的信息。virtual void on_start_following(const ::braft::LeaderChangeContext& ctx);
};//节点的状态
enum State {// 如果您不确定用法，请不要更改顺序。STATE_LEADER = 1,STATE_TRANSFERRING = 2,STATE_CANDIDATE = 3,STATE_FOLLOWER = 4,STATE_ERROR = 5,STATE_UNINITIALIZED = 6,STATE_SHUTTING = 7,STATE_SHUTDOWN = 8,STATE_END,
};//状态转成字符串
inline const char* state2str(State state) {const char* str[] = {"LEADER", "TRANSFERRING", "CANDIDATE", "FOLLOWER", "ERROR", "UNINITIALIZED", "SHUTTING", "SHUTDOWN", };if (state < STATE_END) {return str[(int)state - 1];} else {return "UNKNOWN";}
}// 如果| s | 指出节点处于活动状态，返回true
inline bool is_active_state(State s) {// This should be as fast as possiblereturn s < STATE_ERROR;
}// 此类封装了on_start_following和on_stop_following接口的参数。
class LeaderChangeContext {DISALLOW_COPY_AND_ASSIGN(LeaderChangeContext);//关闭隐式构造函数
public:// 构造函数LeaderChangeContext(const PeerId& leader_id, int64_t term, const butil::Status& status): _leader_id(leader_id), _term(term) , _st(status){};// 对于on_start_following，leader_id和term是新领导者的；// 对于on_stop_following，leader_id和term是旧领导者的。const PeerId& leader_id() const { return _leader_id; }int64_t term() const { return _term; }// 返回有关为什么调用on_start_following或on_stop_following的信息。const butil::Status& status() const { return _st; }private:PeerId _leader_id;int64_t _term;butil::Status _st;
};//重载输出符号
inline std::ostream& operator<<(std::ostream& os, const LeaderChangeContext& ctx) {os << "{ leader_id=" << ctx.leader_id()<< ", term=" << ctx.term()<< ", status=" << ctx.status()<< "}";return os;
}class UserLog {DISALLOW_COPY_AND_ASSIGN(UserLog);//关闭隐式构造
public:// 构造函数UserLog() {};UserLog(int64_t log_index, const butil::IOBuf& log_data): _index(log_index), _data(log_data){};// 返回日志索引号int64_t log_index() const { return _index; }// 返回日志内容const butil::IOBuf& log_data() const { return _data; }// 设置日志内容和索引void set_log_index(const int64_t log_index) { _index = log_index; }void set_log_data(const butil::IOBuf& log_data) { _data = log_data; }// 清空void reset() {_index = 0;_data.clear();}private:int64_t _index;butil::IOBuf _data;
};// 重载输出符号
inline std::ostream& operator<<(std::ostream& os, const UserLog& user_log) {os << "{user_log: index=" << user_log.log_index()<< ", data size=" << user_log.log_data().size()<< "}";return os;
}// peer的状态
struct PeerStatus {//构造函数PeerStatus(): valid(false), installing_snapshot(false), next_index(0), last_rpc_send_timestamp(0), flying_append_entries_size(0), readonly_index(0), consecutive_error_times(0){}bool valid;bool installing_snapshot;int64_t next_index;int64_t last_rpc_send_timestamp;int64_t flying_append_entries_size;int64_t readonly_index;int consecutive_error_times;
};// 节点状态
struct NodeStatus {typedef std::map<PeerId, PeerStatus> PeerStatusMap;NodeStatus(): state(STATE_END), readonly(false), term(0), committed_index(0), known_applied_index(0), pending_index(0), pending_queue_size(0), applying_index(0), first_index(0), last_index(-1), disk_index(0){}State state;PeerId peer_id;//节点的idPeerId leader_id;//;leader的idbool readonly;//是否只读int64_t term;// 两个索引，一个已提交索引，一个已应用索引int64_t committed_index;int64_t known_applied_index;// 等待提交的日志的起始索引。// 如果值为0，则表示没有待处理的日志。// // 警告：如果该值不为0，并且长时间保持该值不变，则表示发生了某些事情导致节点很有可能无法提交日志，因此用户应仔细检查以找出原因。int64_t pending_index;// 有多少日志等待提交。// // 警告：待处理的日志过多，意味着处理速度无法满足写入速度。 用户可以考虑降低写入速度，以避免资源耗尽。int64_t pending_queue_size;// 当前的应用索引。 如果值为0，则表示没有应用日志。//// 警告：如果该值不为0，并且长时间保持不变，则表示应用线程已挂起，用户应检查是否发生了死锁，或是否正在处理一些耗时的操作。int64_t applying_index;// 节点的第一个日志，包括内存和磁盘中的日志。int64_t first_index;// 节点的最后一个日志，包括内存和磁盘中的日志。int64_t last_index;// 磁盘中的最后一个日志。int64_t disk_index;// 稳定的跟随者是当前配置中的peers// 如果该节点不是领导者，则此映射为空。PeerStatusMap stable_followers;// 不稳定的跟随者是不在当前配置中的peers// 举例来说，如果添加了一个新的peer,但现在还未赶上leader，它在此映射中。PeerStatusMap unstable_followers;
};// 租赁状态。 以下是典型的租约状态更改图：
// =========================================================================
// event: become leader become follower
// ^ on leader start ^ on leader stop
// | ^ | ^
// time: ----------|-----------|-----------------|---|-------
// lease state: EXPIRED | NOT_READY | VALID | EXPIRED
// =========================================================================
enum LeaseState {// Lease is disabled，只有在以下情况下才会返回此状态// |raft_enable_leader_lease == false|.LEASE_DISABLED = 1,// Lease is expired, 此节点不再是领导者。LEASE_EXPIRED = 2,// 该节点是领导者，但我们不确定数据是最新的。 // 此状态一直持续到| on_leader_start |。 或领导下台。LEASE_NOT_READY = 3,// Lease is valid.LEASE_VALID = 4,
};// Status of a leader lease.
struct LeaderLeaseStatus {// 构造函数LeaderLeaseStatus(): state(LEASE_DISABLED), term(0), lease_epoch(0){}LeaseState state;// 这些字段仅在| state == LEASE_VALID |时才有意义。// The term of this leaseint64_t term;// 当转移leader超时时，一个特定的TERM可能具有多个租赁。 在节点的生命周期中，保证租赁时期将单调增加。int64_t lease_epoch;
};struct NodeOptions {// 如果在| election_timeout_ms |毫秒的时间里未收到来自领导者的任何消息，则追随者将成为候选人。// 默认为: 1000 (1s)int election_timeout_ms; //follower to candidate timeout// 最大时钟浮动时间，它将用于维护领导者租赁的安全。// 默认为: 1000 (1s)int max_clock_drift_ms;// 如果将其置为正数，每|snapshot_interval_s| 秒会触发一次快照保存。// 如果 |snapshot_interval_s| <= 0, 基于时间保存快照的功能将被禁用。// 默认为: 3600 (1 hour)int snapshot_interval_s;// 我们认为一个新增的peer节点已经追赶上当前进度，假如peer节点的last_log_index和// leader节点之间的差值小于 |catchup_margin|// 默认为: 1000int catchup_margin;// 如果节点从一个新环境启动（LogStorage和SnapshotStorage均为空）// 那么它会使用| initial_conf | 作为组的配置，否则它将从现有环境中加载配置。// 默认: A empty groupConfiguration initial_conf;// 在pthread而非bthread中运行user callbacks and user closures// // 默认: falsebool usercode_in_pthread;// 特定的StateMachine实现了您的业务逻辑，该业务逻辑必须是有效实例。StateMachine* fsm;// 如果|node_owns_fsm|为true，那么当不再引用节点时，|fsm|将被销毁// 默认: falsebool node_owns_fsm;// 在业务层实现的特定LogStorage，应为有效实例，否则默认使用SegmentLogStorage。//// 默认: nullLogStorage* log_storage;// 如果| node_owns_log_storage | 是true。 | log_storage | 当不再引用后备节点时，它将被销毁。//// 默认: truebool node_owns_log_storage;// 以${类型}://${参数}格式描述特定的LogStorage// 当 |log_storage| 是 null 时有效std::string log_uri;// 以${类型}://${参数}格式描述特定的RaftMetaStorage// 到目前为止提供了三种类型：// 1. type=local// FileBasedSingleMetaStorage(旧名是LocalRaftMetaStorage) 将被使用// 它基于protobuf文件，并且仅管理一个节点的稳定meta// 典型格式: local://${node_path}// 2. type=local-merged// KVBasedMergedMetaStorage 将被使用, 其底层基于KV存储，并在同一磁盘上管理一批节点。 // 当Multi-raft下拥有大量节点时,该解决方案旨在解决在领导者选举期间由非常多小的同步IO引起的性能问题。// 典型格式: local-merged://${disk_path}// 3. type=local-mixed// MixedMetaStorage 将被使用, 当升级或降级后，这将使上述两种类型的元存储重复写入。// 典型格式: // local-mixed://merged_path=${disk_path}&&single_path=${node_path}// // 升级和降级步骤：// upgrade from Single to Merged: local -> mixed -> merged// downgrade from Merged to Single: merged -> mixed -> localstd::string raft_meta_uri;// 以${类型}://${参数}格式描述特定的SnapshotStoragestd::string snapshot_uri;// 如果启用，在从远程复制快照之前，将过滤掉重复的文件，以避免无用的传输。// 仅当本地文件和远程文件中的两个文件具有相同的文件名// 和相同的checksum（存储在文件meta中）时，才会开始复制// 默认: falsebool filter_before_copy_remote;// 如果 non-null, 我们将把这个snapshot_file_system_adaptor传递给SnapshotStorage// 默认: NULLscoped_refptr<FileSystemAdaptor>* snapshot_file_system_adaptor; // 如果 non-null, 我们将把此snapshot_throttle传递给SnapshotExecutor// 默认: NULLscoped_refptr<SnapshotThrottle>* snapshot_throttle;// 如果为true，将拒绝通过raft_cli进行的RPC。// 默认:falsebool disable_cli;// 构造一个默认实例NodeOptions();
};
//构造函数的默认值
inline NodeOptions::NodeOptions() : election_timeout_ms(1000), max_clock_drift_ms(1000), snapshot_interval_s(3600), catchup_margin(1000), usercode_in_pthread(false), fsm(NULL), node_owns_fsm(false), log_storage(NULL), node_owns_log_storage(true), filter_before_copy_remote(false), snapshot_file_system_adaptor(NULL), snapshot_throttle(NULL), disable_cli(false)
{}class NodeImpl;//类声明
//节 点 类
class Node {
public://构造函数和析构函数Node(const GroupId& group_id, const PeerId& peer_id);virtual ~Node();// get node idNodeId node_id();// get leader PeerId, 重定向PeerId leader_id();// 如果这是所属组的leader，则返回truebool is_leader();// 如果这是领导者且领导者租约有效，则返回true。// | raft_enable_leader_lease == false |时，它始终为false。// 在以下情况下，返回的true是不可信的：// - 并非所有在raft组中的节点将 |raft_enable_leader_lease| 设置为 true,// 正在转移leader或者正在投票;// - 在raft组中,一个节点中的| election_timeout_ms |比另外一和节点中的|election_timeout_ms + max_clock_drift_ms| 大bool is_leader_lease_valid();// 获取领导者租赁状态以进行更复杂的检查void get_leader_lease_status(LeaderLeaseStatus* status);// 使用节点设置来初始化nodeint init(const NodeOptions& options);// 关闭本地副本。// done 是用户定义的功能, 可能响应客户端或清理一些资源// [注意] 应用后的代码无法完成访问资源void shutdown(Closure* done);// 阻塞线程，直到节点成功停止。void join();// [Thread-safe and wait-free]// 将任务应用于复制状态机//// 关于所有权:// |task.data|: 出于性能考虑，我们将删除内容。如果要保留内容，请在调用此函数之前将其复制// |task.done|: 如果data已成功提交到raft组。我们将所有权传递给StateMachine::on_apply.// 否则，我们将指定错误并调用它。void apply(const Task& task);// 列出raft组中的peer，只有leader返回ok// [注意]当list_peers与add_peer / remove_peer并发时,返回可能会被阻塞// 因为add_peer / remove_peer会立即修改内存中的配置butil::Status list_peers(std::vector<PeerId>* peers);// 在raft组中添加一个新的peer节点// 此操作完成后，将调用done-> Run（），描述结果详细信息。void add_peer(const PeerId& peer, Closure* done);// 从raft组中移除一个peer节点// 此操作完成后，将调用done-> Run（），描述结果详细信息。void remove_peer(const PeerId& peer, Closure* done);// 将raft组的配置优雅地更改为|new_peers|// 此操作完成后，将调用done-> Run（），描述结果详细信息。void change_peers(const Configuration& new_peers, Closure* done);// 单独重置此节点的配置，在此节点成为领导者之前，无需和其他peers节点有任何复制// 当大多数复制组都已失效，并且您希望在可用性方面恢复服务时，该功能应该被调用//// 请注意，在这种情况下，既不能保证一致性也不可以保证共识，在处理此方法时**请务必小心**。butil::Status reset_peers(const Configuration& new_peers);// 如果可能，立即启动快照。// 快照完成时将调用done-> Run（），描述详细结果。void snapshot(Closure* done);// 用户触发投票// 重设eletection_timeout，这样某些节点更有可能成为leaderbutil::Status vote(int election_timeout);// 对于该节点重置| election_timeout_ms | ,| max_clock_drift_ms | 也已调整// 保持| election_timeout_ms |的总和 和| max_clock_drift_ms | 不变。butil::Status reset_election_timeout_ms(int election_timeout_ms);// 强制重置| election_timeout_ms | 和| max_clock_drift_ms |。 它可能会破坏领导者的租约安全性，应小心。// 以下是安全更改 |election_timeout_ms| 的建议// 1. 三步来升级|election_timeout_ms| 到更大数值// - 在所有同级节点中放大| max_clock_drift_ms | 确保 |old election_timeout_ms + new max_clock_drift_ms|比|new election_timeout_ms + old max_clock_drift_ms|更大.// - 至少等待 |old election_timeout_ms + new max_clock_drift_ms| 的时间以确保以前所有选举都完成。// - 升级| election_timeout_ms | 到新的值，同时| max_clock_drift_ms | 可以重新设置为旧值。// 2. 三步来升级 |election_timeout_ms| 到小数值:// - 同时调整| election_timeout_ms | 和| max_clock_drift_ms |，使| election_timeout_ms + max_clock_drift_ms |的总和不变。// - 至少等待 |election_timeout_ms + max_clock_drift_ms| 的时间以确保以前的所有选举都完成。// - 升级| max_clock_drift_ms | 回到旧值void reset_election_timeout_ms(int election_timeout_ms, int max_clock_drift_ms);// 尝试将leader位置转移给| peer |。// 如果peer是ANY_PEER，下个term将选出合适的follower作为leader// 成功返回0，否则返回-1。int transfer_leadership_to(const PeerId& peer);// 从给定索引中读取第一个提交的用户日志。// 成功返回OK，并为user_log分配数据。// 请注意，user_log可能不是给定索引处的确切日志，而是从给定索引到last_committed_index的第一个可用用户日志。// 否则，将返回适当的错误：// - 当日志被删除后返回ELOGDELETED；// - 当我们即使到达了last_committed_index都无法获得用户日志时，返回ENOMOREUSERLOG。// [注意]考虑到安全性，在代码实现中我们使用last_applied_index代替last_committed_indexbutil::Status read_committed_user_log(const int64_t index, UserLog* user_log);// 获取此节点的内部状态，信息与我们在网站上看到的信息基本相同。void get_status(NodeStatus* status);// 使该节点进入只读模式。// 只读模式仅应在某些极端情况下用于保护系统。// 例如，在存储系统中，太多的写请求涌入系统// 不幸的，系统处于耗尽容量的危险中，没有足够的时间添加新计算机并等待容量平衡。一旦许多磁盘已满，raft组就会发生仲裁死亡。此示例中的一种选择是只读模式，该模式允许领导者拒绝新的写请求，但仍处理读请求和配置更改。// 如果跟随节点变为只读，则leader停止将新日志复制到该节点。 在leader仍然可写的情况下，这可能导致数据远远落后于leader。 跟随节点退出只读模式后，leader将继续复制丢失的日志。void enter_readonly_mode();// 使该节点离开只读模式。void leave_readonly_mode();// 检查此节点是否为只读。// 有两种情况，如果节点是只读的：// - 通过调用enter_readonly_mode（）将该节点标记为只读// - 此节点是leader，并且组中可写节点的数量少于大多数。bool readonly();private:NodeImpl* _impl;
};
//引导程序选项
struct BootstrapOptions {// 包含此raft组的初始成员// 默认: empty confConfiguration group_conf;// 转储快照中包含的最后一个索引// 默认: 0int64_t last_log_index;// 将要转储第一个快照的特定StateMachine// 如果last_log_index不为0，则fsm必须是有效实例。// 默认: NULLStateMachine* fsm;// 如果| node_owns_fsm | 是true。 当不再引用后备节点时，| fsm | 将被销毁。// 默认: falsebool node_owns_fsm;// 在pthread而非bthread中运行user callbacks and user closures// 默认: falsebool usercode_in_pthread;// 以${类型}://${参数}格式描述特定的LogStoragestd::string log_uri;// // 以${类型}://${参数}格式描述特定的RaftMetaStoragestd::string raft_meta_uri;// 以${类型}://${参数}格式描述特定的SnapshotStoragestd::string snapshot_uri;// 构造默认选项BootstrapOptions();};// 引导一个非空的raft节点
int bootstrap(const BootstrapOptions& options);// 将raft服务附加到| server |（自己的服务端），这使raft服务与用户服务共享相同的侦听地址。
// 注意：目前，我们仅允许使用特定的侦听地址启动备用服务器，如果要从一个范围端口启动该服务器，则行为是不确定的。
// 成功返回0，否则返回-1。
int add_service(brpc::Server* server, const butil::EndPoint& listen_addr);
int add_service(brpc::Server* server, int port);
int add_service(brpc::Server* server, const char* listen_ip_and_port);// 回收
struct GCOptions {// raft实例的Versioned-groupid。// 版本是很重要的， 因为具有相同groupid的实例在销毁后可能很快会再次创建。VersionedGroupId vgid;std::string log_uri;std::string raft_meta_uri;std::string snapshot_uri;
};// 如果磁盘被删除但是并没有从global_mss_manager释放，之后再次添加该怎么办? 看起来没问题，因为该磁盘上的所有实例在删除磁盘本身之前都会被销毁，因此不会有垃圾。
//
// 由于某种原因破坏实例时，对raft实例的数据进行回收。
//
// 成功返回0，否则返回-1
int gc_raft_data(const GCOptions& gc_options);} // namespace braft#endif //BRAFT_RAFT_H
storage.h
#ifndef BRAFT_RAFT_STORAGE_H
#define BRAFT_RAFT_STORAGE_H
// 一堆头文件
#include <string>
#include <vector>
#include <gflags/gflags.h>
#include <butil/status.h>
#include <butil/class_name.h>
#include <brpc/extension.h>
#include <butil/strings/string_piece.h>
#include "braft/configuration.h"
#include "braft/configuration_manager.h"namespace google {
namespace protobuf {
class Message;
} // namespace protobuf
} // namespace googlenamespace braft {
// 使用的是谷歌的gflags，大致就是可以使用这三个参数做调参
DECLARE_bool(raft_sync);
DECLARE_bool(raft_sync_meta);
DECLARE_bool(raft_create_parent_directories);struct LogEntry;//声明，log条目// 个人感觉像是统计运行时长用的
struct IOMetric {
public:IOMetric() : start_time_us(butil::cpuwide_time_us()), bthread_queue_time_us(0), open_segment_time_us(0), append_entry_time_us(0), sync_segment_time_us(0) {}int64_t start_time_us;int64_t bthread_queue_time_us;int64_t open_segment_time_us;int64_t append_entry_time_us;int64_t sync_segment_time_us;
};// 重载输出符号
inline std::ostream& operator<<(std::ostream& os, const IOMetric& m) {return os << " bthread_queue_time_us: " << m.bthread_queue_time_us<< " open_segment_time_us: " << m.open_segment_time_us << " append_entry_time_us: " << m.append_entry_time_us<< " sync_segment_time_us: " << m.sync_segment_time_us;
}
//解析uri
inline butil::StringPiece parse_uri(butil::StringPiece* uri, std::string* parameter) {// ${protocol}://${parameters}size_t pos = uri->find("://");if (pos == butil::StringPiece::npos) {return butil::StringPiece();}butil::StringPiece protocol = uri->substr(0, pos);uri->remove_prefix(pos + 3/* length of '://' */);protocol.trim_spaces();parameter->reserve(uri->size());parameter->clear();size_t removed_spaces = 0;for (butil::StringPiece::const_iterator iter = uri->begin(); iter != uri->end(); ++iter) {if (!isspace(*iter)) {parameter->push_back(*iter);} else {++removed_spaces;}}LOG_IF(WARNING, removed_spaces) << "Removed " << removed_spaces << " spaces from `" << *uri << '\'';return protocol;
}
// 回收资源
inline int gc_dir(const std::string& path) {butil::File::Error e;butil::FilePath target_path(path);butil::FilePath tmp_path(path + ".tmp");// delete tmp path firstly in case there is garbageif (!butil::DeleteFile(tmp_path, true)) {LOG(ERROR) << "Fail to delete tmp file, path: " << tmp_path.value();return -1;}if (butil::PathExists(target_path)) {const bool rc = butil::ReplaceFile(butil::FilePath(target_path),butil::FilePath(tmp_path), &e);if (!rc) {LOG(ERROR) << "Fail to rename `" << target_path.value()<< " to `" << tmp_path.value() << "' : " << e;return -1;}if (!butil::DeleteFile(tmp_path, true)) {LOG(ERROR) << "Fail to delete tmp file, path: " << tmp_path.value();return -1;}} else {LOG(INFO) << "Target path not exist, so no need to gc, path: " << target_path.value();}return 0;
}// log存储结构
class LogStorage {
public:virtual ~LogStorage() {}// 初始化，检查一致性和完整性,纯虚函数，由子类实现virtual int init(ConfigurationManager* configuration_manager) = 0;// 日志中的第一个日志索引virtual int64_t first_log_index() = 0;// 日志中的最后一个日志索引virtual int64_t last_log_index() = 0;// 通过索引号获取日志条目virtual LogEntry* get_entry(const int64_t index) = 0;// 通过索引号获取日志条目的所属termvirtual int64_t get_term(const int64_t index) = 0;// 将条目追加到日志virtual int append_entry(const LogEntry* entry) = 0;// 追加条目以记录和更新IOMetric，返回追加成功编号virtual int append_entries(const std::vector<LogEntry*>& entries, IOMetric* metric) = 0;// 从存储的头部删除日志，[first_log_index，first_index_kept）将被丢弃virtual int truncate_prefix(const int64_t first_index_kept) = 0;// 从存储的尾部删除未提交的日志，（last_index_kept，last_log_index]将被丢弃virtual int truncate_suffix(const int64_t last_index_kept) = 0;// 删除所有现有日志，然后将下一个日志索引重置为| next_log_index |。// 在安装从leader节点传来的快照后调用此函数virtual int reset(const int64_t next_log_index) = 0;// 使用| uri |中编码的参数创建此类LogStorage的实例。// 成功则返回引用到实例的指针，否则返回NULL。virtual LogStorage* new_instance(const std::string& uri) const = 0;static LogStorage* create(const std::string& uri);// 回收使用| uri |参数的实例的资源virtual butil::Status gc_instance(const std::string& uri) const {CHECK(false) << butil::class_name_str(*this)<< " didn't implement gc_instance interface while deleting"" raft log in " << uri;butil::Status status;status.set_error(ENOSYS, "gc_instance interface is not implemented");return status;}static butil::Status destroy(const std::string& uri);
};// 第二大块存储结构，用来存放一些RAFT算法自身的状态数据， 比如term, vote_for等信息.
class RaftMetaStorage {
public:virtual ~RaftMetaStorage() {}//构造函数// 初始化virtual butil::Status init() = 0;// 设置term 和votedfor信息virtual butil::Status set_term_and_votedfor(const int64_t term, const PeerId& peer_id, const VersionedGroupId& group) = 0;// 得到 term 和votedfor信息virtual butil::Status get_term_and_votedfor(int64_t* term, PeerId* peer_id, const VersionedGroupId& group) = 0;// 使用| uri |中编码的参数创建此类RaftMetaStorage的实例。// 成功则返回引用到实例的指针，否则返回NULL。virtual RaftMetaStorage* new_instance(const std::string& uri) const = 0;static RaftMetaStorage* create(const std::string& uri);// 回收使用| uri |参数的实例的资源virtual butil::Status gc_instance(const std::string& uri, const VersionedGroupId& vgid) const {CHECK(false) << butil::class_name_str(*this)<< " didn't implement gc_instance interface while deleting"" raft stable meta in " << uri;butil::Status status;status.set_error(ENOSYS, "gc_instance interface is not implemented");return status;}static butil::Status destroy(const std::string& uri, const VersionedGroupId& vgid);};// 快照
class Snapshot : public butil::Status {
public:Snapshot() {}//virtual ~Snapshot() {}// 获取快照的路径virtual std::string get_path() = 0;// 列出当前快照中的所有现有文件virtual void list_files(std::vector<std::string> *files) = 0;// 获取实现定义的file_meta（implementation-defined file_meta）virtual int get_file_meta(const std::string& filename, ::google::protobuf::Message* file_meta) {(void)filename;file_meta->Clear();return 0;}
};// 快照写
class SnapshotWriter : public Snapshot {
public:SnapshotWriter() {}virtual ~SnapshotWriter() {}// 保存被raft框架使用的快照的元信息。virtual int save_meta(const SnapshotMeta& meta) = 0;// 将文件添加到快照。// |file_meta| 是一个实现定义的原型消息 （implmentation-defined protobuf message， 不会翻译）// 所有实现都必须处理|file_meta| 为null的情况，确保不会引发错误// 请注意，是否将文件创建到后备存储器是由实现定义（implmentation-defined）的。virtual int add_file(const std::string& filename) { return add_file(filename, NULL);}virtual int add_file(const std::string& filename, const ::google::protobuf::Message* file_meta) = 0;// 从快照中删除文件// 请注意，是否从后备存储器中删除文件是由实现定义（implmentation-defined）的。virtual int remove_file(const std::string& filename) = 0;
};// 快照读
class SnapshotReader : public Snapshot {
public:SnapshotReader() {}virtual ~SnapshotReader() {}// Load meta from virtual int load_meta(SnapshotMeta* meta) = 0;// 为其他peers生成uri以复制此快照。// 如果发生某些错误，则返回一个空字符串virtual std::string generate_uri_for_copy() = 0;
};// 从给定资源复制快照
class SnapshotCopier : public butil::Status {
public:virtual ~SnapshotCopier() {}// 取消复制作业virtual void cancel() = 0;// 阻塞线程，直到此复制作业完成或发生某些错误。virtual void join() = 0;// 获取代表复制的快照的SnapshotReadervirtual SnapshotReader* get_reader() = 0;
};// 类声明
class SnapshotHook;
class FileSystemAdaptor;
class SnapshotThrottle;// 快照存储
class SnapshotStorage {
public:virtual ~SnapshotStorage() {}// 以下三种都有参数设定virtual int set_filter_before_copy_remote() {CHECK(false) << butil::class_name_str(*this) << " doesn't support filter before copy remote";return -1;}virtual int set_file_system_adaptor(FileSystemAdaptor* fs) {(void)fs;CHECK(false) << butil::class_name_str(*this) << " doesn't support file system adaptor";return -1;}virtual int set_snapshot_throttle(SnapshotThrottle* st) {(void)st;CHECK(false) << butil::class_name_str(*this) << " doesn't support snapshot throttle";return -1;}// 初始化virtual int init() = 0;// 创建一个新的 快照编写器virtual SnapshotWriter* create() = 0;// 关闭快照编写器virtual int close(SnapshotWriter* writer) = 0;// 获取最新的快照阅读器virtual SnapshotReader* open() = 0;// 关闭快照阅读器virtual int close(SnapshotReader* reader) = 0;// 从uri复制快照并作为SnapshotReader打开virtual SnapshotReader* copy_from(const std::string& uri) WARN_UNUSED_RESULT = 0;virtual SnapshotCopier* start_to_copy_from(const std::string& uri) = 0;virtual int close(SnapshotCopier* copier) = 0;// 使用| uri |中编码的参数创建此类SnapshotStorage的实例。// 成功则返回引用到实例的指针，否则返回NULL。virtual SnapshotStorage* new_instance(const std::string& uri) const WARN_UNUSED_RESULT = 0;static SnapshotStorage* create(const std::string& uri);// 回收使用| uri |参数的 SnapshotStorage实例的资源virtual butil::Status gc_instance(const std::string& uri) const {CHECK(false) << butil::class_name_str(*this)<< " didn't implement gc_instance interface while deleting"" raft snapshot in " << uri;butil::Status status;status.set_error(ENOSYS, "gc_instance interface is not implemented");return status;}static butil::Status destroy(const std::string& uri);
};inline brpc::Extension<const LogStorage>* log_storage_extension() {return brpc::Extension<const LogStorage>::instance();
}inline brpc::Extension<const RaftMetaStorage>* meta_storage_extension() {return brpc::Extension<const RaftMetaStorage>::instance();
}inline brpc::Extension<const SnapshotStorage>* snapshot_storage_extension() {return brpc::Extension<const SnapshotStorage>::instance();
}}#endif //~BRAFT_RAFT_STORAGE_H
cli.h
#ifndef BRAFT_CLI_H
#define BRAFT_CLI_H#include "braft/raft.h"namespace braft {
namespace cli {struct CliOptions {int timeout_ms;int max_retry;CliOptions() : timeout_ms(-1), max_retry(3) {}
};// 将一个新的peer节点添加到由| conf |组成的复制组中。
// 成功返回OK，否则返回错误信息。
butil::Status add_peer(const GroupId& group_id, const Configuration& conf,const PeerId& peer_id, const CliOptions& options);// 将一个peer节点从由| conf |组成的复制组中移除。
// 成功返回OK，否则返回错误信息。
butil::Status remove_peer(const GroupId& group_id, const Configuration& conf,const PeerId& peer_id, const CliOptions& options);// 优雅地更改复制组的peers节点
butil::Status change_peers(const GroupId& group_id, const Configuration& conf, const Configuration& new_peers,const CliOptions& options);// 将该复制组的leader转移到目标peer节点
butil::Status transfer_leader(const GroupId& group_id, const Configuration& conf,const PeerId& peer, const CliOptions& options);// 将目标peer节点的peer set重置
butil::Status reset_peer(const GroupId& group_id, const PeerId& peer_id,const Configuration& new_conf,const CliOptions& options);// 要求peer立即转储快照
butil::Status snapshot(const GroupId& group_id, const PeerId& peer_id,const CliOptions& options);} // namespace cli
} // namespace braft#endif //BRAFT_CLI_H
route_table.h
#ifndef BRAFT_ROUTE_TABLE_H
#define BRAFT_ROUTE_TABLE_H#include "braft/configuration.h" // Configuration
#include "braft/raft.h"// Maintain routes to raft groupsnamespace braft {
namespace rtb {// 更新路由表中组的配置
int update_configuration(const GroupId& group, const Configuration& conf);
int update_configuration(const GroupId& group, const std::string& conf_str);// 得到该组leader
// 返回:
// 0 : 成功
// 1 : 不确定leader是谁
// -1: 其他情况
int select_leader(const GroupId& group, PeerId* leader);// 更新leader
int update_leader(const GroupId& group, const PeerId& leader);
int update_leader(const GroupId& group, const std::string& leader_str);// 阻塞线程，直到query_leader完成
butil::Status refresh_leader(const GroupId& group, int timeout_ms);// 从路由表中删除该组
int remove_group(const GroupId& group);} // namespace rtb
} // namespace braft #endif //BRAFT_ROUTE_TABLE_H