Minor changes

arangodb · mchacki · Sep 9, 2021 · Jun 22, 2021 · Jun 23, 2021 · Jun 23, 2021
commit aa2245b04a0c41a6e7b64542a09145b098aa18cc
diff --git a/arangod/Replication2/ReplicatedLog/LogCommon.cpp b/arangod/Replication2/ReplicatedLog/LogCommon.cpp
@@ -51,6 +51,14 @@ auto replication2::operator<<(std::ostream& os, const LogIndex& idx) -> std::ost
   return os << idx.value;
 }
 
+auto LogIndex::saturatedDecrement() const noexcept -> LogIndex {
+  if (value > 0) {
+    return LogIndex{value - 1};
+  }
+
+  return LogIndex{0};
+}
+
 LogEntry::LogEntry(LogTerm logTerm, LogIndex logIndex, LogPayload payload)
     : _logTerm{logTerm}, _logIndex{logIndex}, _payload{std::move(payload)} {}
 

diff --git a/arangod/Replication2/ReplicatedLog/LogCommon.h b/arangod/Replication2/ReplicatedLog/LogCommon.h
@@ -64,6 +64,8 @@ struct LogIndex : implement_compare<LogIndex> {
   constexpr explicit LogIndex(std::uint64_t value) noexcept : value{value} {}
   std::uint64_t value;
 
+  [[nodiscard]] auto saturatedDecrement() const noexcept -> LogIndex;
+
   [[nodiscard]] auto operator<=(LogIndex) const -> bool;
 
   auto operator+(std::uint64_t delta) const -> LogIndex;

diff --git a/arangod/Replication2/ReplicatedLog/LogLeader.cpp b/arangod/Replication2/ReplicatedLog/LogLeader.cpp
@@ -167,12 +167,8 @@ void replicated_log::LogLeader::executeAppendEntriesRequests(
 
   for (auto& it : requests) {
     if (it.has_value()) {
-      // Capture self(shared_from_this()) instead of this
-      // additionally capture a weak pointer that will be locked
-      // when the request returns. If the locking is successful
-      // we are still in the same term.
       delayedFuture(it->_executionDelay).thenFinal([it = it, logMetrics](auto&&) mutable {
-        // we need to lock here, because we access _follower
+        // Let's check whether our log is still there, and abort otherwise
         auto guard = it->_parentLog.lock();
         if (guard == nullptr) {
           LOG_TOPIC("de312", DEBUG, Logger::REPLICATION2)
@@ -184,8 +180,11 @@ void replicated_log::LogLeader::executeAppendEntriesRequests(
         LOG_CTX("1b0ec", TRACE, it->_follower->logContext)
             << "sending append entries, messageId = " << messageId;
         auto startTime = std::chrono::steady_clock::now();
+        // Capture a weak pointer `parentLog` that will be locked
+        // when the request returns. If the locking is successful
+        // we are still in the same term.
         it->_follower->_impl->appendEntries(std::move(it->_request))
-            .thenFinal([parentLog = it->_parentLog, &follower = *it->_follower,
+            .thenFinal([parentLog = it->_parentLog, follower = it->_follower,
                         lastIndex = it->_lastIndex, currentCommitIndex = it->_currentCommitIndex,
                         currentTerm = it->_currentTerm, messageId = messageId, startTime,
                         logMetrics = logMetrics](futures::Try<AppendEntriesResult>&& res) noexcept {
@@ -199,18 +198,18 @@ void replicated_log::LogLeader::executeAppendEntriesRequests(
                 using namespace std::chrono_literals;
                 auto const duration = endTime - startTime;
                 self->_logMetrics->replicatedLogAppendEntriesRttUs->count(duration / 1us);
-                LOG_CTX("8ff44", TRACE, follower.logContext)
+                LOG_CTX("8ff44", TRACE, follower->logContext)
                     << "received append entries response, messageId = " << messageId;
                 auto [preparedRequests, resolvedPromises] = std::invoke(
                     [&]() -> std::pair<std::vector<std::optional<PreparedAppendEntryRequest>>, ResolvedPromiseSet> {
                       auto guarded = self->acquireMutex();
                       if (!guarded->_didResign) {
                         // Is throwing the right thing to do here? - No, we are in a finally
                         return guarded->handleAppendEntriesResponse(
-                            parentLog, follower, lastIndex, currentCommitIndex, currentTerm,
+                            parentLog, *follower, lastIndex, currentCommitIndex, currentTerm,
                             std::move(res), endTime - startTime, messageId);
                       } else {
-                        LOG_CTX("da116", DEBUG, follower.logContext)
+                        LOG_CTX("da116", DEBUG, follower->logContext)
                             << "received response from follower but leader "
                                "already resigned, messageId = "
                             << messageId;
@@ -500,7 +499,7 @@ auto replicated_log::LogLeader::GuardedLeaderData::prepareAppendEntry(
   req.leaderTerm = _self._currentTerm;
   req.leaderId = _self._id;
   req.waitForSync = _self._config.waitForSync;
-  req.messageId = ++follower.lastSendMessageId;
+  req.messageId = ++follower.lastSentMessageId;
 
   if (lastAcked) {
     req.prevLogIndex = lastAcked->logIndex();
@@ -534,7 +533,7 @@ auto replicated_log::LogLeader::GuardedLeaderData::prepareAppendEntry(
 
   follower.requestInFlight = true;
   auto preparedRequest = PreparedAppendEntryRequest{};
-  preparedRequest._follower = std::shared_ptr<FollowerInfo>(parentLog.lock(), &follower);
+  preparedRequest._follower = std::shared_ptr<FollowerInfo>(_self.shared_from_this(), &follower);
   preparedRequest._currentTerm = _self._currentTerm;
   preparedRequest._currentCommitIndex = currentCommitIndex;
   preparedRequest._lastIndex = lastIndex;
@@ -569,7 +568,7 @@ auto replicated_log::LogLeader::GuardedLeaderData::handleAppendEntriesResponse(
 
   follower._lastRequestLatency = latency;
 
-  if (follower.lastSendMessageId == messageId) {
+  if (follower.lastSentMessageId == messageId) {
     LOG_CTX("35a32", TRACE, follower.logContext)
         << "received message " << messageId << " - no other requests in flight";
     // there is no request in flight currently
@@ -578,7 +577,7 @@ auto replicated_log::LogLeader::GuardedLeaderData::handleAppendEntriesResponse(
   if (res.hasValue()) {
     auto& response = res.get();
     TRI_ASSERT(messageId == response.messageId);
-    if (follower.lastSendMessageId == response.messageId) {
+    if (follower.lastSentMessageId == response.messageId) {
       LOG_CTX("35134", TRACE, follower.logContext)
           << "received append entries response, messageId = " << response.messageId
           << ", errorCode = " << to_string(response.errorCode)
@@ -596,26 +595,23 @@ auto replicated_log::LogLeader::GuardedLeaderData::handleAppendEntriesResponse(
           case AppendEntriesErrorReason::NO_PREV_LOG_MATCH:
             follower.numErrorsSinceLastAnswer = 0;
             TRI_ASSERT(response.conflict.has_value());
-            follower.lastAckedEntry.index = response.conflict.value().index;
+            follower.lastAckedEntry.index = response.conflict.value().index.saturatedDecrement();
             LOG_CTX("33c6d", DEBUG, follower.logContext)
                 << "reset last acked index to " << follower.lastAckedEntry;
-            if (follower.lastAckedEntry.index > LogIndex{0}) {
-              follower.lastAckedEntry.index = LogIndex{follower.lastAckedEntry.index.value - 1};
-            }
             break;
           default:
             LOG_CTX("1bd0b", DEBUG, follower.logContext)
                 << "received error from follower, reason = " << to_string(response.reason)
                 << " message id = " << messageId;
-            follower.numErrorsSinceLastAnswer += 1;
+            ++follower.numErrorsSinceLastAnswer;
         }
       }
     } else {
       LOG_CTX("056a8", DEBUG, follower.logContext)
           << "received outdated response from follower "
           << follower._impl->getParticipantId() << ": "
           << response.messageId << ", expected " << messageId
-          << ", latest " << follower.lastSendMessageId;
+          << ", latest " << follower.lastSentMessageId;
     }
   } else if (res.hasException()) {
     ++follower.numErrorsSinceLastAnswer;
@@ -638,6 +634,7 @@ auto replicated_log::LogLeader::GuardedLeaderData::handleAppendEntriesResponse(
     TRI_ASSERT(false);
     FATAL_ERROR_EXIT();
   }
+
   // try sending the next batch
   return std::make_pair(prepareAppendEntries(parentLog), std::move(toBeResolved));
 }

diff --git a/arangod/Replication2/ReplicatedLog/LogLeader.h b/arangod/Replication2/ReplicatedLog/LogLeader.h
@@ -122,7 +122,7 @@ class LogLeader : public std::enable_shared_from_this<LogLeader>, public LogPart
     std::shared_ptr<AbstractFollower> _impl;
     TermIndexPair lastAckedEntry = TermIndexPair{LogTerm{0}, LogIndex{0}};
     LogIndex lastAckedCommitIndex = LogIndex{0};
-    MessageId lastSendMessageId{0};
+    MessageId lastSentMessageId{0};
     std::size_t numErrorsSinceLastAnswer = 0;
     AppendEntriesErrorReason lastErrorReason = AppendEntriesErrorReason::NONE;
     LoggerContext const logContext;
@@ -152,11 +152,20 @@ class LogLeader : public std::enable_shared_from_this<LogLeader>, public LogPart
 
   struct PreparedAppendEntryRequest {
     // TODO Write a constructor, delete the default constructor
+    // TODO Make this a weak_ptr. It will always be an alias of _parentLog.
     std::shared_ptr<FollowerInfo> _follower;
+    // TODO Currently, the AppendEntriesRequest is built before we wait for
+    //      the _executionDelay. This is suboptimal in the sense that we might
+    //      be able to squeeze some additional log entries in the request if we
+    //      build it only after the wait.
+    //      So the task is to remove _request from this struct, and only build
+    //      it right before the request should actually be sent.
     AppendEntriesRequest _request;
     std::weak_ptr<LogLeader> _parentLog;
     TermIndexPair _lastIndex;
     LogIndex _currentCommitIndex;
+    // TODO _currentTerm is probably not necessary here, we could read it in
+    //     executeAppendEntries, as we don't need to acquire the lock to read it.
     LogTerm _currentTerm;
     std::chrono::steady_clock::duration _executionDelay;
   };