More tests for multiplexer.

arangodb · maierlars · Sep 1, 2021 · Aug 17, 2021 · Aug 17, 2021 · Aug 17, 2021
commit 3127b80b60495e3f939f53b9cd1a8cf4b96024d9
diff --git a/arangod/Replication2/LogMultiplexer.h b/arangod/Replication2/LogMultiplexer.h
@@ -59,7 +59,7 @@ struct LogDemultiplexer
   virtual auto listen() -> void = 0;
 
 
-  static auto construct(std::shared_ptr<arangodb::replication2::replicated_log::LogFollower>)
+  static auto construct(std::shared_ptr<arangodb::replication2::replicated_log::ILogParticipant>)
       -> std::shared_ptr<LogDemultiplexer>;
 
  protected:

diff --git a/arangod/Replication2/LogMultiplexer.tpp b/arangod/Replication2/LogMultiplexer.tpp
@@ -65,22 +65,6 @@ auto resolvePromiseSets(stream_descriptor_set<Descriptors...>, std::tuple<Pairs.
 }
 }  // namespace
 
-template <typename Spec, typename Store>
-struct LogMultiplexerImplementationBase {
-  template <typename StreamDescriptor, typename T = stream_descriptor_type_t<StreamDescriptor>,
-            typename E = StreamEntryView<T>>
-  auto getIteratorInternal() -> std::unique_ptr<TypedLogRangeIterator<E>> {
-    using BlockType = StreamInformationBlock<StreamDescriptor>;
-
-    return _guardedData.template doUnderLock([](Store& self) {
-      auto& block = std::get<BlockType>(self._blocks);
-      return block.getAllEntriesIterator();
-    });
-  }
-
-  Guarded<Store, basics::UnshackledMutex> _guardedData;
-};
-
 template <typename Spec, typename Interface>
 struct LogDemultiplexerImplementation
     : LogDemultiplexer<Spec>,
@@ -346,9 +330,9 @@ struct LogMultiplexerImplementation
 }  // namespace arangodb::replication2::streams
 
 template <typename Spec>
-auto LogDemultiplexer<Spec>::construct(std::shared_ptr<replicated_log::LogFollower> interface)
+auto LogDemultiplexer<Spec>::construct(std::shared_ptr<replicated_log::ILogParticipant> interface)
     -> std::shared_ptr<LogDemultiplexer> {
-  return std::make_shared<streams::LogDemultiplexerImplementation<Spec, replicated_log::LogFollower>>(
+  return std::make_shared<streams::LogDemultiplexerImplementation<Spec, replicated_log::ILogParticipant>>(
       std::move(interface));
 }
 

diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
@@ -313,7 +313,7 @@ set(ARANGODB_REPLICATION2_TEST_SOURCES
         Replication2/Mocks/PersistedLog.h
         Replication2/ReplicatedLog/InMemoryLogTest.cpp
         Replication2/Streams/LogDemultiplexerTest.cpp
-)
+        Replication2/Mocks/FakeReplicatedLog.cpp Replication2/Mocks/FakeReplicatedLog.h)
 
 
 if (LINUX)

diff --git a/tests/Replication2/Mocks/FakeReplicatedLog.cpp b/tests/Replication2/Mocks/FakeReplicatedLog.cpp
@@ -0,0 +1,45 @@
+////////////////////////////////////////////////////////////////////////////////
+/// DISCLAIMER
+///
+/// Copyright 2021-2021 ArangoDB GmbH, Cologne, Germany
+///
+/// Licensed under the Apache License, Version 2.0 (the "License");
+/// you may not use this file except in compliance with the License.
+/// You may obtain a copy of the License at
+///
+///     http://www.apache.org/licenses/LICENSE-2.0
+///
+/// Unless required by applicable law or agreed to in writing, software
+/// distributed under the License is distributed on an "AS IS" BASIS,
+/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+/// See the License for the specific language governing permissions and
+/// limitations under the License.
+///
+/// Copyright holder is ArangoDB GmbH, Cologne, Germany
+///
+/// @author Lars Maier
+////////////////////////////////////////////////////////////////////////////////
+
+#include "FakeReplicatedLog.h"
+
+using namespace arangodb;
+using namespace arangodb::replication2;
+using namespace arangodb::replication2::replicated_log;
+using namespace arangodb::replication2::test;
+
+
+auto TestReplicatedLog::becomeFollower(ParticipantId const& id, LogTerm term, ParticipantId leaderId)
+-> std::shared_ptr<DelayedFollowerLog> {
+  auto ptr = ReplicatedLog::becomeFollower(id, term, std::move(leaderId));
+  return std::make_shared<DelayedFollowerLog>(ptr);
+}
+
+auto TestReplicatedLog::becomeLeader(ParticipantId const& id, LogTerm term,
+                                     std::vector<std::shared_ptr<replicated_log::AbstractFollower>> const& follower,
+                                     std::size_t writeConcern)
+    -> std::shared_ptr<replicated_log::LogLeader> {
+  LogConfig config;
+  config.writeConcern = writeConcern;
+  config.waitForSync = false;
+  return becomeLeader(config, id, term, follower);
+}
diff --git a/tests/Replication2/Mocks/FakeReplicatedLog.h b/tests/Replication2/Mocks/FakeReplicatedLog.h
@@ -0,0 +1,131 @@
+////////////////////////////////////////////////////////////////////////////////
+/// DISCLAIMER
+///
+/// Copyright 2021-2021 ArangoDB GmbH, Cologne, Germany
+///
+/// Licensed under the Apache License, Version 2.0 (the "License");
+/// you may not use this file except in compliance with the License.
+/// You may obtain a copy of the License at
+///
+///     http://www.apache.org/licenses/LICENSE-2.0
+///
+/// Unless required by applicable law or agreed to in writing, software
+/// distributed under the License is distributed on an "AS IS" BASIS,
+/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+/// See the License for the specific language governing permissions and
+/// limitations under the License.
+///
+/// Copyright holder is ArangoDB GmbH, Cologne, Germany
+///
+/// @author Lars Maier
+////////////////////////////////////////////////////////////////////////////////
+
+#pragma once
+#include <deque>
+
+#include "Replication2/Mocks/ReplicatedLogMetricsMock.h"
+#include "Replication2/ReplicatedLog/ILogParticipant.h"
+#include "Replication2/ReplicatedLog/InMemoryLog.h"
+#include "Replication2/ReplicatedLog/LogCore.h"
+#include "Replication2/ReplicatedLog/LogFollower.h"
+#include "Replication2/ReplicatedLog/LogLeader.h"
+#include "Replication2/ReplicatedLog/LogStatus.h"
+#include "Replication2/ReplicatedLog/PersistedLog.h"
+#include "Replication2/ReplicatedLog/ReplicatedLog.h"
+#include "Replication2/ReplicatedLog/types.h"
+
+namespace arangodb::replication2::test {
+
+struct DelayedFollowerLog : replicated_log::AbstractFollower, replicated_log::ILogParticipant {
+  explicit DelayedFollowerLog(std::shared_ptr<replicated_log::LogFollower> follower)
+      : _follower(std::move(follower)) {}
+
+  DelayedFollowerLog(LoggerContext const& logContext,
+                     std::shared_ptr<ReplicatedLogMetricsMock> logMetricsMock,
+                     ParticipantId const& id, std::unique_ptr<replicated_log::LogCore> logCore,
+                     LogTerm term, ParticipantId leaderId)
+      : DelayedFollowerLog([&] {
+          auto inMemoryLog = replicated_log::InMemoryLog{*logCore};
+          return std::make_shared<replicated_log::LogFollower>(
+              logContext, std::move(logMetricsMock), id, std::move(logCore),
+              term, std::move(leaderId), std::move(inMemoryLog));
+        }()) {}
+
+  auto appendEntries(replicated_log::AppendEntriesRequest req)
+      -> arangodb::futures::Future<replicated_log::AppendEntriesResult> override {
+    auto future = _asyncQueue.doUnderLock([&](auto& queue) {
+      return queue.emplace_back(std::make_shared<AsyncRequest>(std::move(req)))
+          ->promise.getFuture();
+    });
+    return std::move(future).thenValue([this](auto&& result) mutable {
+      return _follower->appendEntries(std::forward<decltype(result)>(result));
+    });
+  }
+
+  void runAsyncAppendEntries() {
+    auto asyncQueue = _asyncQueue.doUnderLock([](auto& _queue) {
+      auto queue = std::move(_queue);
+      _queue.clear();
+      return queue;
+    });
+
+    for (auto& p : asyncQueue) {
+      p->promise.setValue(std::move(p->request));
+    }
+  }
+
+  using WaitForAsyncPromise = futures::Promise<replicated_log::AppendEntriesRequest>;
+
+  struct AsyncRequest {
+    explicit AsyncRequest(replicated_log::AppendEntriesRequest request)
+        : request(std::move(request)) {}
+    replicated_log::AppendEntriesRequest request;
+    WaitForAsyncPromise promise;
+  };
+  [[nodiscard]] auto pendingAppendEntries() const
+      -> std::deque<std::shared_ptr<AsyncRequest>> {
+    return _asyncQueue.copy();
+  }
+  [[nodiscard]] auto hasPendingAppendEntries() const -> bool {
+    return _asyncQueue.doUnderLock(
+        [](auto const& queue) { return !queue.empty(); });
+  }
+
+  auto getParticipantId() const noexcept -> ParticipantId const& override {
+    return _follower->getParticipantId();
+  }
+
+  auto getStatus() const -> replicated_log::LogStatus override {
+    return _follower->getStatus();
+  }
+
+  [[nodiscard]] auto resign() && -> std::tuple<std::unique_ptr<replicated_log::LogCore>, DeferredAction> override {
+    return std::move(*_follower).resign();
+  }
+
+  auto waitFor(LogIndex index) -> WaitForFuture override { return _follower->waitFor(index); }
+
+  auto waitForIterator(LogIndex index) -> WaitForIteratorFuture override {
+    return _follower->waitForIterator(index);
+  }
+
+  auto release(LogIndex doneWithIdx) -> Result override {
+    return _follower->release(doneWithIdx);
+  }
+
+ private:
+  Guarded<std::deque<std::shared_ptr<AsyncRequest>>> _asyncQueue;
+  std::shared_ptr<replicated_log::LogFollower> _follower;
+};
+
+struct TestReplicatedLog : replicated_log::ReplicatedLog {
+  using ReplicatedLog::becomeLeader;
+  using ReplicatedLog::ReplicatedLog;
+  auto becomeFollower(ParticipantId const& id, LogTerm term, ParticipantId leaderId)
+      -> std::shared_ptr<DelayedFollowerLog>;
+
+  auto becomeLeader(ParticipantId const& id, LogTerm term,
+                    std::vector<std::shared_ptr<replicated_log::AbstractFollower>> const&,
+                    std::size_t writeConcern) -> std::shared_ptr<replicated_log::LogLeader>;
+};
+}  // namespace arangodb::replication2::test
diff --git a/tests/Replication2/ReplicatedLog/TestHelper.cpp b/tests/Replication2/ReplicatedLog/TestHelper.cpp
@@ -28,32 +28,3 @@
 #inclu
F438
de "Replication2/ReplicatedLog/types.h"
 
 #include <utility>
-
-using namespace arangodb;
-using namespace arangodb::replication2;
-using namespace arangodb::replication2::replicated_log;
-using namespace arangodb::replication2::test;
-
-
-auto TestReplicatedLog::becomeFollower(ParticipantId const& id, LogTerm term, ParticipantId leaderId)
-    -> std::shared_ptr<DelayedFollowerLog> {
-  auto ptr = ReplicatedLog::becomeFollower(id, term, std::move(leaderId));
-  return std::make_shared<DelayedFollowerLog>(ptr);
-}
-
-auto TestReplicatedLog::becomeLeader(LogConfig config, ParticipantId id, LogTerm term,
-                                     std::vector<std::shared_ptr<AbstractFollower>> const& follower)
-    -> std::shared_ptr<LogLeader> {
-  return ReplicatedLog::becomeLeader(config, std::move(id), term, follower);
-}
-
-auto TestReplicatedLog::becomeLeader(ParticipantId const& id, LogTerm term,
-                                     std::vector<std::shared_ptr<AbstractFollower>> const& follower,
-                                     std::size_t writeConcern)
-    -> std::shared_ptr<LogLeader> {
-  LogConfig config;
-  config.writeConcern = writeConcern;
-  config.waitForSync = false;
-
-  return becomeLeader(config, id, term, follower);
-}
diff --git a/tests/Replication2/ReplicatedLog/TestHelper.h b/tests/Replication2/ReplicatedLog/TestHelper.h
@@ -40,107 +40,13 @@
 #include <memory>
 #include <utility>
 
-#include "../Mocks/PersistedLog.h"
+#include "Replication2/Mocks/PersistedLog.h"
+#include "Replication2/Mocks/FakeReplicatedLog.h"
 
 namespace arangodb::replication2::test {
 
 using namespace replicated_log;
 
-
-
-struct DelayedFollowerLog : AbstractFollower {
-  explicit DelayedFollowerLog(std::shared_ptr<LogFollower> follower)
-      : _follower(std::move(follower)) {}
-
-  DelayedFollowerLog(LoggerContext const& logContext,
-                     std::shared_ptr<ReplicatedLogMetricsMock> logMetricsMock,
-                     ParticipantId const& id, std::unique_ptr<LogCore> logCore,
-                     LogTerm term, ParticipantId leaderId)
-      : DelayedFollowerLog([&] {
-          auto inMemoryLog = InMemoryLog{*logCore};
-          return std::make_shared<LogFollower>(logContext, std::move(logMetricsMock),
-                                               id, std::move(logCore), term,
-                                               std::move(leaderId),
-                                               std::move(inMemoryLog));
-        }()) {}
-
-  auto appendEntries(AppendEntriesRequest req)
-      -> arangodb::futures::Future<AppendEntriesResult> override {
-    auto future = _asyncQueue.doUnderLock([&](auto& queue) {
-      return queue.emplace_back(std::make_shared<AsyncRequest>(std::move(req)))
-          ->promise.getFuture();
-    });
-    return std::move(future).thenValue(
-        [this](auto&& result) mutable {
-          return _follower->appendEntries(std::forward<decltype(result)>(result));
-        });
-  }
-
-  void runAsyncAppendEntries() {
-    auto asyncQueue = _asyncQueue.doUnderLock([](auto& _queue) {
-      auto queue = std::move(_queue);
-      _queue.clear();
-      return queue;
-    });
-
-    for (auto& p : asyncQueue) {
-      p->promise.setValue(std::move(p->request));
-    }
-  }
-
-  using WaitForAsyncPromise = futures::Promise<AppendEntriesRequest>;
-
-  struct AsyncRequest {
-    explicit AsyncRequest(AppendEntriesRequest request)
-        : request(std::move(request)) {}
-    AppendEntriesRequest request;
-    WaitForAsyncPromise promise;
-  };
-  [[nodiscard]] auto pendingAppendEntries() const
-  -> std::deque<std::shared_ptr<AsyncRequest>> {
-    return _asyncQueue.copy();
-  }
-  [[nodiscard]] auto hasPendingAppendEntries() const -> bool {
-    return _asyncQueue.doUnderLock(
-        [](auto const& queue) { return !queue.empty(); });
-  }
-
-  auto getParticipantId() const noexcept -> ParticipantId const& override {
-    return _follower->getParticipantId();
-  }
-
-  auto getStatus() const -> LogStatus {
-    return _follower->getStatus();
-  }
-
-  auto resign() && {
-    return std::move(*_follower).resign();
-  }
-
-  auto waitFor(LogIndex index) {
-    return _follower->waitFor(index);
-  }
-
-  auto waitForIterator(LogIndex index) {
-    return _follower->waitForIterator(index);
-  }
- private:
-  Guarded<std::deque<std::shared_ptr<AsyncRequest>>> _asyncQueue;
-  std::shared_ptr<LogFollower> _follower;
-};
-
-struct TestReplicatedLog : ReplicatedLog {
-  using ReplicatedLog::ReplicatedLog;
-  auto becomeFollower(ParticipantId const& id, LogTerm term, ParticipantId leaderId)
-      -> std::shared_ptr<DelayedFollowerLog>;
-  auto becomeLeader(ParticipantId const& id, LogTerm term,
-                    std::vector<std::shared_ptr<AbstractFollower>> const& follower,
-                    std::size_t writeConcern) -> std::shared_ptr<LogLeader>;
-  auto becomeLeader(LogConfig config, ParticipantId id, LogTerm term,
-                    std::vector<std::shared_ptr<AbstractFollower>> const& follower)
-      -> std::shared_ptr<LogLeader>;
-};
-
 struct ReplicatedLogTest : ::testing::Test {
 
   auto makeLogCore(LogId id) -> std::unique_ptr<LogCore> {