Raft Leader implementation. RaftCore is now finished.

This commit is contained in:
Pavel Kirienko
2015-05-08 14:41:15 +03:00
parent 6985c72dd3
commit a7c09ed714
3 changed files with 49 additions and 2 deletions
@@ -8,8 +8,8 @@ uint16 DEFAULT_REQUEST_TIMEOUT_MS = 100
uint16 DEFAULT_BASE_ELECTION_TIMEOUT_MS = 1000 # request timeout * max cluster size * 2 requests
uint32 term
uint32 prev_log_term
uint8 prev_log_index
uint8 prev_log_term
uint8 leader_commit
# Request message fits one CAN frame when there's no entries to send.
Entry[<=5] entries
@@ -139,6 +139,8 @@ enum TraceEvent
TraceRaftCommitIndexUpdate, // new commit index value
TraceRaftNewerTermInResponse, // new term value
TraceRaftNewEntryCommitted, // new commit index value
// 25
TraceRaftAppendEntriesCallFailure, // error code (may be negated)
NumTraceEventCodes
};
@@ -497,7 +499,7 @@ class RaftCore : private TimerBase
bool active_mode_;
ServerState server_state_;
uint8_t next_server_index_; ///< Next server to query for AE or RV RPC
uint8_t next_server_index_; ///< Next server to query AE from
uint8_t num_votes_received_in_this_campaign_;
PendingAppendEntriesFields pending_append_entries_fields_;
@@ -960,6 +960,51 @@ void RaftCore::updateCandidate()
void RaftCore::updateLeader()
{
propagateCommitIndex();
// Leader simply emits one AppendEntry at every update, iterating over all available servers
if (next_server_index_ >= cluster_.getClusterSize())
{
next_server_index_ = 0;
}
const NodeID node_id = cluster_.getRemoteServerNodeIDAtIndex(next_server_index_);
UAVCAN_ASSERT(node_id.isUnicast());
AppendEntries::Request req;
req.term = persistent_state_.getCurrentTerm();
req.leader_commit = commit_index_;
req.prev_log_index = Log::Index(cluster_.getServerNextIndex(node_id) - 1U);
const Entry* const entry = persistent_state_.getLog().getEntryAtIndex(req.prev_log_index);
if (entry == NULL)
{
UAVCAN_ASSERT(0);
handlePersistentStateUpdateError(-ErrLogic);
return;
}
req.prev_log_term = entry->term;
for (Log::Index index = cluster_.getServerNextIndex(node_id);
index <= persistent_state_.getLog().getLastIndex();
index++)
{
req.entries.push_back(*persistent_state_.getLog().getEntryAtIndex(index));
if (req.entries.size() == req.entries.capacity())
{
break;
}
}
pending_append_entries_fields_.num_entries = req.entries.size();
pending_append_entries_fields_.prev_log_index = req.prev_log_index;
const int res = append_entries_client_.call(node_id, req);
if (res < 0)
{
trace(TraceRaftAppendEntriesCallFailure, res);
}
}
void RaftCore::switchState(const ServerState new_state)