From a7c09ed714cceeb1e44e22077382f8e5186951c3 Mon Sep 17 00:00:00 2001 From: Pavel Kirienko Date: Fri, 8 May 2015 14:41:15 +0300 Subject: [PATCH] Raft Leader implementation. RaftCore is now finished. --- .../server/220.AppendEntries.uavcan | 2 +- .../dynamic_node_id_allocation_server.hpp | 4 +- .../uc_dynamic_node_id_allocation_server.cpp | 45 +++++++++++++++++++ 3 files changed, 49 insertions(+), 2 deletions(-) diff --git a/dsdl/uavcan/protocol/dynamic_node_id/server/220.AppendEntries.uavcan b/dsdl/uavcan/protocol/dynamic_node_id/server/220.AppendEntries.uavcan index 5551e160f5..2e49f4c26c 100644 --- a/dsdl/uavcan/protocol/dynamic_node_id/server/220.AppendEntries.uavcan +++ b/dsdl/uavcan/protocol/dynamic_node_id/server/220.AppendEntries.uavcan @@ -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 diff --git a/libuavcan/include/uavcan/protocol/dynamic_node_id_allocation_server.hpp b/libuavcan/include/uavcan/protocol/dynamic_node_id_allocation_server.hpp index 1fa392ec46..8c2ec8ce1e 100644 --- a/libuavcan/include/uavcan/protocol/dynamic_node_id_allocation_server.hpp +++ b/libuavcan/include/uavcan/protocol/dynamic_node_id_allocation_server.hpp @@ -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_; diff --git a/libuavcan/src/protocol/uc_dynamic_node_id_allocation_server.cpp b/libuavcan/src/protocol/uc_dynamic_node_id_allocation_server.cpp index c8a7ba4410..6ff4bfc61c 100644 --- a/libuavcan/src/protocol/uc_dynamic_node_id_allocation_server.cpp +++ b/libuavcan/src/protocol/uc_dynamic_node_id_allocation_server.cpp @@ -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)