Fix resolvePromise code wth bug in clang.

arangodb · maierlars · Sep 1, 2021 · Aug 17, 2021 · Aug 17, 2021 · Aug 17, 2021
commit d7e8e615a018f72124119353211aaf1a728219d0
diff --git a/arangod/Replication2/Streams/LogMultiplexer.tpp b/arangod/Replication2/Streams/LogMultiplexer.tpp
@@ -52,16 +52,32 @@
 namespace arangodb::replication2::streams {
 
 namespace {
-
+template <typename Descriptor, typename Queue, typename Result, typename Block = StreamInformationBlock<Descriptor>>
+auto allUnresolved(std::pair<Queue, Result>& q) {
+  return std::all_of(std::begin(q.first), std::end(q.first),
+                     [&](auto const& pair) { return !pair.second.isFulfilled(); });
+}
 template <typename Descriptor, typename Queue, typename Result, typename Block = StreamInformationBlock<Descriptor>>
 auto resolvePromiseSet(std::pair<Queue, Result>& q) {
-  std::for_each(std::begin(q.first), std::end(q.first),
-                [&](auto& pair) { pair.second.setValue(q.second); });
+  TRI_ASSERT(allUnresolved<Descriptor>(q));
+  std::for_each(std::begin(q.first), std::end(q.first), [&](auto& pair) {
+    TRI_ASSERT(!pair.second.isFulfilled());
+    if (!pair.second.isFulfilled()) {
+      pair.second.setValue(q.second);
+    }
+  });
+}
+
+template <typename... Descriptors, typename... Pairs, std::size_t... Idxs>
+auto resolvePromiseSets(stream_descriptor_set<Descriptors...>, std::index_sequence<Idxs...>, std::tuple<Pairs...>& pairs) {
+  (resolvePromiseSet<Descriptors>(std::get<Idxs>(pairs)), ...);
 }
 
 template <typename... Descriptors, typename... Pairs>
 auto resolvePromiseSets(stream_descriptor_set<Descriptors...>, std::tuple<Pairs...>& pairs) {
-  (resolvePromiseSet<Descriptors>(std::get<Pairs>(pairs)), ...);
+  TRI_ASSERT((allUnresolved<Descriptors>(std::get<Pairs>(pairs)) && ...));
+  resolvePromiseSets(stream_descriptor_set<Descriptors...>{}, std::index_sequence_for<Descriptors...>{}, pairs);
+
 }
 }  // namespace
 
@@ -234,7 +250,9 @@ struct LogMultiplexerImplementation
         return futures::Future<W>{std::in_place};
       }
       auto& block = std::get<StreamInformationBlock<StreamDescriptor>>(self._blocks);
-      return block.registerWaitFor(index);
+      return block.registerWaitFor(index).thenValue([index](auto&& value) {
+        return std::forward<decltype(value)>(value);
+      });
     });
   }
 
@@ -275,7 +293,8 @@ struct LogMultiplexerImplementation
 
   void triggerWaitForIndex(LogIndex waitForIndex) {
     auto f = _interface->waitFor(waitForIndex);
-    std::move(f).thenValue([weak = this->weak_from_this()](replicated_log::WaitForResult&& result) {
+    std::move(f).thenValue([weak = this->weak_from_this()](
+                               replicated_log::WaitForResult&& result) noexcept {
       // First lock the shared pointer
       if (auto locked = weak.lock(); locked) {
         auto that = std::static_pointer_cast<SelfClass>(locked);

diff --git a/tests/Replication2/Streams/MultiplexerConcurrencyTest.cpp b/tests/Replication2/Streams/MultiplexerConcurrencyTest.cpp
@@ -22,6 +22,7 @@
 
 #include <gtest/gtest.h>
 
+#include <Basics/VelocyPackHelper.h>
 #include <Replication2/Mocks/AsyncFollower.h>
 #include <utility>
 
@@ -56,7 +57,7 @@ struct LogMultiplexerConcurrencyTest : LogMultiplexerTestBase {
           auto [start, stop] = iter->range();
           TRI_ASSERT(start != stop);
           while (auto memtry = iter->next()) {
-            self->_state.emplace(*memtry);
+            self->_observedLog.emplace(*memtry);
           }
           self->waitForStream(stop);
         } else {
@@ -65,7 +66,7 @@ struct LogMultiplexerConcurrencyTest : LogMultiplexerTestBase {
       });
     }
 
-    std::map<LogIndex, ValueType> _state;
+    std::map<LogIndex, ValueType> _observedLog;
     std::shared_ptr<streams::Stream<ValueType>> _stream;
   };
 
@@ -122,11 +123,37 @@ TEST_F(LogMultiplexerConcurrencyTest, test) {
   auto leaderInstance = std::make_shared<LeaderInstance>(leader);
 
   auto producer = leaderInstance->_mux->getStreamById<test::my_int_stream_id>();
-  auto index = LogIndex{0};
-  for (std::size_t i = 0; i < 100000; i++) {
-    index = producer->insert(i);
-  }
 
-  leader->waitFor(index).wait();
+  constexpr std::size_t num_threads = 5;
+  constexpr std::size_t num_inserts_per_thread = 10000;
+
+  std::vector<std::thread> threads;
+  std::generate_n(std::back_inserter(threads), num_threads, [&]{
+    return std::thread([producer]{
+      auto index = LogIndex{0};
+      for (std::size_t i = 0; i < num_inserts_per_thread; i++) {
+        index = producer->insert(i);
+      }
+      producer->waitFor(index).wait();
+    });
+  });
+
+  std::for_each(std::begin(threads), std::end(threads), [](std::thread& t) {
+    t.join();
+  });
+  asyncFollower->waitFor(LogIndex{num_threads * num_inserts_per_thread + 1}).wait();
   asyncFollower->stop();
+
+  auto iterA = follower->waitForIterator(LogIndex{1}).get();
+  auto iterB = leader->waitForIterator(LogIndex{1}).get();
+
+  EXPECT_EQ(iterA->range(), iterB->range());
+  while (auto A = iterA->next()) {
+    auto B = iterB->next();
+    ASSERT_TRUE(B.has_value());
+    EXPECT_EQ(A->logIndex(), B->logIndex());
+    bool equal = basics::VelocyPackHelper::equal(A->logPayload(), B->logPayload(), true);
+    EXPECT_TRUE(equal) << A->logPayload().toJson() << " " << B->logPayload().toJson();
+  }
+  EXPECT_FALSE(iterB->next().has_value());
 }