Revert "Reuse nodes for readDB."

This reverts commit 6ec4e06.
arangodb · maierlars · Jul 17, 2023 · May 26, 2023 · May 30, 2023 · May 30, 2023
commit 18d2f7f07adb03c52f2d13c7a3241643c37f29ac
diff --git a/arangod/Agency/Agent.cpp b/arangod/Agency/Agent.cpp
@@ -30,7 +30,6 @@
 
 #include "Agency/AgencyFeature.h"
 #include "Agency/AgentCallback.h"
-#include "Agency/Node.h"
 #include "Agency/Supervision.h"
 #include "ApplicationFeatures/ApplicationServer.h"
 #include "Basics/ReadLocker.h"
@@ -949,6 +948,8 @@ void Agent::advanceCommitIndex() {
   std::sort(temp.begin(), temp.end());
   index_t index = temp[temp.size() - quorum];
 
+  term_t t = _constituent.term();
+
   auto ci = _commitIndex.load(std::memory_order_relaxed);
   auto slices = _state.slices(ci + 1, index);
   {
@@ -960,11 +961,8 @@ void Agent::advanceCommitIndex() {
           << "Critical mass for commiting " << ci + 1 << " through " << index
           << " to read db";
 
-      auto node = _inflightNodes.find(index);
-      ADB_PROD_ASSERT(node != _inflightNodes.end())
-          << "inflight did not contain index " << index;
-      _readDB.setRootNode(node->second);
-      _inflightNodes.erase(_inflightNodes.begin(), ++node);
+      // Change _readDB and _commitIndex atomically together:
+      _readDB.applyLogEntries(slices, ci, t);
 
       LOG_TOPIC("e24aa", DEBUG, Logger::AGENCY)
           << "Critical mass for commiting " << ci + 1 << " through " << index
@@ -1124,15 +1122,17 @@ void Agent::load() {
     _supervision->start(this);
   }
 
-  if (size() > 1) {  // resilient agency only
-    TRI_ASSERT(_inception != nullptr);
+  if (_inception != nullptr) {  // resilient agency only
     _inception->start();
-    _loaded = true;
   } else {
     std::lock_guard guard{_ioLock};  // need this for callback to set _spearhead
     READ_LOCKER(guard2, _outputLock);
+    _spearhead = _readDB;
     activateAgency();
-    _loaded = true;
+  }
+
+  _loaded = true;
+  if (size() == 1) {
     endPrepareLeadership();
   }
 }
@@ -1293,9 +1293,6 @@ trans_ret_t Agent::transact(velocypack::Slice qs) {
                        ? _state.logLeaderSingle(query[0], currentTerm,
                                                 query[2].copyString())
                        : _state.logLeaderSingle(query[0], currentTerm);
-          ADB_PROD_ASSERT(_inflightNodes.empty() ||
-                          _inflightNodes.rbegin()->first < maxind);
-          _inflightNodes.emplace(maxind, _spearhead.nodePtr());
           ret->add(VPackValue(maxind));
         } else {
           _spearhead.read(res.failed->slice(), *ret);
@@ -1487,23 +1484,9 @@ write_ret_t Agent::write(velocypack::Slice query, WriteMode const& wmode) {
 
       std::lock_guard ioLocker{_ioLock};
 
-      std::vector<std::shared_ptr<Node const>> states;
-      applied = _spearhead.applyTransactions(chunk.slice(), wmode, &states);
-      auto indexes = _state.logLeaderMulti(chunk.slice(), applied, currentTerm);
-      ADB_PROD_ASSERT(states.size() == indexes.size())
-          << "produced " << states.size() << " states, but got "
-          << indices.size() << " indexes.";
-      for (std::size_t k = 0; k < states.size(); k++) {
-        if (indexes[k] == 0) {
-          continue;
-        }
-        ADB_PROD_ASSERT(_inflightNodes.empty() ||
-                        _inflightNodes.rbegin()->first < indexes[k]);
-        auto [iter, inserted] =
-            _inflightNodes.emplace(indexes[k], std::move(states[k]));
-        ADB_PROD_ASSERT(inserted) << "duplicated log index " << indexes[k];
-      }
-      indices.insert(indices.end(), indexes.begin(), indexes.end());
+      applied = _spearhead.applyTransactions(chunk.slice(), wmode);
+      auto tmp = _state.logLeaderMulti(chunk.slice(), applied, currentTerm);
+      indices.insert(indices.end(), tmp.begin(), tmp.end());
     }
     _write_hist_msec.count(
         duration<float, std::milli>(high_resolution_clock::now() - start)

diff --git a/arangod/Agency/Agent.h b/arangod/Agency/Agent.h
@@ -416,9 +416,6 @@ class Agent final : public arangodb::ServerThread<ArangodServer>,
   /// @brief Committed (read) kv-store
   Store _readDB;
 
-  // TODO use a deque instead?
-  std::map<index_t, std::shared_ptr<Node const>> _inflightNodes;
-
   /// @brief Committed (read) kv-store for transient data. This is
   /// protected by the _transientLock mutex.
   Store _transient;

diff --git a/arangod/Agency/Node.cpp b/arangod/Agency/Node.cpp
@@ -667,7 +667,7 @@ Node::Children const& Node::children() const {
 
 Builder Node::toBuilder() const {
   Builder builder;
-  toBuilder(builder, true);
+  toBuilder(builder);
   return builder;
 }
 

diff --git a/arangod/Agency/Store.cpp b/arangod/Agency/Store.cpp
@@ -87,14 +87,10 @@ index_t Store::applyTransactions(std::vector<log_t> const& queries) {
 /// Apply array of transactions multiple queries to store
 /// Return vector of according success
 std::vector<apply_ret_t> Store::applyTransactions(
-    VPackSlice query, Agent::WriteMode const& wmode,
-    std::vector<std::shared_ptr<Node const>>* states) {
+    VPackSlice query, Agent::WriteMode const& wmode) {
   std::vector<apply_ret_t> success;
 
   if (query.isArray()) {
-    if (states) {
-      states->reserve(query.length());
-    }
     try {
       for (auto const& i : VPackArrayIterator(query)) {
         if (!wmode.privileged()) {
@@ -111,27 +107,16 @@ std::vector<apply_ret_t> Store::applyTransactions(
             continue;
           }
         }
-        bool ok;
+
         std::lock_guard storeLocker{_storeLock};
-        if (states) {
-          states->emplace_back(nullptr);
-        }
         switch (i.length()) {
           case 1:  // No precondition
-            ok = applies(i[0]);
-            success.push_back(ok ? APPLIED : UNKNOWN_ERROR);
-            if (states) {
-              states->back() = _node;
-            }
+            success.push_back(applies(i[0]) ? APPLIED : UNKNOWN_ERROR);
             break;
           case 2:  // precondition + uuid
           case 3:
             if (check(i[1]).successful()) {
-              ok = applies(i[0]);
-              success.push_back(ok ? APPLIED : UNKNOWN_ERROR);
-              if (states) {
-                states->back() = _node;
-              }
+              success.push_back(applies(i[0]) ? APPLIED : UNKNOWN_ERROR);
             } else {  // precondition failed
               LOG_TOPIC("f6873", TRACE, Logger::AGENCY)
                   << "Precondition failed!";
@@ -778,8 +763,3 @@ void Store::registerPrefixTrigger(std::string const& prefix,
 std::vector<std::string> Store::split(std::string_view str) {
   return Node::split(str);
 }
-
-void Store::setRootNode(std::shared_ptr<const Node> root) noexcept {
-  std::lock_guard storeLocker{_storeLock};
-  _node = std::move(root);
-}
diff --git a/arangod/Agency/Store.h b/arangod/Agency/Store.h
@@ -100,8 +100,7 @@ class Store {
   /// in the next method.
   std::vector<apply_ret_t> applyTransactions(
       arangodb::velocypack::Slice query,
-      AgentInterface::WriteMode const& wmode = AgentInterface::WriteMode(),
-      std::vector<std::shared_ptr<Node const>>* states = nullptr);
+      AgentInterface::WriteMode const& wmode = AgentInterface::WriteMode());
 
   index_t applyTransactions(std::vector<log_t> const& queries);
 
@@ -167,8 +166,6 @@ class Store {
 
   void registerPrefixTrigger(std::string const& prefix, AgencyTriggerCallback);
 
-  void setRootNode(std::shared_ptr<Node const>) noexcept;
-
  private:
   /// @brief Apply single slice
   bool applies(arangodb::velocypack::Slice const&);