在发送空 AppendEntries 中的最后一步，如果找到了 Follower 的对应的 next_index，就调用 Replicator::_send_entries() 开始日志复制。该函数实际上是从本地取 log 发送给 Follower，我们先不着急看 Replicator::_send_entries() 的实现，先从用户提交任务开始分析。
当客户端请求过来的时候，服务端需要将 request 转化为 log entry (IOBuf)，并构造一个 braft::Task，将 Task 的 data 设置为 log，并将回调函数 done 构造Closure 传给 Task 的 done，当函数最终成功执行或者失败的时候会执行回调。下面是 example 里面的 counter 的 fetch_add 接口实现：
void fetch_add(const FetchAddRequest* request,
CounterResponse* response,
google::protobuf::Closure* done) {
brpc::ClosureGuard done_guard(done);
// Serialize request to IOBuf
const int64_t term = _leader_term.load(butil::memory_order_relaxed);
if (term < 0) {
return redirect(response);
}
butil::IOBuf log;
butil::IOBufAsZeroCopyOutputStream wrapper(&log);
if (!request->SerializeToZeroCopyStream(&wrapper)) {
LOG(ERROR) << "Fail to serialize request";
response->set_success(false);
return;
}
// Apply this log as a braft::Task
braft::Task task;
task.data = &log;
// This callback would be iovoked when the task actually excuted or
// fail
task.done = new FetchAddClosure(this, request, response,
done_guard.release());
if (FLAGS_check_term) {
// ABA problem can be avoid if expected_term is set
task.expected_term = term;
}
// Now the task is applied to the group, waiting for the result.
return _node->apply(task);
}void NodeImpl::apply(const Task& task) {
LogEntry* entry = new LogEntry;
entry->AddRef();
entry->data.swap(*task.data);
LogEntryAndClosure m;
m.entry = entry;
m.done = task.done;
m.expected_term = task.expected_term;
if (_apply_queue->execute(m, &bthread::TASK_OPTIONS_INPLACE, NULL) != 0) {
task.done->status().set_error(EPERM, "Node is down");
entry->Release();
return run_closure_in_bthread(task.done);
}
}它会把 Task 和回调一起放到 _apply_queue 去执行。执行的时候调用 NodeImpl::execute_applying_tasks()，该函数会把 task 遍历一遍放到 task 数组，然后调用NodeImpl::apply(LogEntryAndClosure tasks[], size_t size)。
NodeImpl::execute_applying_tasks() 会遍历 Task 列表，==以 raft_apply_batch 大小为单位 batch 执行任务==。
int NodeImpl::execute_applying_tasks(
void* meta, bthread::TaskIterator<LogEntryAndClosure>& iter) {
if (iter.is_queue_stopped()) {
return 0;
}
// TODO: the batch size should limited by both task size and the total log
// size
const size_t batch_size = FLAGS_raft_apply_batch;
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);
cur_size = 0;
}
tasks[cur_size++] = *iter;
}
if (cur_size > 0) {
m->apply(tasks, cur_size);
}
return 0;
}NodeImpl::apply(LogEntryAndClosure tasks[], size_t size) 函数会检查当前的状态是否为 Leader，以及 task 的 expected_term 是否等于当前 term等。一旦出错就会调用 task 的 done 返回给用户。没有错误的情况下会遍历所有 task，然后把 task 里面的 entry 放到 entries 数组里面，设置 entry 的 term，并将 task 放到 ballot 的 pending_task 用于投票。然后调用 _log_manager->append_entries() 试图 append entries 的本地日志，然后更新当前配置。
void NodeImpl::apply(LogEntryAndClosure tasks[], size_t size) {
g_apply_tasks_batch_counter << size;
std::vector<LogEntry*> entries;
entries.reserve(size);
std::unique_lock<raft_mutex_t> lck(_mutex);
// 各种 check
// ... ...
for (size_t i = 0; i < size; ++i) {
// term check
// .. ...
entries.push_back(tasks[i].entry);
entries.back()->id.term = _current_term;
entries.back()->type = ENTRY_TYPE_DATA;
_ballot_box->append_pending_task(_conf.conf,
_conf.stable() ? NULL : &_conf.old_conf,
tasks[i].done);
}
_log_manager->append_entries(&entries,
new LeaderStableClosure(
NodeId(_group_id, _server_id),
entries.size(),
_ballot_box));
// update _conf.first
_log_manager->check_and_set_configuration(&_conf);
}关于 BallotBox：
BallotBox 用于决定一个 task 是否被复制到多数派，
BallotBox::append_pending_task()用于提交一个任务，注意该函数有 conf 和 old_conf 两个 Configuration 类型的参数，用于在成员变更过程中实现在新旧两个配置都达成一致。ConfigurationEntry 中会保存新旧两个配置，当没有旧配置的时候
ConfigurationEntry::stable()才会返回 true，在成员变更的过程中会修改_conf。struct ConfigurationEntry { LogId id; Configuration conf; Configuration old_conf; bool stable() const { return old_conf.empty(); } // ... ... }
LogManager::append_entries 会给 entries 分配 index，把它们存到 LogManager::_logs_in_memory 缓存里面，然后把任务提交到 LogManager::_disk_queue 里面，最后 LogManager::disk_thread() 会把 entries 持久化到 storage，持久化的时候加了个 batcher，分批写入文件。
Tips:
日志写到缓存里后就可以被 Replicator 读取发送到 Follower 了，不需要等到本地持久化
日志 index 分配
int LogManager::check_and_resolve_conflict( std::vector<LogEntry*> *entries, StableClosure* done) { AsyncClosureGuard done_guard(done); if (entries->front()->id.index == 0) { // Node is currently the leader and |entries| are from the user who // don't know the correct indexes the logs should assign to. So we have // to assign indexes to the appending entries for (size_t i = 0; i < entries->size(); ++i) { (*entries)[i]->id.index = ++_last_log_index; } done_guard.release(); return 0; } else { // ... ... } }
braft 默认使用 SegmentLogStorage 持久化，它会把entry 按照一定格式写入固定大小的 segment 里面。成功后调用 LeaderStableClosure::Run()，该函数会为这些 entries 投出 Leader 本身的一票。
也就是说，提交的操作就是把请求变成 entry 按照顺序存到内存并持久化。有专门的线程会 wait entry 到来并发起 append entries，后面会讲到。
NodeImpl::Apply() 只是把日志写到本地，日志发送给 Follower 是一个异步的过程，在 Replicator::_on_rpc_returned() 找到正确的 next_index 后就调用 Replicator::_send_entries() 开始发送日志。另外，因为日志复制的 AppendEntries RPC 回复也是调用 Replicator::_on_rpc_returned() 处理的，因此每一个日志复制 RPC 收到回复后也会调用 Replicator::_send_entries() 发送日志。
-
如果有太多的已发送未收到回复的 entries，本次先不发送（flying entries 相关的参数可以配置）
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) { BRAFT_VLOG << "node " << _options.group_id << ":" << _options.server_id << " skip sending AppendEntriesRequest to " << _options.peer_id << ", too many requests in flying, or the replicator is in block," << " next_index " << _next_index << " flying_size " << _flying_append_entries_size; CHECK_EQ(0, bthread_id_unlock(_id)) << "Fail to unlock " << _id; return; }
-
调用
Replicator::_fill_common_fields()填充 request，如果填充失败，意味着本地没有 next_index 的日志，需要 Follower 安装快照。if (_fill_common_fields(request.get(), _next_index - 1, false) != 0) { _reset_next_index(); return _install_snapshot(); }
-
然后计算
max_entries_size，并调用Replicator::_prepare_entry()获取 entry 并添加到 request 中。Replicator::_prepare_entry()会从logManager 获取 entry，首先从缓存获取，失败则从文件获取。EntryMeta em; const int max_entries_size = FLAGS_raft_max_entries_size - _flying_append_entries_size; int prepare_entry_rc = 0; CHECK_GT(max_entries_size, 0); for (int i = 0; i < max_entries_size; ++i) { prepare_entry_rc = _prepare_entry(i, &em, &cntl->request_attachment()); if (prepare_entry_rc != 0) { break; } request->add_entries()->Swap(&em); } -
如果
request->entries_size()为 0，并且 Leader 是只读模式的话将状态机设置为 idle 并返回。 -
否则说明当前没有日志可以发送了，调用
Replicator::_wait_more_entries()去等待新的任务到来。 如果有新的 log 过来，就会调用Replicator::_continue_sending()继续Replicator::_send_entries()。if (request->entries_size() == 0) { // _id is unlock in _wait_more if (_next_index < _options.log_manager->first_log_index()) { _reset_next_index(); return _install_snapshot(); } // NOTICE: a follower's readonly mode does not prevent install_snapshot // as we need followers to commit conf log(like add_node) when // leader reaches readonly as well if (prepare_entry_rc == EREADONLY) { if (_flying_append_entries_size == 0) { _st.st = IDLE; } CHECK_EQ(0, bthread_id_unlock(_id)) << "Fail to unlock " << _id; return; } return _wait_more_entries();
-
更新一些参数并更新状态机的状态，并发起 AppendEntries RPC，结束后继续等待新的任务（
Replicator::_wait_more_entries()）_append_entries_in_fly.push_back(FlyingAppendEntriesRpc(_next_index, request->entries_size(), cntl->call_id())); _append_entries_counter++; _next_index += request->entries_size(); _flying_append_entries_size += request->entries_size(); g_send_entries_batch_counter << request->entries_size(); _st.st = APPENDING_ENTRIES; _st.first_log_index = _min_flying_index(); _st.last_log_index = _next_index - 1; 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); _wait_more_entries();
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()) {
// expect 的 log index 是 _next_index - 1
_wait_id = _options.log_manager->wait(
_next_index - 1, _continue_sending, (void*)_id.value);
_is_waiter_canceled = false;
BRAFT_VLOG << "node " << _options.group_id << ":" << _options.peer_id
<< " wait more entries, wait_id " << _wait_id;
}
if (_flying_append_entries_size == 0) {
_st.st = IDLE;
}
CHECK_EQ(0, bthread_id_unlock(_id)) << "Fail to unlock " << _id;
}该函数调用 LogManager::wait() 等待有日志写入，其回调函数是 Replicator::_continue_sending()。我们看看 LogManager::wait() 的实现：
LogManager::WaitId LogManager::wait(
int64_t expected_last_log_index,
int (*on_new_log)(void *arg, int error_code), void *arg) {
WaitMeta* wm = butil::get_object<WaitMeta>();
if (BAIDU_UNLIKELY(wm == NULL)) {
PLOG(FATAL) << "Fail to new WaitMeta";
abort();
return -1;
}
wm->on_new_log = on_new_log;
wm->arg = arg;
wm->error_code = 0;
return notify_on_new_log(expected_last_log_index, wm);
}
LogManager::WaitId LogManager::notify_on_new_log(
int64_t expected_last_log_index, WaitMeta* wm) {
std::unique_lock<raft_mutex_t> lck(_mutex);
// LogManager::wait() 的原因是本地日志都读完了, 即 expected_last_log_index == _last_log_index
// 但代码执行到这里的时候可能已经有新的日志写入了 (expected_last_log_index != _last_log_index)
// 这种情况下直接执行回调函数
if (expected_last_log_index != _last_log_index || _stopped) {
wm->error_code = _stopped ? ESTOP : 0;
lck.unlock();
bthread_t tid;
if (bthread_start_urgent(&tid, NULL, run_on_new_log, wm) != 0) {
PLOG(ERROR) << "Fail to start bthread";
run_on_new_log(wm);
}
return 0; // Not pushed into _wait_map
}
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;
}在 LogManager::append_entries() 的最后，会调用 LogManager::wakeup_all_waiter() 唤醒所有等待的读任务。
void LogManager::wakeup_all_waiter(std::unique_lock<raft_mutex_t>& lck) {
if (_wait_map.empty()) {
return;
}
WaitMeta* wm[_wait_map.size()];
size_t nwm = 0;
for (butil::FlatMap<int64_t, WaitMeta*>::const_iterator
iter = _wait_map.begin(); iter != _wait_map.end(); ++iter) {
wm[nwm++] = iter->second;
}
_wait_map.clear();
const int error_code = _stopped ? ESTOP : 0;
lck.unlock();
for (size_t i = 0; i < nwm; ++i) {
wm[i]->error_code = error_code;
bthread_t tid;
bthread_attr_t attr = BTHREAD_ATTR_NORMAL | BTHREAD_NOSIGNAL;
if (bthread_start_background(
&tid, &attr,
run_on_new_log, wm[i]) != 0) {
PLOG(ERROR) << "Fail to start bthread";
run_on_new_log(wm[i]);
}
}
bthread_flush();
}前面的步骤和收到空的 entries 是一样的，最后会构造一个 FollowerStableClosure 传给 LogManager::append_entries() 试图追加 entries。
// check out-of-order cache
check_append_entries_cache(index);
FollowerStableClosure* c = new FollowerStableClosure(
cntl, request, response, done_guard.release(),
this, _current_term);
_log_manager->append_entries(&entries, c);
// update configuration after _log_manager updated its memory status
_log_manager->check_and_set_configuration(&_conf);接下来看看 LogManager::append_entries() 的逻辑：
-
首先，调用
LogManager::check_and_resolve_conflict()检查并解决冲突，如果有冲突，则释放所有 entries 并返回std::unique_lock<raft_mutex_t> lck(_mutex); if (!entries->empty() && check_and_resolve_conflict(entries, done) != 0) { lck.unlock(); // release entries for (size_t i = 0; i < entries->size(); ++i) { (*entries)[i]->Release(); } entries->clear(); return; }
-
对于配置变更类型的 AppendEntries 特殊处理（后面分析成员变更的时候再详细介绍）
for (size_t i = 0; i < entries->size(); ++i) { // Add ref for disk_thread (*entries)[i]->AddRef(); if ((*entries)[i]->type == ENTRY_TYPE_CONFIGURATION) { ConfigurationEntry conf_entry(*((*entries)[i])); _config_manager->add(conf_entry); } }
-
解决完冲突之后将 entries 插入到缓存中，再将 done 提交到
_disk_queue持久化if (!entries->empty()) { done->_first_log_index = entries->front()->id.index; _logs_in_memory.insert(_logs_in_memory.end(), entries->begin(), entries->end()); } done->_entries.swap(*entries); int ret = bthread::execution_queue_execute(_disk_queue, done); CHECK_EQ(0, ret) << "execq execute failed, ret: " << ret << " err: " << berror(); wakeup_all_waiter(lck);
-
持久化成功后调用
done->Run()，也就是FollowerStableClosure::Run()，该函数最后会检查一下 term 来判断 Leader 有没有变化，如果一切正常，则调用BallotBox::set_last_committed_indexcommit index()更新 commit index// FollowerStableClosure::run() const int64_t committed_index = std::min(_request->committed_index(), // ^^^ committed_index is likely less than the // last_log_index _request->prev_log_index() + _request->entries_size() // ^^^ The logs after the appended entries are // untrustable so we can't commit them even if their // indexes are less than request->committed_index() ); //_ballot_box is thread safe and tolerates disorder. _node->_ballot_box->set_last_committed_index(committed_index);
-
如果更新成功，就调用
FsmCaller::on_committed()，on_committed 将构造一个任务提交到 execution_queue 里面，最后调用FSMCaller::do_committed()去调用用户传入的自定义的StateMachine::on_apply()函数执行状态机的操作。
entries->front()->id.index 为 0 的情况是用于用户提交 Task 后 Leader 为其分配 log index 的场景，在 Follower 上我们忽略该逻辑。
-
如果发过来的第一个 entry 的 index 大于 Follower 的
_last_log_index + 1，说明发送来的日志不连续，返回 -1// Node is currently a follower and |entries| are from the leader. We // should check and resolve the confliction between the local logs and // |entries| if (entries->front()->id.index > _last_log_index + 1) { done->status().set_error(EINVAL, "There's gap between first_index=%" PRId64 " and last_log_index=%" PRId64, entries->front()->id.index, _last_log_index); return -1; }
-
如果发过来的最后一个 entry 的 index 小于等于
applied_index则忽略const int64_t applied_index = _applied_id.index; if (entries->back()->id.index <= applied_index) { LOG(WARNING) << "Received entries of which the last_log=" << entries->back()->id.index << " is not greater than _applied_index=" << applied_index << ", return immediately with nothing changed"; return 1; }
-
如果发过来的第一个 entry 的 index 等于
_last_log_index + 1，则说明没有冲突且日志连续if (entries->front()->id.index == _last_log_index + 1) { // Fast path _last_log_index = entries->back()->id.index;
-
否则说明==日志有冲突==或是==重复日志==，遍历 entries 找到两者冲突的位置 conflicting_index（conflicting_index 之前的日志没有冲突，是重复日志），如果是重复日志最后得到的
conflicting_index将和entries->size()相等} else { // Appending entries overlap the local ones. We should find if there // is a conflicting index from which we should truncate the local // ones. size_t conflicting_index = 0; for (; conflicting_index < entries->size(); ++conflicting_index) { if (unsafe_get_term((*entries)[conflicting_index]->id.index) != (*entries)[conflicting_index]->id.term) { break; } }
-
如果有冲突（
conflicting_index != entries->size()），就把本地有冲突的日志截断，然后把重复的地方是放掉；如果是重复日志就除了释放 entries 外什么都不用做。if (conflicting_index != entries->size()) { if ((*entries)[conflicting_index]->id.index <= _last_log_index) { // Truncate all the conflicting entries to make local logs // consensus with the leader. unsafe_truncate_suffix( (*entries)[conflicting_index]->id.index - 1); } _last_log_index = entries->back()->id.index; } // else this is a duplicated AppendEntriesRequest, we have // nothing to do besides releasing all the entries // Release all the entries before the conflicting_index and the rest // would be append to _logs_in_memory and _log_storage after this // function returns for (size_t i = 0; i < conflicting_index; ++i) { (*entries)[i]->Release(); } entries->erase(entries->begin(), entries->begin() + conflicting_index);
当 RPC 返回后会调用 Replicator::_on_rpc_returned()，前面的部分和空的 AppendEntries 一样，但是 entries_size > 0 的条件成立，它会调用 BallotBox::commit_at() 去投票并决定是否更新 commit index。
const int entries_size = request->entries_size();
const int64_t rpc_last_log_index = request->prev_log_index() + entries_size;
if (entries_size > 0) {
r->_options.ballot_box->commit_at(
min_flying_index, rpc_last_log_index,
r->_options.peer_id);
// ... ...
}下面看一下 BallotBox::commit_at() 的代码：
int BallotBox::commit_at(
int64_t first_log_index, int64_t last_log_index, const PeerId& peer) {
// FIXME(chenzhangyi01): The cricital section is unacceptable because it
// blocks all the other Replicators and LogManagers
std::unique_lock<raft_mutex_t> lck(_mutex);
if (_pending_index == 0) {
return EINVAL;
}
if (last_log_index < _pending_index) {
return 0;
}
if (last_log_index >= _pending_index + (int64_t)_pending_meta_queue.size()) {
return ERANGE;
}
int64_t last_committed_index = 0;
const int64_t start_at = std::max(_pending_index, first_log_index);
Ballot::PosHint pos_hint;
for (int64_t log_index = start_at; log_index <= last_log_index; ++log_index) {
Ballot& bl = _pending_meta_queue[log_index - _pending_index];
pos_hint = bl.grant(peer, pos_hint);
if (bl.granted()) {
last_committed_index = log_index;
}
}
if (last_committed_index == 0) {
return 0;
}
// When removing a peer off the raft group which contains even number of
// peers, the quorum would decrease by 1, e.g. 3 of 4 changes to 2 of 3. In
// this case, the log after removal may be committed before some previous
// logs, since we use the new configuration to deal the quorum of the
// removal request, we think it's safe to commit all the uncommitted
// previous logs, which is not well proved right now
// TODO: add vlog when committing previous logs
for (int64_t index = _pending_index; index <= last_committed_index; ++index) {
_pending_meta_queue.pop_front();
}
_pending_index = last_committed_index + 1;
_last_committed_index.store(last_committed_index, butil::memory_order_relaxed);
lck.unlock();
// The order doesn't matter
_waiter->on_committed(last_committed_index);
return 0;
}首先从 _pending_meta_queue 取出对应的 Ballot，然后调用 Ballot::grant() 把 _quorum 减 1，然后判断 bl.granted() ，如果 _quorum 小于等于 0 则返回 true。_quorum 代表着 majority，是 peer 数量的一半加 1。如果 granted 了，就表明 majority 达成一致，更新 _last_committed_index，就调用FsmCaller::on_committed() 去执行状态机的操作，和 Follower 的提交类似。
_pending_meta_queue是一个std::deque<Ballot>，里面保存了每个 log index 的提交状态（Ballot）。Q：Raft 需要严格按顺序提交，这里是如何保证顺序的？
日志复制使用跟 empty AppendEntries 和 Heartbeat 相同的 RPC 定义和接口，因此前面提到的一些逻辑处理在日志复制中也可能发生，比如 Follower 复制太慢了，导致 next_index 在 Leader 上已经删除了，就会触发 Install Snapshot 等。
braft 对任务提交和日志复制的优化：
-
本地 IO Batch 写入：参数
raft_apply_batch定义每次 apply task 的 batch 数量，然后批量提交给 LogManager 写入本地磁盘 -
流水线复制：Leader 跟其他节点之间的 Log 同步是串行 batch 的方式，每个 batch 发送过程中之后到来的请求需要等待 batch 同步完成之后才能继续发送，这样会导致较长的延迟。这个可以通过 Leader 跟其他节点之间的 PipeLine 复制来改进，有效降低更新的延迟。
具体实现：所有的 RPC 调用都是异步的（异步访问），通过传递一个 done 实现回调。通过
raft_max_parallel_append_entries_rpc_num和raft_max_entries_size两个参数分别控制已经发出去但未收到回复的 batch 数量和 entries 数量的。 -
Leader 慢节点优化：RAFT 中 Client 的读写都通过 Leader 完成，一旦 Leader 出现 IO 慢节点，将会影响服务质量，需要对读写进行分别优化。 写入的时候Leader 需要先将 Log Entry 写到本地，然后再向其他节点进行复制，这样写入的延迟就是 IO_Leader + Min(IO_Others)，IO 延迟较高。其实 RAFT 的模型要求的是一条 LogEntry 在多数节点上写入成功即可认为是 Committed 状态，就可以向状态机进行 Apply，可以将 Leader 写本地和复制异步进行，只需要在内存中保存未 Committed 的 Log Entry，在多数节点已经应答的情况下，无需等待 Leader 本地 IO 完成，将内存中的 Log Entry 直接 Apply 给状态机即可。即使会造成持久化的 Base 数据比 Log 数据新，因为节点启动都是先加载上一个 Snapshot 再加载其后的 Log，对数据一致性也不会造成影响。
具体实现：
- apply task 提交日志到 LogManager 后就可以读取了，在
Replicator::_prepare_entry()的时候，通过LogManager::get_entry()取到的日志可能是内存中的日志，还没写到磁盘。 - 在
LogManager::append_entries()的最后会调用wakeup_all_waiter()唤醒等待读日志的 Replicator，此时日志已经写到了缓存，不保证已经持久化到了磁盘。
- apply task 提交日志到 LogManager 后就可以读取了，在