目录
下面以一个我手写的raft的一个函数为例(逻辑可能存在问题，勿喷)
同步调用
这里有个逻辑是要向peer列表发送RequestVote请求，这里我暂时采用的是多future + 同步request的方式调用的，并在lbd函数的开头增加了快速判断退出
这里我的预期是多个请求中只要有超过半数的请求同意或者发现更高的任期其他的请求就不需要执行了，我也不需要进行等待结果了。
很显然用同步request是达不到这样的结果的，因为
grpc::Status status = stub->RequestVote(&context, request, &response);
这一步是阻塞调用的
如果我其他的线程已经返回了，并且票数也超过了半数
有一个线程一直卡在获取status这一步，下面的wait仍然会阻塞住
for (auto& f : futures) {
f.wait();
}
这会影响性能
/**
* @brief 开始领导人选举过程
* 1.增加当前任期并为自己投票。
* 2.重置选举定时器，防止在选举过程中触发新的选举。
* 3.准备 RequestVote RPC 的请求参数。
* 4.向所有其他节点发送 RequestVote RPC。
* 5.处理每个节点的响应：
* 6.如果发现更高的任期，立即变为跟随者。
* 7.统计获得的选票。
* 8.如果获得多数票，立即成为领导者。
* 处理 RPC 调用失败的情况。
* 如果选举失败（没有获得多数票），保持候选人状态。选举定时器会在之后触发新的选举。
* 注意事项：
* 这个实现使用了同步 RPC 调用，在实际应用中可能需要考虑使用异步调用以提高性能。
* 需要确保 becomeLeader() 方法正确实现，包括初始化领导者状态和开始发送心跳。
* 可能需要添加超时机制，以防止在网络问题时无限等待。
* 在实际部署中，可能需要考虑更多的错误处理和日志记录。
* 这个实现提供了 Raft 选举算法的基本框架，您可以根据具体需求进行进一步的优化和调整。
*/
void RaftNode::startElection() {
currentTerm_++;
votedFor_ = id_;
persistVotedFor(); // 持久化投票状态
resetElectionTimer();
int votesNeeded = (peers_.size() + 1) / 2 + 1;
std::atomic<int> votesReceived{1}; // 包括自己的一票
std::atomic<bool> electionDecided{false};
raft::RequestVoteRequest request;
request.set_term(currentTerm_);
request.set_candidateid(id_);
request.set_lastlogindex(getLastLogIndex());
request.set_lastlogterm(getLastLogTerm());
std::vector<std::future<void>> futures;
for (const auto& peer : peers_) {
futures.push_back(std::async(std::launch::async, [this, &request, &votesReceived, &electionDecided, &votesNeeded, peer]() {
if (electionDecided) return;
std::unique_ptr<raft::RaftService::Stub> stub(raft::RaftService::NewStub(
grpc::CreateChannel(peer, grpc::InsecureChannelCredentials())));
grpc::ClientContext context;
raft::RequestVoteResponse response;
grpc::Status status = stub->RequestVote(&context, request, &response);
if (status.ok()) {
std::lock_guard<std::mutex> lock(mutex_);
// 一旦获得足够的票数或发现更高的任期，立即标记选举结束，避免等待剩余的 RPC。
if (response.term() > currentTerm_) {
becomeFollower(response.term());
electionDecided = true;
return;
}
if (response.votegranted()) {
if (++votesReceived >= votesNeeded) {
becomeLeader();
electionDecided = true;
}
}
} else {
LOG(ERROR) << "RPC failed for peer " << peer << ": " << status.error_message();
}
}));
}
for (auto& f : futures) {
f.wait();
}
if (!electionDecided) {
LOG(INFO) << "Election ended without a clear winner. Remaining as candidate.";
}
}
异步调用
以使用 gRPC 的异步 API 和超时机制来优化一下
void RaftNode::startElection() {
currentTerm_++;
votedFor_ = id_;
persistVotedFor(); // 持久化投票状态
resetElectionTimer();
int votesNeeded = (peers_.size() + 1) / 2 + 1;
std::atomic<int> votesReceived{1}; // 包括自己的一票
std::atomic<bool> electionDecided{false};
std::atomic<int> responsesReceived{0};
raft::RequestVoteRequest request;
request.set_term(currentTerm_);
request.set_candidateid(id_);
request.set_lastlogindex(getLastLogIndex());
request.set_lastlogterm(getLastLogTerm());
grpc::CompletionQueue cq;
std::vector<std::unique_ptr<grpc::ClientAsyncResponseReader<raft::RequestVoteResponse>>> rpc_calls;
std::vector<raft::RequestVoteResponse> responses(peers_.size());
std::vector<grpc::Status> statuses(peers_.size());
std::vector<grpc::ClientContext> contexts(peers_.size());
for (size_t i = 0; i < peers_.size(); ++i) {
contexts[i].set_deadline(std::chrono::system_clock::now() + std::chrono::milliseconds(100)); // 100ms per RPC
auto stub = raft::RaftService::NewStub(grpc::CreateChannel(peers_[i], grpc::InsecureChannelCredentials()));
auto rpc = stub->AsyncRequestVote(&contexts[i], request, &cq);
rpc->Finish(&responses[i], &statuses[i], (void*)i);
rpc_calls.push_back(std::move(rpc));
}
auto electionDeadline = std::chrono::system_clock::now() + std::chrono::milliseconds(200); // 200ms total election timeout
while (!electionDecided && responsesReceived < peers_.size() && std::chrono::system_clock::now() < electionDeadline) {
void* got_tag;
bool ok = false;
auto status = cq.AsyncNext(&got_tag, &ok, electionDeadline);
if (status == grpc::CompletionQueue::TIMEOUT) continue;
if (ok) {
size_t index = (size_t)got_tag;
responsesReceived++;
if (statuses[index].ok()) {
std::lock_guard<std::mutex> lock(mutex_);
if (responses[index].term() > currentTerm_) {
becomeFollower(responses[index].term());
electionDecided = true;
} else if (responses[index].votegranted()) {
if (++votesReceived >= votesNeeded) {
becomeLeader();
electionDecided = true;
}
}
} else {
LOG(ERROR) << "RPC failed for peer " << peers_[index] << ": " << statuses[index].error_message();
}
}
}
if (!electionDecided) {
LOG(INFO) << "Election ended without a clear winner. Remaining as candidate.";
}
}
这个优化版本有以下改进：
- 使用 gRPC 的异步 API：我们使用
AsyncRequestVote而不是同步的RequestVote。这允许我们同时发起多个 RPC 调用而不会阻塞。 - 使用 CompletionQueue：这允许我们等待任何 RPC 调用完成，而不是按顺序等待。
- 添加超时机制：我们设置了一个 200ms 的超时，防止无限等待。
- 非阻塞轮询：我们使用 AsyncNext 来非阻塞地检查是否有完成的 RPC 调用。
- 早期退出：一旦选举决定（获得多数票或发现更高任期），我们就停止等待其他 RPC 响应。
这个实现有以下优点：
- 并发性更好：所有 RPC 调用同时进行。
- 响应更快：一旦获得足够的票数，立即结束选举，不会等待剩余的 RPC。
- 有超时保护：防止因网络问题导致的长时间等待。
- 更高效：不会在单个 RPC 调用上阻塞。
这个实现应该能够解决您提到的性能问题，同时保持 Raft 算法的正确性。但请注意，这个实现比之前的版本更复杂，可能需要更多的测试来确保其正确性。
整个选举过程有一个总体的时间限制（在这个例子中是 200ms）。
单个 RPC 请求没有独立的超时设置，它们的超时受总体 deadline 的限制。
这种方法的优点是它确保了选举过程不会无限期地持续，同时允许尽可能多的响应在时间限制内被处理。然而，它也意味着如果某些 RPC 请求特别慢，它们可能会在总体超时到达之前就被截断。
为每个单独的 RPC 请求设置超时，需要在创建 ClientContext 时为每个请求单独设置截止时间。例如：
grpc::ClientContext context;
context.set_deadline(std::chrono::system_clock::now() + std::chrono::milliseconds(100)); // 100ms per RPC
这样可以确保每个 RPC 请求都有自己的超时，但可能会使整体选举过程的控制变得更复杂。
No comments yet.