优化 1：Batch
Batch 队列
从客户端调用 apply 接口提交 Task 到日志成功复制，并回调用户状态机的 on_apply 接口，日志依次经过了 Apply Queue、Disk Queue、Apply Task Queue 这 3 个队列。这些队列都是 brpc 的 ExecutionQueue 实现，其消费函数都做了 batch 优化：
ApplyQueue : 用户提交的 Task 会进入该队列，在该队列的消费函数中会将这些 Task 对应的日志进行打包，并调用 LogManager 的 append_entries 函数进行追加日志，打包的日志既用于持久化，也用于复制。默认一次打包 32 条日志。
DiskQueue : LogManager 在接收到这批日志后，需对其进行持久化处理，故会往该队列中提交一个持久化任务（一个任务对应一批日志）；在该队列的消费函数中会将这些任务进行打包，将这批任务对应的所有日志写入磁盘。默认一次最多打包 256 个持久化任务，而每个任务最多包含 32 条日志，所以其一次 Bacth Write 最多会写入 256 * 32 = 8192 条日志对应的数据。当然其也受字节数限制，默认每次 Batch Write 最多写入 256KB。
ApplyTaskQueue ：当日志的复制数（包含持久化）达到 Quorum 后，会调用 on_committed 往 ApplyTaskQueue 中提交一个 ApplyTask（每个 ApplyTask 对应一批已提交的日志）；在该队列的消费函数中会将这些 ApplyTask 打包成 Iterator，并作为参数回调用户状态机的 on_apply 函数。 默认一次最多打包 512 个 ApplyTask，而每个 ApplyTask 最多包含 32 条日志，所以每一次 on_apply 参数中的 Iterator 最多包含 512 * 32 = 16384 条日志。
从以上看出，日志的复制、持久化、应用，全链路都是经过 Batch 优化的。
Follower 的 Batch
以上讨论的是节点作为 Leader 时的 Batch 优化，当节点为 Follower 时，其优化也是一样的，因为其用的是相同的代码逻辑，唯一的区别在于：
ApplyQueue : Follower 不会接受用户提交的日志（Task），其批量日志来源于 Leader 的复制
DiskQueue : 日志批量落盘的逻辑是一样的，Follower 在接收到 Leader 的一批日志之后也是直接调用 LogManager 的 append_entries 函数
ApplyTaskQueue : 批量 on_apply 的逻辑是一样的，区别在于 Follower 的 commitIndex 来源于 Leader 在 RPC 中携带的 commitIndex，并非通过自身的 Quorum 计算
相关配置
当然，框架也提供了一些配置项来调整这些 Batch 大小：
该打包大小影响复制、持久化以及on_apply；上限为 512
raft_max_append_buffer_size
raft_fsm_caller_commit_batch
每次 on_apply 的最大日志数为 raft_apply_batch * raft_fsm_caller_commit_batch
另外，配置 raft_max_entries_size 也会影响每次复制（AppendEntries 请求）携带的日志数量，其默认值为 1024
具体实现
ApplyQueue 消费函数：
Copy int NodeImpl::execute_applying_tasks(
void* meta, bthread::TaskIterator<LogEntryAndClosure>& iter) {
...
// (1) FLAGS_raft_apply_batch 默认为 32
// TODO: the batch size should limited by both task size and the total log size
const size_t batch_size = FLAGS_raft_apply_batch;
// (2) 定义一个 tasks 数组，数组大小 = min(batch_size, 256)
DEFINE_SMALL_ARRAY(LogEntryAndClosure, tasks, batch_size, 256);
size_t cur_size = 0;
NodeImpl* m = (NodeImpl*)meta;
for (; iter; ++iter) {
if (cur_size == batch_size) {
m->apply(tasks, cur_size); // (4) 调用批量 apply 函数
cur_size = 0;
}
tasks[cur_size++] = *iter; // (3) batch 打包
}
...
return 0;
} DiskQueue 消费函数：
Copy int LogManager::disk_thread(void* meta,
bthread::TaskIterator<StableClosure*>& iter) {
...
// (1) 定义 AppendBatcher 的 Batch 大小为 256
StableClosure* storage[256];
AppendBatcher ab(storage, ARRAY_SIZE(storage), &last_id, log_manager);
for (; iter; ++iter) {
// ^^^ Must iterate to the end to release to corresponding
// even if some error has occurred
StableClosure* done = *iter;
...
if (!done->_entries.empty()) {
ab.append(done); // (2) batch 打包持久化任务
} else {
ab.flush(); // (3) 批量持久化
...
}
}
...
return 0;
}
class AppendBatcher {
public:
...
void flush() {
if (_size > 0) {
// 写入持久化存储
_lm->append_to_storage(&_to_append, _last_id, &metric);
...
}
...
}
void append(LogManager::StableClosure* done) {
// FLAGS_raft_max_append_buffer_size 默认为 256KB
// 一次 Batch Write 写入的字节数不能超过该配置值
if (_size == _cap ||
_buffer_size >= (size_t)FLAGS_raft_max_append_buffer_size) {
flush();
}
_storage[_size++] = done;
_to_append.insert(_to_append.end(),
done->_entries.begin(), done->_entries.end());
for (size_t i = 0; i < done->_entries.size(); ++i) {
_buffer_size += done->_entries[i]->data.length();
}
}
} ApplyTaskQueue 消费函数：
Copy int FSMCaller::run(void* meta, bthread::TaskIterator<ApplyTask>& iter) {
FSMCaller* caller = (FSMCaller*)meta;
...
int64_t max_committed_index = -1;
int64_t counter = 0;
// FLAGS_raft_fsm_caller_commit_batch 默认为 512
size_t batch_size = FLAGS_raft_fsm_caller_commit_batch;
for (; iter; ++iter) {
if (iter->type == COMMITTED && counter < batch_size) {
if (iter->committed_index > max_committed_index) {
max_committed_index = iter->committed_index;
counter++;
}
} else {
if (max_committed_index >= 0) {
caller->_cur_task = COMMITTED;
caller->do_committed(max_committed_index);
max_committed_index = -1;
counter = 0;
batch_size = FLAGS_raft_fsm_caller_commit_batch;
}
...
}
}
...
return 0;
}
// 生成 Iterator，调用用户状态机的 `on_apply`
void FSMCaller::do_committed(int64_t committed_index) {
...
IteratorImpl iter_impl(_fsm, _log_manager, &closure, first_closure_index,
last_applied_index, committed_index, &_applying_index);
for (; iter_impl.is_good();) {
...
Iterator iter(&iter_impl);
_fsm->on_apply(iter);
...
// Try move to next in case that we pass the same log twice.
iter.next();
}
...
} 优化 2：并行持久化日志
持久化与复制
在 Raft 的实现中，Leader 需要先在本地持久化日志，再向所有的 Follower 复制日志，显然这样的实现具有较高的时延。特别地客户端的写都要经过 Leader，导致 Leader 的压力会变大，从而导致 IO 延迟变高成为慢节点，而本地持久化也会阻塞后续的 Follower 复制。
所以在 braft 中，Leader 本地持久化和 Follower 复制是并行的，即 Leader 会先将日志写入内存，同时异步地进行持久化和向 Follower 复制。并且只要大多数节点写入成功就可以应用日志，不必等待 Leader 本地持久化成功。
具体实现
具体的实现我们已经在上一节中详细介绍过了，参见 <4.1 复制流程> 。
优化 3：流水线复制
pipeline
Raft 默认是串行复制日志，需要等待一个 AppendEntries 发送成功后再发送下一个，显然这样不是最高效的。可以采用 Pipeline 的复制方式，即 Leader 发送完 AppendEntries 请求后不必等待其响应，立马发送一下批日志。当然，这样的实现对于接受端（Follower）来说，可能会带来乱序、空洞等问题，为此，braft 在 Follower 端引入了日志缓存，将不是顺序的日志先缓存起来，待其前面的日志都接受到后再写入该日志，以达到日志连续的目的。
特别需要注意的是，pipeline 优化默认是关闭的，需要用户通过以下 2 个配置项开启：
Copy DEFINE_bool(raft_enable_append_entries_cache, false,
"enable cache for out-of-order append entries requests, should used when "
"pipeline replication is enabled (raft_max_parallel_append_entries_rpc_num > 1).");
DEFINE_int32(raft_max_parallel_append_entries_rpc_num, 1,
"The max number of parallel AppendEntries requests"); 具体实现
Leader 端 pipeline 的流程如下：
(1) 发送 AppendEntries 请求时会判断并行的请求数（flying）是否达到并行上限
(1.1) 如果是的话，直接返回；等待收到响应后回调 _on_rpc_returned 将 flying 计数减一
(1.2) 否则继续发送 AppendEntries 请求，并将 flying 计数加一
(2) 发送完后，继续判断 flying 数是否已达限，是的话如同步骤（1.1）
(3) 否则，判断是否还有日志可以发送：
(3.2) 否则注册 waiter 等待新日志到来
具体流程见下面代码解析:
Copy void Replicator::_send_entries() {
// (1) 如果已经发送的 AppendEntries 请求数大于等于 `raft_max_parallel_append_entries_rpc_num`，
// 则返回等待 `_on_rpc_returned`
if (_flying_append_entries_size >= FLAGS_raft_max_entries_size ||
_append_entries_in_fly.size() >= (size_t)FLAGS_raft_max_parallel_append_entries_rpc_num ||
_st.st == BLOCKING) {
...
return;
}
...
// (2) 否则发送 AppendEntries 请求
google::protobuf::Closure* done = brpc::NewCallback(
_on_rpc_returned, _id.value, cntl.get(),
request.get(), response.get(), butil::monotonic_time_ms());
RaftService_Stub stub(&_sending_channel);
stub.append_entries(cntl.release(), request.release(),
response.release(), done);
// (3) 发送完后调用 `_wait_more_entries`
_wait_more_entries();
}
// (4) `_wait_more_entries` 首先判断已经并行的 `AppendEntries` 请求数是否已达上限
// (4.1) 如果已达上限，则直接返回，等待收到响应后回调 `_on_rpc_returned` 即可
// (4.2) 否则调用 LogManager::wait 判断是否还有日志没同步
// 如果还有日志，则继续发送；否则创建 waiter 在后台等待新日志到来
void Replicator::_wait_more_entries() {
if (_wait_id == 0 && FLAGS_raft_max_entries_size > _flying_append_entries_size &&
(size_t)FLAGS_raft_max_parallel_append_entries_rpc_num > _append_entries_in_fly.size()) {
_wait_id = _options.log_manager->wait(
_next_index - 1, _continue_sending, (void*)_id.value);
...
}
...
}
// (5) LogManager::wait 会调用 LogManager::notify_on_new_log
LogManager::WaitId LogManager::wait(
int64_t expected_last_log_index,
int (*on_new_log)(void *arg, int error_code), void *arg) {
...
return notify_on_new_log(expected_last_log_index, wm);
}
// (6) 判断是否还有日志没同步：
LogManager::WaitId LogManager::notify_on_new_log(
int64_t expected_last_log_index, WaitMeta* wm) {
// (6.1) 如果有的话，调用 run_on_new_log 继续发送
if (expected_last_log_index != _last_log_index || _stopped) {
...
if (bthread_start_urgent(&tid, NULL, run_on_new_log, wm) != 0) {
...
}
return 0; // Not pushed into _wait_map
}
// (6.2) 没有可发送的日志，创建 waiter 在后台等待新日志
if (_next_wait_id == 0) { // skip 0
++_next_wait_id;
}
const int wait_id = _next_wait_id++;
_wait_map[wait_id] = wm;
return wait_id;
}
// (7) run_on_new_log 会调用 `_continue_sending`
void* LogManager::run_on_new_log(void *arg) {
WaitMeta* wm = (WaitMeta*)arg;
wm->on_new_log(wm->arg, wm->error_code); // on_new_log: _continue_sending
...
return NULL;
}
// (8) 而 _continue_sending 则调用 `_send_entries` 继续发送日志
int Replicator::_continue_sending(void* arg, int error_code) {
...
r->_send_entries();
...
} Follower 接收到 AppendEntries 请求后，其处理逻辑如下：
对于到来的日志，如果其前面的日志没有接受到，先将其缓存起来，并要求 Leader 重发前面的日志
待前面的日志都达到后，再将缓存中在其之后的顺序的日志一起写入磁盘
Copy void NodeImpl::handle_append_entries_request(brpc::Controller* cntl,
const AppendEntriesRequest* request,
AppendEntriesResponse* response,
google::protobuf::Closure* done,
bool from_append_entries_cache) {
// (1) 当到来的日志的前面的日志没有接受到时，先将其缓存起来
if (local_prev_log_term != prev_log_term) {
int64_t last_index = _log_manager->last_log_index();
int64_t saved_term = request->term();
int saved_entries_size = request->entries_size();
std::string rpc_server_id = request->server_id();
// (1.1) 调用 handle_out_of_order_append_entries 将日志缓存起来
if (!from_append_entries_cache &&
handle_out_of_order_append_entries(
cntl, request, response, done, last_index)) {
// It's not safe to touch cntl/request/response/done after this point,
// since the ownership is tranfered to the cache.
...
return;
}
// (1.2) 并将响应设为 False, 要求 Leader 重发前面的日志
response->set_success(false);
response->set_term(_current_term);
response->set_last_log_index(last_index); // follower 的 lastLogIndex
...
return;
}
// (2) 当前的日志前面的日志都已达到（即日志是顺序的）
// 这时候查看缓存中是否还有连续的日志，如果有的话，
// 将以缓存的日志再次调用 handle_append_entries_request，
// 并将这些日志一并写入磁盘
// check out-of-order cache
check_append_entries_cache(index);
// (3) 将日志持久化
FollowerStableClosure* c = new FollowerStableClosure(
cntl, request, response, done_guard.release(),
this, _current_term);
_log_manager->append_entries(&entries, c);
} 优化 4：raft sync
sync 配置
日志每次 Batch Write 是先将日志写入 Page Cache，最后再调用一次 fsync 操作。显然 sync 操作会增加时延，为此 braft 提供了一些配置项来控制日志的 sync 行为。用户可以根据业务数据丢失的容忍度高低，灵活调整这些配置以达到性能和可靠性之间的权衡。例如对于数据丢失容忍度较高的业务，可以选择将配置项 raft_sync 设置为 Flase，这将有助于提升性能。
以下是控制日志 sync 的相关参数，前三项针对的是每次的 Batch Write，最后两项针对的是 Segment：
每次 Batch Write 后是否需要 sync
对于每次 Batch Write 后的 sync 策略，有立即 sync （RAFT_SYNC_IMMEDIATELY）和按字节数 sync (RAFT_SYNC_BY_BYTES)
在 RAFT_SYNC_BY_BYTES 的策略下，每多少字节 sync 一次
每当日志写满一个 Segment 需要切换时是否需要 sync，每个 Segment 默认存储 8MB 的日志
具体实现
Batch Write 写入 Page Cache 后会调用 Segment::sync 进行 sync 操作：
Copy int Segment::sync(bool will_sync, bool has_conf) {
...
if (will_sync) {
// (1) 如果 `raft_sync` 为 False，则直接返回
if (!FLAGS_raft_sync) {
return 0;
}
// (2) 如果是按字节 `sync` 并且当前未 `sync` 的字节小于配置值，则直接返回
if (FLAGS_raft_sync_policy == RaftSyncPolicy::RAFT_SYNC_BY_BYTES
&& FLAGS_raft_sync_per_bytes > _unsynced_bytes
&& !has_conf) {
return 0;
}
_unsynced_bytes = 0;
// (3) 调用 fsync(fd)
return raft_fsync(_fd);
}
return 0;
} 每当一个 Segment 文件写完后，都会调用 Segment::close 将其转换成 closed segment：
Copy int Segment::close(bool will_sync) {
...
if (FLAGS_raft_sync_segments && will_sync) {
ret = raft_fsync(_fd);
}
...
} 优化 5：异步 Apply
Copy void on_apply(braft::Iterator& iter) {
// A batch of tasks are committed, which must be processed through
// |iter|
for (; iter.valid(); iter.next()) {
// This guard helps invoke iter.done()->Run() asynchronously to
// avoid that callback blocks the StateMachine.
braft::AsyncClosureGuard closure_guard(iter.done());
// Parse operation from iter.data() and execute this operation
// op = parse(iter.data());
// result = process(op)
// The purpose of following logs is to help you understand the way
// this StateMachine works.
// Remove these logs in performance-sensitive servers.
LOG_IF(INFO, FLAGS_log_applied_task)
<< "Exeucted operation " << op
<< " and the result is " << result
<< " at log_index=" << iter.index();
}
} 当日志被提交时，框架会串行调用用户状态机的 on_apply，虽然这里做了 Batch 优化，但是对于那些不支持批量更新的状态机来说，仍然是低效的。为此，用户可以将 on_apply 函数异步执行，让那些可以并行的操作尽可能并行起来。
需要注意的是，当 on_apply 异步后，需要处理好节点成为 Leader 时日志回放的问题。当节点刚成为 Leader 时，需要回放（on_apply）之前任期的日志，这时候需要将这些日志全部应用到状态机后才能处理读取操作，不然可能会违背线性一致性。因为这些日志在之前的任期可能被提交了，客户端能读取到，而在新任期回放的时候，由于这些日志是异步执行的（on_apply 返回了，但还在异步执行中），可能还没应用到状态机，这时候客户端去读取可能是读取不到的。
参考
