Merge branch 'T0941-MG-implement-basic-raft-version' of github.com:memgraph/memgraph into T0912-MG-in-memory-shard-map

.github/workflows/diff.yaml

@@ -171,7 +171,7 @@ jobs:
 
           # Run leftover CTest tests (all except unit and benchmark tests).
           cd build
-          ctest -E "(memgraph__unit|memgraph__benchmark)" --output-on-failure
+          ctest -E "(memgraph__unit|memgraph__benchmark|memgraph__simulation)" --output-on-failure
 
       - name: Run drivers tests
         run: |
@@ -262,6 +262,15 @@ jobs:
           cd build
           ctest -R memgraph__unit --output-on-failure -j$THREADS
 
+      - name: Run simulation tests
+        run: |
+          # Activate toolchain.
+          source /opt/toolchain-v4/activate
+
+          # Run unit tests.
+          cd build
+          ctest -R memgraph__simulation --output-on-failure -j$THREADS
+
       - name: Run e2e tests
         run: |
           # TODO(gitbuda): Setup mgclient and pymgclient properly.

src/io/CMakeLists.txt

@@ -2,11 +2,7 @@ set(io_src_files
     network/addrinfo.cpp
     network/endpoint.cpp
     network/socket.cpp
-    network/utils.cpp
-    future.hpp
-    address.hpp
-    errors.hpp
-    transport.hpp)
+    network/utils.cpp)
 
 find_package(fmt REQUIRED)
 find_package(Threads REQUIRED)

src/io/address.hpp

@@ -13,8 +13,10 @@
 
 #include <compare>
 
+#include <fmt/format.h>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/uuid/uuid.hpp>
+#include <boost/uuid/uuid_io.hpp>
 
 namespace memgraph::io {
 struct Address {
@@ -36,21 +38,26 @@ struct Address {
   }
 
   bool operator==(const Address &other) const {
-    return ((unique_id == other.unique_id) && last_known_ip == other.last_known_ip) &&
+    return (unique_id == other.unique_id) && (last_known_ip == other.last_known_ip) &&
            (last_known_port == other.last_known_port);
   }
 
   /// unique_id is most dominant for ordering, then last_known_ip, then last_known_port
   bool operator<(const Address &other) const {
-    if (unique_id == other.unique_id) {
-      if (last_known_ip == other.last_known_ip) {
-        return last_known_port < other.last_known_port;
-      } else {
-        return last_known_ip < other.last_known_ip;
-      }
-    } else {
+    if (unique_id != other.unique_id) {
       return unique_id < other.unique_id;
     }
+
+    if (last_known_ip != other.last_known_ip) {
+      return last_known_ip < other.last_known_ip;
+    }
+
+    return last_known_port < other.last_known_port;
+  }
+
+  std::string ToString() const {
+    return fmt::format("Address {{ unique_id: {}, last_known_ip: {}, last_known_port: {} }}",
+                       boost::uuids::to_string(unique_id), last_known_ip.to_string(), last_known_port);
   }
 };
 };  // namespace memgraph::io

src/io/future.hpp

@@ -27,11 +27,11 @@ namespace memgraph::io {
 // construct a Promise or Future is to pass a Shared in. This
 // ensures that Promises and Futures can only be constructed
 // in this translation unit.
-namespace {
+namespace details {
 template <typename T>
 class Shared {
-  std::condition_variable cv_;
-  std::mutex mu_;
+  mutable std::condition_variable cv_;
+  mutable std::mutex mu_;
   std::optional<T> item_;
   bool consumed_ = false;
   bool waiting_ = false;
@@ -47,8 +47,7 @@ class Shared {
   ~Shared() = default;
 
   /// Takes the item out of our optional item_ and returns it.
-  /// Requires caller holds mutex, proving it by passing reference.
-  T Take(std::unique_lock<std::mutex> &) {
+  T Take() {
     MG_ASSERT(item_, "Take called without item_ being present");
     MG_ASSERT(!consumed_, "Take called on already-consumed Future");
 
@@ -87,7 +86,7 @@ class Shared {
           break;
         }
       }
-      if (!simulator_progressed) {
+      if (!simulator_progressed) [[likely]] {
         cv_.wait(lock);
       }
       MG_ASSERT(!consumed_, "Future consumed twice!");
@@ -95,7 +94,7 @@ class Shared {
 
     waiting_ = false;
 
-    return Take(lock);
+    return Take();
   }
 
   bool IsReady() {
@@ -107,10 +106,10 @@ class Shared {
     std::unique_lock<std::mutex> lock(mu_);
 
     if (item_) {
-      return Take(lock);
-    } else {
-      return std::nullopt;
+      return Take();
     }
+
+    return std::nullopt;
   }
 
   void Fill(T item) {
@@ -131,25 +130,25 @@ class Shared {
     return waiting_;
   }
 };
-}  // namespace
+}  // namespace details
 
 template <typename T>
 class Future {
   bool consumed_or_moved_ = false;
-  std::shared_ptr<Shared<T>> shared_;
+  std::shared_ptr<details::Shared<T>> shared_;
 
  public:
-  explicit Future(std::shared_ptr<Shared<T>> shared) : shared_(shared) {}
+  explicit Future(std::shared_ptr<details::Shared<T>> shared) : shared_(shared) {}
 
   Future() = delete;
-  Future(Future &&old) {
+  Future(Future &&old) noexcept {
     MG_ASSERT(!old.consumed_or_moved_, "Future moved from after already being moved from or consumed.");
     shared_ = std::move(old.shared_);
     consumed_or_moved_ = old.consumed_or_moved_;
     old.consumed_or_moved_ = true;
   }
 
-  Future &operator=(Future &&old) {
+  Future &operator=(Future &&old) noexcept {
     MG_ASSERT(!old.consumed_or_moved_, "Future moved from after already being moved from or consumed.");
     shared_ = std::move(old.shared_);
     old.consumed_or_moved_ = true;
@@ -198,20 +197,20 @@ class Future {
 
 template <typename T>
 class Promise {
-  std::shared_ptr<Shared<T>> shared_;
+  std::shared_ptr<details::Shared<T>> shared_;
   bool filled_or_moved_ = false;
 
  public:
-  explicit Promise(std::shared_ptr<Shared<T>> shared) : shared_(shared) {}
+  explicit Promise(std::shared_ptr<details::Shared<T>> shared) : shared_(shared) {}
 
   Promise() = delete;
-  Promise(Promise &&old) {
+  Promise(Promise &&old) noexcept {
     MG_ASSERT(!old.filled_or_moved_, "Promise moved from after already being moved from or filled.");
     shared_ = std::move(old.shared_);
     old.filled_or_moved_ = true;
   }
 
-  Promise &operator=(Promise &&old) {
+  Promise &operator=(Promise &&old) noexcept {
     MG_ASSERT(!old.filled_or_moved_, "Promise moved from after already being moved from or filled.");
     shared_ = std::move(old.shared_);
     old.filled_or_moved_ = true;
@@ -242,7 +241,7 @@ class Promise {
 
 template <typename T>
 std::pair<Future<T>, Promise<T>> FuturePromisePair() {
-  std::shared_ptr<Shared<T>> shared = std::make_shared<Shared<T>>();
+  std::shared_ptr<details::Shared<T>> shared = std::make_shared<details::Shared<T>>();
 
   Future<T> future = Future<T>(shared);
   Promise<T> promise = Promise<T>(shared);
@@ -252,7 +251,7 @@ std::pair<Future<T>, Promise<T>> FuturePromisePair() {
 
 template <typename T>
 std::pair<Future<T>, Promise<T>> FuturePromisePairWithNotifier(std::function<bool()> simulator_notifier) {
-  std::shared_ptr<Shared<T>> shared = std::make_shared<Shared<T>>(simulator_notifier);
+  std::shared_ptr<details::Shared<T>> shared = std::make_shared<details::Shared<T>>(simulator_notifier);
 
   Future<T> future = Future<T>(shared);
   Promise<T> promise = Promise<T>(shared);

src/io/rsm/raft.hpp

@@ -29,10 +29,12 @@
 namespace memgraph::io::rsm {
 
 using memgraph::io::Address;
+using memgraph::io::Duration;
 using memgraph::io::Io;
 using memgraph::io::ResponseEnvelope;
 using memgraph::io::ResponseFuture;
 using memgraph::io::ResponseResult;
+using memgraph::io::Time;
 using memgraph::io::simulator::Simulator;
 using memgraph::io::simulator::SimulatorConfig;
 using memgraph::io::simulator::SimulatorStats;
@@ -40,8 +42,6 @@ using memgraph::io::simulator::SimulatorTransport;
 
 using Term = uint64_t;
 using LogIndex = uint64_t;
-using Time = uint64_t;
-using Duration = uint64_t;
 using RequestId = uint64_t;
 
 template <typename WriteOperation>
@@ -132,14 +132,14 @@ struct PendingClientRequest {
 struct Leader {
   std::map<Address, FollowerTracker> followers;
   std::deque<PendingClientRequest> pending_client_requests;
-  Time last_broadcast = 0;
+  Time last_broadcast = Time::min();
 
   void Print() { std::cout << "\tLeader   \t"; }
 };
 
 struct Candidate {
   std::map<Address, LogIndex> successful_votes;
-  Time election_began = 0;
+  Time election_began = Time::min();
   std::set<Address> outstanding_votes;
 
   void Print() { std::cout << "\tCandidate\t"; }
@@ -327,8 +327,22 @@ class Raft {
   // Raft paper - 5.2
   // Raft uses randomized election timeouts to ensure that split votes are rare and that they are resolved quickly
   Duration RandomTimeout(Duration min, Duration max) {
-    std::uniform_int_distribution time_distrib(min, max);
-    return io_.Rand(time_distrib);
+    auto min_micros = std::chrono::duration_cast<std::chrono::milliseconds>(min).count();
+    auto max_micros = std::chrono::duration_cast<std::chrono::milliseconds>(max).count();
+
+    std::uniform_int_distribution time_distrib(min_micros, max_micros);
+
+    auto rand_micros = io_.Rand(time_distrib);
+
+    return std::chrono::microseconds{rand_micros};
+  }
+
+  Duration RandomTimeout(int min_micros, int max_micros) {
+    std::uniform_int_distribution time_distrib(min_micros, max_micros);
+
+    int rand_micros = io_.Rand(time_distrib);
+
+    return std::chrono::microseconds{rand_micros};
   }
 
   Term PreviousTermFromIndex(LogIndex index) {
@@ -366,9 +380,11 @@ class Raft {
   template <typename... Ts>
   void Log(Ts &&...args) {
     const Time now = io_.Now();
+    auto micros = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
+
     const Term term = state_.term;
 
-    std::cout << '\t' << now << "\t" << term << "\t" << io_.GetAddress().last_known_port;
+    std::cout << '\t' << micros << "\t" << term << "\t" << io_.GetAddress().last_known_port;
 
     std::visit([&](auto &&role) { role.Print(); }, role_);
 
@@ -404,10 +420,11 @@ class Raft {
   std::optional<Role> Cron(Candidate &candidate) {
     const auto now = io_.Now();
     const Duration election_timeout = RandomTimeout(100000, 200000);
+    auto election_timeout_us = std::chrono::duration_cast<std::chrono::milliseconds>(election_timeout).count();
 
     if (now - candidate.election_began > election_timeout) {
       state_.term++;
-      Log("becoming Candidate for term ", state_.term, " after leader timeout of ", election_timeout,
+      Log("becoming Candidate for term ", state_.term, " after leader timeout of ", election_timeout_us,
           " elapsed since last election attempt");
 
       const VoteRequest request{

src/io/simulator/CMakeLists.txt

@@ -1,8 +1,5 @@
 set(io_simulator_sources
-    simulator.hpp
-    simulator_handle.hpp
-    simulator_stats.hpp
-    simulator_config.hpp)
+    simulator_handle.cpp)
 
 find_package(fmt REQUIRED)
 find_package(Threads REQUIRED)

src/io/simulator/message_conversion.hpp

@@ -0,0 +1,172 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+#include "io/transport.hpp"
+
+namespace memgraph::io::simulator {
+
+using memgraph::io::Duration;
+using memgraph::io::Message;
+using memgraph::io::Time;
+
+struct OpaqueMessage {
+  Address from_address;
+  uint64_t request_id;
+  std::any message;
+
+  /// Recursively tries to match a specific type from the outer
+  /// variant's parameter pack against the type of the std::any,
+  /// and if it matches, make it concrete and return it. Otherwise,
+  /// move on and compare the any with the next type from the
+  /// parameter pack.
+  ///
+  /// Return is the full std::variant<Ts...> type that holds the
+  /// full parameter pack without interfering with recursive
+  /// narrowing expansion.
+  template <typename Return, Message Head, Message... Rest>
+  std::optional<Return> Unpack(std::any &&a) {
+    if (typeid(Head) == a.type()) {
+      Head concrete = std::any_cast<Head>(std::move(a));
+      return concrete;
+    }
+
+    if constexpr (sizeof...(Rest) > 0) {
+      return Unpack<Return, Rest...>(std::move(a));
+    } else {
+      return std::nullopt;
+    }
+  }
+
+  /// High level "user-facing" conversion function that lets
+  /// people interested in conversion only supply a single
+  /// parameter pack for the types that they want to compare
+  /// with the any and potentially include in the returned
+  /// variant.
+  template <Message... Ms>
+  requires(sizeof...(Ms) > 0) std::optional<std::variant<Ms...>> VariantFromAny(std::any &&a) {
+    return Unpack<std::variant<Ms...>, Ms...>(std::move(a));
+  }
+
+  template <Message... Ms>
+  requires(sizeof...(Ms) > 0) std::optional<RequestEnvelope<Ms...>> Take() && {
+    std::optional<std::variant<Ms...>> m_opt = VariantFromAny<Ms...>(std::move(message));
+
+    if (m_opt) {
+      return RequestEnvelope<Ms...>{
+          .message = std::move(*m_opt),
+          .request_id = request_id,
+          .from_address = from_address,
+      };
+    }
+
+    return std::nullopt;
+  }
+};
+
+class OpaquePromiseTraitBase {
+ public:
+  virtual const std::type_info *TypeInfo() const = 0;
+  virtual bool IsAwaited(void *ptr) const = 0;
+  virtual void Fill(void *ptr, OpaqueMessage &&) const = 0;
+  virtual void TimeOut(void *ptr) const = 0;
+
+  virtual ~OpaquePromiseTraitBase() = default;
+  OpaquePromiseTraitBase() = default;
+  OpaquePromiseTraitBase(const OpaquePromiseTraitBase &) = delete;
+  OpaquePromiseTraitBase &operator=(const OpaquePromiseTraitBase &) = delete;
+  OpaquePromiseTraitBase(OpaquePromiseTraitBase &&old) = delete;
+  OpaquePromiseTraitBase &operator=(OpaquePromiseTraitBase &&) = delete;
+};
+
+template <typename T>
+class OpaquePromiseTrait : public OpaquePromiseTraitBase {
+ public:
+  const std::type_info *TypeInfo() const override { return &typeid(T); };
+
+  bool IsAwaited(void *ptr) const override { return static_cast<ResponsePromise<T> *>(ptr)->IsAwaited(); };
+
+  void Fill(void *ptr, OpaqueMessage &&opaque_message) const override {
+    T message = std::any_cast<T>(std::move(opaque_message.message));
+    auto response_envelope = ResponseEnvelope<T>{.message = std::move(message),
+                                                 .request_id = opaque_message.request_id,
+                                                 .from_address = opaque_message.from_address};
+    auto promise = static_cast<ResponsePromise<T> *>(ptr);
+    auto unique_promise = std::unique_ptr<ResponsePromise<T>>(promise);
+    unique_promise->Fill(std::move(response_envelope));
+  };
+
+  void TimeOut(void *ptr) const override {
+    auto promise = static_cast<ResponsePromise<T> *>(ptr);
+    auto unique_promise = std::unique_ptr<ResponsePromise<T>>(promise);
+    ResponseResult<T> result = TimedOut{};
+    unique_promise->Fill(std::move(result));
+  }
+};
+
+class OpaquePromise {
+  void *ptr_;
+  std::unique_ptr<OpaquePromiseTraitBase> trait_;
+
+ public:
+  OpaquePromise(OpaquePromise &&old) noexcept : ptr_(old.ptr_), trait_(std::move(old.trait_)) { old.ptr_ = nullptr; }
+
+  OpaquePromise &operator=(OpaquePromise &&old) noexcept {
+    MG_ASSERT(this != &old);
+    ptr_ = old.ptr_;
+    trait_ = std::move(old.trait_);
+    old.ptr_ = nullptr;
+    return *this;
+  }
+
+  OpaquePromise(const OpaquePromise &) = delete;
+  OpaquePromise &operator=(const OpaquePromise &) = delete;
+
+  template <typename T>
+  std::unique_ptr<ResponsePromise<T>> Take() && {
+    MG_ASSERT(typeid(T) == *trait_->TypeInfo());
+    MG_ASSERT(ptr_ != nullptr);
+
+    auto ptr = static_cast<ResponsePromise<T> *>(ptr_);
+
+    ptr_ = nullptr;
+
+    return std::unique_ptr<T>(ptr);
+  }
+
+  template <typename T>
+  explicit OpaquePromise(std::unique_ptr<ResponsePromise<T>> promise)
+      : ptr_(static_cast<void *>(promise.release())), trait_(std::make_unique<OpaquePromiseTrait<T>>()) {}
+
+  bool IsAwaited() {
+    MG_ASSERT(ptr_ != nullptr);
+    return trait_->IsAwaited(ptr_);
+  }
+
+  void TimeOut() {
+    MG_ASSERT(ptr_ != nullptr);
+    trait_->TimeOut(ptr_);
+    ptr_ = nullptr;
+  }
+
+  void Fill(OpaqueMessage &&opaque_message) {
+    MG_ASSERT(ptr_ != nullptr);
+    trait_->Fill(ptr_, std::move(opaque_message));
+    ptr_ = nullptr;
+  }
+
+  ~OpaquePromise() {
+    MG_ASSERT(ptr_ == nullptr, "OpaquePromise destroyed without being explicitly timed out or filled");
+  }
+};
+
+}  // namespace memgraph::io::simulator

src/io/simulator/simulator.hpp

@@ -39,7 +39,7 @@ class Simulator {
   Io<SimulatorTransport> Register(Address address) {
     std::uniform_int_distribution<uint64_t> seed_distrib;
     uint64_t seed = seed_distrib(rng_);
-    return Io(SimulatorTransport(simulator_handle_, address, seed), address);
+    return Io{SimulatorTransport{simulator_handle_, address, seed}, address};
   }
 
   void IncrementServerCountAndWaitForQuiescentState(Address address) {

src/io/simulator/simulator_handle.cpp

@@ -0,0 +1,156 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include "io/simulator/simulator_handle.hpp"
+#include "io/address.hpp"
+#include "io/errors.hpp"
+#include "io/simulator/simulator_config.hpp"
+#include "io/simulator/simulator_stats.hpp"
+#include "io/time.hpp"
+#include "io/transport.hpp"
+
+namespace memgraph::io::simulator {
+
+using memgraph::io::Duration;
+using memgraph::io::Time;
+
+void SimulatorHandle::ShutDown() {
+  std::unique_lock<std::mutex> lock(mu_);
+  should_shut_down_ = true;
+  cv_.notify_all();
+}
+
+bool SimulatorHandle::ShouldShutDown() const {
+  std::unique_lock<std::mutex> lock(mu_);
+  return should_shut_down_;
+}
+
+void SimulatorHandle::IncrementServerCountAndWaitForQuiescentState(Address address) {
+  std::unique_lock<std::mutex> lock(mu_);
+  server_addresses_.insert(address);
+
+  while (true) {
+    const size_t blocked_servers = BlockedServers();
+
+    const bool all_servers_blocked = blocked_servers == server_addresses_.size();
+
+    if (all_servers_blocked) {
+      return;
+    }
+
+    cv_.wait(lock);
+  }
+}
+
+size_t SimulatorHandle::BlockedServers() {
+  size_t blocked_servers = blocked_on_receive_;
+
+  for (auto &[promise_key, opaque_promise] : promises_) {
+    if (opaque_promise.promise.IsAwaited()) {
+      if (server_addresses_.contains(promise_key.requester_address)) {
+        blocked_servers++;
+      }
+    }
+  }
+
+  return blocked_servers;
+}
+
+bool SimulatorHandle::MaybeTickSimulator() {
+  std::unique_lock<std::mutex> lock(mu_);
+
+  const size_t blocked_servers = BlockedServers();
+
+  if (blocked_servers < server_addresses_.size()) {
+    // we only need to advance the simulator when all
+    // servers have reached a quiescent state, blocked
+    // on their own futures or receive methods.
+    return false;
+  }
+
+  stats_.simulator_ticks++;
+
+  cv_.notify_all();
+
+  TimeoutPromisesPastDeadline();
+
+  if (in_flight_.empty()) {
+    // return early here because there are no messages to schedule
+
+    // We tick the clock forward when all servers are blocked but
+    // there are no in-flight messages to schedule delivery of.
+    std::poisson_distribution<> time_distrib(50);
+    Duration clock_advance = std::chrono::microseconds{time_distrib(rng_)};
+    cluster_wide_time_microseconds_ += clock_advance;
+
+    MG_ASSERT(cluster_wide_time_microseconds_ < config_.abort_time,
+              "Cluster has executed beyond its configured abort_time, and something may be failing to make progress "
+              "in an expected amount of time.");
+    return true;
+  }
+
+  if (config_.scramble_messages) {
+    // scramble messages
+    std::uniform_int_distribution<size_t> swap_distrib(0, in_flight_.size() - 1);
+    const size_t swap_index = swap_distrib(rng_);
+    std::swap(in_flight_[swap_index], in_flight_.back());
+  }
+
+  auto [to_address, opaque_message] = std::move(in_flight_.back());
+  in_flight_.pop_back();
+
+  std::uniform_int_distribution<int> drop_distrib(0, 99);
+  const int drop_threshold = drop_distrib(rng_);
+  const bool should_drop = drop_threshold < config_.drop_percent;
+
+  if (should_drop) {
+    stats_.dropped_messages++;
+  }
+
+  PromiseKey promise_key{.requester_address = to_address,
+                         .request_id = opaque_message.request_id,
+                         .replier_address = opaque_message.from_address};
+
+  if (promises_.contains(promise_key)) {
+    // complete waiting promise if it's there
+    DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key));
+    promises_.erase(promise_key);
+
+    const bool normal_timeout = config_.perform_timeouts && (dop.deadline < cluster_wide_time_microseconds_);
+
+    if (should_drop || normal_timeout) {
+      stats_.timed_out_requests++;
+      dop.promise.TimeOut();
+    } else {
+      stats_.total_responses++;
+      dop.promise.Fill(std::move(opaque_message));
+    }
+  } else if (should_drop) {
+    // don't add it anywhere, let it drop
+  } else {
+    // add to can_receive_ if not
+    const auto &[om_vec, inserted] = can_receive_.try_emplace(to_address, std::vector<OpaqueMessage>());
+    om_vec->second.emplace_back(std::move(opaque_message));
+  }
+
+  return true;
+}
+
+Time SimulatorHandle::Now() const {
+  std::unique_lock<std::mutex> lock(mu_);
+  return cluster_wide_time_microseconds_;
+}
+
+SimulatorStats SimulatorHandle::Stats() {
+  std::unique_lock<std::mutex> lock(mu_);
+  return stats_;
+}
+}  // namespace memgraph::io::simulator

src/io/simulator/simulator_handle.hpp

@@ -11,9 +11,8 @@
 
 #pragma once
 
-#include <compare>
-
 #include <any>
+#include <compare>
 #include <iostream>
 #include <map>
 #include <memory>
@@ -25,6 +24,7 @@
 
 #include "io/address.hpp"
 #include "io/errors.hpp"
+#include "io/simulator/message_conversion.hpp"
 #include "io/simulator/simulator_config.hpp"
 #include "io/simulator/simulator_stats.hpp"
 #include "io/time.hpp"
@@ -35,167 +35,24 @@ namespace memgraph::io::simulator {
 using memgraph::io::Duration;
 using memgraph::io::Time;
 
-struct OpaqueMessage {
-  Address from_address;
-  uint64_t request_id;
-  std::any message;
-
-  /// Recursively tries to match a specific type from the outer
-  /// variant's parameter pack against the type of the std::any,
-  /// and if it matches, make it concrete and return it. Otherwise,
-  /// move on and compare the any with the next type from the
-  /// parameter pack.
-  ///
-  /// Return is the full std::variant<Ts...> type that holds the
-  /// full parameter pack without interfering with recursive
-  /// narrowing expansion.
-  template <typename Return, Message Head, Message... Rest>
-  std::optional<Return> Unpack(std::any &&a) {
-    if (typeid(Head) == a.type()) {
-      Head concrete = std::any_cast<Head>(std::move(a));
-      return concrete;
-    }
-
-    if constexpr (sizeof...(Rest) > 0) {
-      return Unpack<Return, Rest...>(std::move(a));
-    } else {
-      return std::nullopt;
-    }
-  }
-
-  /// High level "user-facing" conversion function that lets
-  /// people interested in conversion only supply a single
-  /// parameter pack for the types that they want to compare
-  /// with the any and potentially include in the returned
-  /// variant.
-  template <Message... Ms>
-  requires(sizeof...(Ms) > 0) std::optional<std::variant<Ms...>> VariantFromAny(std::any &&a) {
-    return Unpack<std::variant<Ms...>, Ms...>(std::move(a));
-  }
-
-  template <Message... Ms>
-  requires(sizeof...(Ms) > 0) std::optional<RequestEnvelope<Ms...>> Take() {
-    std::optional<std::variant<Ms...>> m_opt = VariantFromAny<Ms...>(std::move(message));
-
-    if (m_opt) {
-      return RequestEnvelope<Ms...>{
-          .message = std::move(*m_opt),
-          .request_id = request_id,
-          .from_address = from_address,
-      };
-    } else {
-      return std::nullopt;
-    }
-  }
-};
-
 struct PromiseKey {
   Address requester_address;
   uint64_t request_id;
+  // TODO(tyler) possibly remove replier_address from promise key
+  // once we want to support DSR.
   Address replier_address;
 
  public:
   bool operator<(const PromiseKey &other) const {
-    if (requester_address == other.requester_address) {
-      return request_id < other.request_id;
-    } else {
+    if (requester_address != other.requester_address) {
       return requester_address < other.requester_address;
     }
-  }
-};
 
-class OpaquePromise {
-  const std::type_info *ti_;
-  void *ptr_;
-  std::function<void(void *)> dtor_;
-  std::function<bool(void *)> is_awaited_;
-  std::function<void(void *, OpaqueMessage)> fill_;
-  std::function<void(void *)> time_out_;
-
- public:
-  OpaquePromise(OpaquePromise &&old)
-      : ti_(old.ti_),
-        ptr_(old.ptr_),
-        dtor_(old.dtor_),
-        is_awaited_(old.is_awaited_),
-        fill_(old.fill_),
-        time_out_(old.time_out_) {
-    old.ptr_ = nullptr;
-  }
-
-  OpaquePromise &operator=(OpaquePromise &&old) {
-    MG_ASSERT(this != &old);
-
-    ptr_ = old.ptr_;
-    ti_ = old.ti_;
-    dtor_ = old.dtor_;
-    is_awaited_ = old.is_awaited_;
-    fill_ = old.fill_;
-    time_out_ = old.time_out_;
-
-    old.ptr_ = nullptr;
-
-    return *this;
-  }
-
-  OpaquePromise(const OpaquePromise &) = delete;
-  OpaquePromise &operator=(const OpaquePromise &) = delete;
-
-  template <typename T>
-  std::unique_ptr<ResponsePromise<T>> Take() {
-    MG_ASSERT(typeid(T) == *ti_);
-    MG_ASSERT(ptr_ != nullptr);
-
-    ResponsePromise<T> *ptr = static_cast<ResponsePromise<T> *>(ptr_);
-
-    ptr_ = nullptr;
-
-    return std::unique_ptr<T>(ptr);
-  }
-
-  template <typename T>
-  explicit OpaquePromise(std::unique_ptr<ResponsePromise<T>> promise)
-      : ti_(&typeid(T)),
-        ptr_(static_cast<void *>(promise.release())),
-        dtor_([](void *ptr) { static_cast<ResponsePromise<T> *>(ptr)->~ResponsePromise<T>(); }),
-        is_awaited_([](void *ptr) { return static_cast<ResponsePromise<T> *>(ptr)->IsAwaited(); }),
-        fill_([](void *this_ptr, OpaqueMessage opaque_message) {
-          T message = std::any_cast<T>(std::move(opaque_message.message));
-          auto response_envelope = ResponseEnvelope<T>{.message = std::move(message),
-                                                       .request_id = opaque_message.request_id,
-                                                       .from_address = opaque_message.from_address};
-          ResponsePromise<T> *promise = static_cast<ResponsePromise<T> *>(this_ptr);
-          auto unique_promise = std::unique_ptr<ResponsePromise<T>>(promise);
-          unique_promise->Fill(std::move(response_envelope));
-        }),
-        time_out_([](void *ptr) {
-          ResponsePromise<T> *promise = static_cast<ResponsePromise<T> *>(ptr);
-          auto unique_promise = std::unique_ptr<ResponsePromise<T>>(promise);
-          ResponseResult<T> result = TimedOut{};
-          unique_promise->Fill(std::move(result));
-        }) {}
-
-  bool IsAwaited() {
-    MG_ASSERT(ptr_ != nullptr);
-    return is_awaited_(ptr_);
-  }
-
-  void TimeOut() {
-    MG_ASSERT(ptr_ != nullptr);
-    time_out_(ptr_);
-    ptr_ = nullptr;
-  }
-
-  void Fill(OpaqueMessage &&opaque_message) {
-    MG_ASSERT(ptr_ != nullptr);
-    fill_(ptr_, std::move(opaque_message));
-    ptr_ = nullptr;
-  }
-
-  ~OpaquePromise() {
-    if (nullptr != ptr_) {
-      dtor_(ptr_);
+    if (request_id != other.request_id) {
+      return request_id < other.request_id;
     }
+
+    return replier_address < other.replier_address;
   }
 };
 
@@ -205,8 +62,8 @@ struct DeadlineAndOpaquePromise {
 };
 
 class SimulatorHandle {
-  std::mutex mu_{};
-  std::condition_variable cv_;
+  mutable std::mutex mu_{};
+  mutable std::condition_variable cv_;
 
   // messages that have not yet been scheduled or dropped
   std::vector<std::pair<Address, OpaqueMessage>> in_flight_;
@@ -225,150 +82,44 @@ class SimulatorHandle {
   std::mt19937 rng_;
   SimulatorConfig config_;
 
- public:
-  explicit SimulatorHandle(SimulatorConfig config)
-      : cluster_wide_time_microseconds_(config.start_time), rng_(config.rng_seed), config_(config) {}
-
-  void IncrementServerCountAndWaitForQuiescentState(Address address) {
-    std::unique_lock<std::mutex> lock(mu_);
-    server_addresses_.insert(address);
-
-    while (true) {
-      size_t blocked_servers = blocked_on_receive_;
-
-      for (auto &[promise_key, opaque_promise] : promises_) {
-        if (opaque_promise.promise.IsAwaited()) {
-          if (server_addresses_.contains(promise_key.requester_address)) {
-            blocked_servers++;
-          }
-        }
-      }
-
-      bool all_servers_blocked = blocked_servers == server_addresses_.size();
-
-      if (all_servers_blocked) {
-        return;
-      }
-
-      cv_.wait(lock);
-    }
-  }
+  /// Returns the number of servers currently blocked on Receive, plus
+  /// the servers that are blocked on Futures that were created through
+  /// SimulatorTransport::Request.
+  ///
+  /// TODO(tyler) investigate whether avoiding consideration of Futures
+  /// increases determinism.
+  size_t BlockedServers();
 
   void TimeoutPromisesPastDeadline() {
     const Time now = cluster_wide_time_microseconds_;
 
     for (auto &[promise_key, dop] : promises_) {
-      // TODO(tyler) queue this up and drop it after its deadline
       if (dop.deadline < now) {
-        std::cout << "timing out request" << std::endl;
-        DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key));
+        spdlog::debug("timing out request from requester {} to replier {}.", promise_key.requester_address.ToString(),
+                      promise_key.replier_address.ToString());
+        std::move(dop).promise.TimeOut();
         promises_.erase(promise_key);
 
         stats_.timed_out_requests++;
-
-        dop.promise.TimeOut();
       }
     }
   }
 
-  bool MaybeTickSimulator() {
-    std::unique_lock<std::mutex> lock(mu_);
+ public:
+  explicit SimulatorHandle(SimulatorConfig config)
+      : cluster_wide_time_microseconds_(config.start_time), rng_(config.rng_seed), config_(config) {}
 
-    size_t blocked_servers = blocked_on_receive_;
+  void IncrementServerCountAndWaitForQuiescentState(Address address);
 
-    for (auto &[promise_key, opaque_promise] : promises_) {
-      if (opaque_promise.promise.IsAwaited()) {
-        if (server_addresses_.contains(promise_key.requester_address)) {
-          blocked_servers++;
-        }
-      }
-    }
+  /// This method causes most of the interesting simulation logic to happen, wrt network behavior.
+  /// It checks to see if all background "server" threads are blocked on new messages, and if so,
+  /// it will decide whether to drop, reorder, or deliver in-flight messages based on the SimulatorConfig
+  /// that was used to create the Simulator.
+  bool MaybeTickSimulator();
 
-    if (blocked_servers < server_addresses_.size()) {
-      // we only need to advance the simulator when all
-      // servers have reached a quiescent state, blocked
-      // on their own futures or receive methods.
-      return false;
-    }
+  void ShutDown();
 
-    stats_.simulator_ticks++;
-
-    cv_.notify_all();
-
-    TimeoutPromisesPastDeadline();
-
-    if (in_flight_.empty()) {
-      // return early here because there are no messages to schedule
-
-      // We tick the clock forward when all servers are blocked but
-      // there are no in-flight messages to schedule delivery of.
-      std::poisson_distribution<> time_distrib(50);
-      Duration clock_advance = std::chrono::microseconds{time_distrib(rng_)};
-      cluster_wide_time_microseconds_ += clock_advance;
-
-      MG_ASSERT(cluster_wide_time_microseconds_ < config_.abort_time,
-                "Cluster has executed beyond its configured abort_time, and something may be failing to make progress "
-                "in an expected amount of time.");
-      return true;
-    }
-
-    if (config_.scramble_messages) {
-      // scramble messages
-      std::uniform_int_distribution<size_t> swap_distrib(0, in_flight_.size() - 1);
-      size_t swap_index = swap_distrib(rng_);
-      std::swap(in_flight_[swap_index], in_flight_.back());
-    }
-
-    auto [to_address, opaque_message] = std::move(in_flight_.back());
-    in_flight_.pop_back();
-
-    std::uniform_int_distribution<int> drop_distrib(0, 99);
-    int drop_threshold = drop_distrib(rng_);
-    bool should_drop = drop_threshold < config_.drop_percent;
-
-    if (should_drop) {
-      stats_.dropped_messages++;
-    }
-
-    PromiseKey promise_key{.requester_address = to_address,
-                           .request_id = opaque_message.request_id,
-                           .replier_address = opaque_message.from_address};
-
-    if (promises_.contains(promise_key)) {
-      // complete waiting promise if it's there
-      DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key));
-      promises_.erase(promise_key);
-
-      const bool normal_timeout = config_.perform_timeouts && (dop.deadline < cluster_wide_time_microseconds_);
-
-      if (should_drop || normal_timeout) {
-        stats_.timed_out_requests++;
-        dop.promise.TimeOut();
-      } else {
-        stats_.total_responses++;
-        dop.promise.Fill(std::move(opaque_message));
-      }
-    } else if (should_drop) {
-      // don't add it anywhere, let it drop
-    } else {
-      // add to can_receive_ if not
-      const auto &[om_vec, inserted] = can_receive_.try_emplace(to_address, std::vector<OpaqueMessage>());
-      om_vec->second.emplace_back(std::move(opaque_message));
-    }
-
-    return true;
-  }
-
-  void ShutDown() {
-    std::unique_lock<std::mutex> lock(mu_);
-    should_shut_down_ = true;
-    cv_.notify_all();
-  }
-
-  bool ShouldShutDown() {
-    std::unique_lock<std::mutex> lock(mu_);
-    return should_shut_down_;
-  }
+  bool ShouldShutDown() const;
 
   template <Message Request, Message Response>
   void SubmitRequest(Address to_address, Address from_address, uint64_t request_id, Request &&request, Duration timeout,
@@ -377,9 +128,9 @@ class SimulatorHandle {
 
     const Time deadline = cluster_wide_time_microseconds_ + timeout;
 
-    std::any message(std::move(request));
+    std::any message(request);
     OpaqueMessage om{.from_address = from_address, .request_id = request_id, .message = std::move(message)};
-    in_flight_.emplace_back(std::make_pair(std::move(to_address), std::move(om)));
+    in_flight_.emplace_back(std::make_pair(to_address, std::move(om)));
 
     PromiseKey promise_key{.requester_address = from_address, .request_id = request_id, .replier_address = to_address};
     OpaquePromise opaque_promise(std::move(promise).ToUnique());
@@ -390,8 +141,6 @@ class SimulatorHandle {
     stats_.total_requests++;
 
     cv_.notify_all();
-
-    return;
   }
 
   template <Message... Ms>
@@ -411,7 +160,7 @@ class SimulatorHandle {
 
           // TODO(tyler) search for item in can_receive_ that matches the desired types, rather
           // than asserting that the last item in can_rx matches.
-          auto m_opt = message.Take<Ms...>();
+          auto m_opt = std::move(message).Take<Ms...>();
 
           blocked_on_receive_ -= 1;
 
@@ -444,10 +193,7 @@ class SimulatorHandle {
     cv_.notify_all();
   }
 
-  Time Now() {
-    std::unique_lock<std::mutex> lock(mu_);
-    return cluster_wide_time_microseconds_;
-  }
+  Time Now() const;
 
   template <class D = std::poisson_distribution<>, class Return = uint64_t>
   Return Rand(D distrib) {
@@ -455,9 +201,6 @@ class SimulatorHandle {
     return distrib(rng_);
   }
 
-  SimulatorStats Stats() {
-    std::unique_lock<std::mutex> lock(mu_);
-    return stats_;
-  }
+  SimulatorStats Stats();
 };
 };  // namespace memgraph::io::simulator

src/io/simulator/simulator_stats.hpp

@@ -11,6 +11,8 @@
 
 #pragma once
 
+#include <cstdint>
+
 namespace memgraph::io::simulator {
 struct SimulatorStats {
   uint64_t total_messages = 0;

src/io/simulator/simulator_transport.hpp

@@ -25,8 +25,8 @@ using memgraph::io::Time;
 
 class SimulatorTransport {
   std::shared_ptr<SimulatorHandle> simulator_handle_;
-  Address address_;
-  std::mt19937 rng_{};
+  const Address address_;
+  std::mt19937 rng_;
 
  public:
   SimulatorTransport(std::shared_ptr<SimulatorHandle> simulator_handle, Address address, uint64_t seed)
@@ -53,9 +53,9 @@ class SimulatorTransport {
     return simulator_handle_->template Send<M>(address, address_, request_id, message);
   }
 
-  Time Now() { return simulator_handle_->Now(); }
+  Time Now() const { return simulator_handle_->Now(); }
 
-  bool ShouldShutDown() { return simulator_handle_->ShouldShutDown(); }
+  bool ShouldShutDown() const { return simulator_handle_->ShouldShutDown(); }
 
   template <class D = std::poisson_distribution<>, class Return = uint64_t>
   Return Rand(D distrib) {

src/io/time.hpp

@@ -15,7 +15,7 @@
 
 namespace memgraph::io {
 
-using Duration = std::chrono::duration<int64_t, std::ratio<1, 1000000>>;
+using Duration = std::chrono::microseconds;
 using Time = std::chrono::time_point<std::chrono::local_t, Duration>;
 
 }  // namespace memgraph::io

src/io/transport.hpp

@@ -22,9 +22,10 @@
 #include "io/time.hpp"
 #include "utils/result.hpp"
 
+namespace memgraph::io {
+
 using memgraph::utils::BasicResult;
 
-namespace memgraph::io {
 // TODO(tyler) ensure that Message continues to represent
 // reasonable constraints around message types over time,
 // as we adapt things to use Thrift-generated message types.
@@ -71,51 +72,52 @@ class Io {
   /// without an explicit timeout set.
   void SetDefaultTimeout(Duration timeout) { default_timeout_ = timeout; }
 
-  /// Issue a request with an explicit timeout in microseconds provided.
+  /// Issue a request with an explicit timeout in microseconds provided. This tends to be used by clients.
   template <Message Request, Message Response>
   ResponseFuture<Response> RequestWithTimeout(Address address, Request request, Duration timeout) {
-    uint64_t request_id = ++request_id_counter_;
+    const uint64_t request_id = ++request_id_counter_;
     return implementation_.template Request<Request, Response>(address, request_id, request, timeout);
   }
 
-  /// Issue a request that times out after the default timeout.
+  /// Issue a request that times out after the default timeout. This tends
+  /// to be used by clients.
   template <Message Request, Message Response>
   ResponseFuture<Response> Request(Address address, Request request) {
-    uint64_t request_id = ++request_id_counter_;
-    Duration timeout = default_timeout_;
+    const uint64_t request_id = ++request_id_counter_;
+    const Duration timeout = default_timeout_;
     return implementation_.template Request<Request, Response>(address, request_id, std::move(request), timeout);
   }
 
   /// Wait for an explicit number of microseconds for a request of one of the
-  /// provided types to arrive.
+  /// provided types to arrive. This tends to be used by servers.
   template <Message... Ms>
   RequestResult<Ms...> ReceiveWithTimeout(Duration timeout) {
     return implementation_.template Receive<Ms...>(timeout);
   }
 
   /// Wait the default number of microseconds for a request of one of the
-  /// provided types to arrive.
+  /// provided types to arrive. This tends to be used by servers.
   template <Message... Ms>
   requires(sizeof...(Ms) > 0) RequestResult<Ms...> Receive() {
-    Duration timeout = default_timeout_;
+    const Duration timeout = default_timeout_;
     return implementation_.template Receive<Ms...>(timeout);
   }
 
-  /// Send a message in a best-effort fashion. If you need reliable delivery,
-  /// this must be built on-top. TCP is not enough for most use cases.
+  /// Send a message in a best-effort fashion. This is used for messaging where
+  /// responses are not necessarily expected, and for servers to respond to requests.
+  /// If you need reliable delivery, this must be built on-top. TCP is not enough for most use cases.
   template <Message M>
   void Send(Address address, uint64_t request_id, M message) {
     return implementation_.template Send<M>(address, request_id, std::move(message));
   }
 
-  /// The current system time in microseconds since the unix epoch.
-  /// This time source should be preferred over any other, because it
-  /// lets us deterministically control clocks from tests for making
+  /// The current system time. This time source should be preferred over any other,
+  /// because it lets us deterministically control clocks from tests for making
   /// things like timeouts deterministic.
-  Time Now() { return implementation_.Now(); }
+  Time Now() const { return implementation_.Now(); }
 
-  /// Returns true of the system should shut-down.
-  bool ShouldShutDown() { return implementation_.ShouldShutDown(); }
+  /// Returns true if the system should shut-down.
+  bool ShouldShutDown() const { return implementation_.ShouldShutDown(); }
 
   /// Returns a random number within the specified distribution.
   template <class D = std::poisson_distribution<>, class Return = uint64_t>

tests/benchmark/CMakeLists.txt

@@ -62,3 +62,6 @@ target_link_libraries(${test_prefix}storage_v2_gc mg-storage-v2)
 
 add_benchmark(storage_v2_property_store.cpp)
 target_link_libraries(${test_prefix}storage_v2_property_store mg-storage-v2)
+
+add_benchmark(future.cpp)
+target_link_libraries(${test_prefix}future mg-io)

tests/benchmark/future.cpp

@@ -0,0 +1,30 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include <benchmark/benchmark.h>
+
+#include "io/future.hpp"
+
+static void FuturePairFillWait(benchmark::State &state) {
+  uint64_t counter = 0;
+  while (state.KeepRunning()) {
+    auto [future, promise] = memgraph::io::FuturePromisePair<int>();
+    promise.Fill(1);
+    std::move(future).Wait();
+
+    ++counter;
+  }
+  state.SetItemsProcessed(counter);
+}
+
+BENCHMARK(FuturePairFillWait)->Unit(benchmark::kNanosecond)->UseRealTime();
+
+BENCHMARK_MAIN();

tests/simulation/CMakeLists.txt

@@ -25,10 +25,10 @@ function(add_simulation_test test_cpp san)
   add_dependencies(memgraph__simulation ${target_name})
 endfunction(add_simulation_test)
 
-add_simulation_test(future.cpp thread)
-
 add_simulation_test(basic_request.cpp address)
 
+add_simulation_test(raft.cpp address)
+
 add_simulation_test(trial_query_storage/query_storage_test.cpp address)
 
 add_simulation_test(sharded_map.cpp address)

tests/simulation/basic_request.cpp

@@ -30,7 +30,7 @@ struct CounterResponse {
 };
 
 void run_server(Io<SimulatorTransport> io) {
-  uint64_t highest_seen;
+  uint64_t highest_seen = 0;
 
   while (!io.ShouldShutDown()) {
     std::cout << "[SERVER] Is receiving..." << std::endl;

tests/simulation/raft.cpp

@@ -9,6 +9,7 @@
 // by the Apache License, Version 2.0, included in the file
 // licenses/APL.txt.
 
+#include <chrono>
 #include <deque>
 #include <iostream>
 #include <map>
@@ -23,10 +24,12 @@
 #include "io/simulator/simulator_transport.hpp"
 
 using memgraph::io::Address;
+using memgraph::io::Duration;
 using memgraph::io::Io;
 using memgraph::io::ResponseEnvelope;
 using memgraph::io::ResponseFuture;
 using memgraph::io::ResponseResult;
+using memgraph::io::Time;
 using memgraph::io::rsm::Raft;
 using memgraph::io::rsm::ReadRequest;
 using memgraph::io::rsm::ReadResponse;
@@ -124,8 +127,8 @@ void RunSimulation() {
       .perform_timeouts = true,
       .scramble_messages = true,
       .rng_seed = 0,
-      .start_time = 256 * 1024,
-      .abort_time = 8 * 1024 * 1024,
+      .start_time = Time::min() + std::chrono::microseconds{256 * 1024},
+      .abort_time = Time::min() + std::chrono::microseconds{8 * 1024 * 1024},
   };
 
   auto simulator = Simulator(config);
@@ -184,10 +187,10 @@ void RunSimulation() {
 
     std::cout << "client sending CasRequest to Leader " << leader.last_known_port << std::endl;
     ResponseFuture<WriteResponse<CasResponse>> cas_response_future =
-        cli_io.RequestWithTimeout<WriteRequest<CasRequest>, WriteResponse<CasResponse>>(leader, cli_req, 50000);
+        cli_io.Request<WriteRequest<CasRequest>, WriteResponse<CasResponse>>(leader, cli_req);
 
     // receive cas_response
-    ResponseResult<WriteResponse<CasResponse>> cas_response_result = cas_response_future.Wait();
+    ResponseResult<WriteResponse<CasResponse>> cas_response_result = std::move(cas_response_future).Wait();
 
     if (cas_response_result.HasError()) {
       std::cout << "client timed out while trying to communicate with leader server " << std::endl;
@@ -233,10 +236,10 @@ void RunSimulation() {
 
     std::cout << "client sending GetRequest to Leader " << leader.last_known_port << std::endl;
     ResponseFuture<ReadResponse<GetResponse>> get_response_future =
-        cli_io.RequestWithTimeout<ReadRequest<GetRequest>, ReadResponse<GetResponse>>(leader, read_req, 50000);
+        cli_io.Request<ReadRequest<GetRequest>, ReadResponse<GetResponse>>(leader, read_req);
 
     // receive response
-    ResponseResult<ReadResponse<GetResponse>> get_response_result = get_response_future.Wait();
+    ResponseResult<ReadResponse<GetResponse>> get_response_result = std::move(get_response_future).Wait();
 
     if (get_response_result.HasError()) {
       std::cout << "client timed out while trying to communicate with leader server " << std::endl;

tests/simulation/trial_query_storage/messages.hpp

@@ -15,8 +15,6 @@
 #include <string>
 #include <vector>
 
-// header
-
 namespace memgraph::tests::simulation {
 
 struct Vertex {

tests/simulation/trial_query_storage/query_storage_test.cpp

@@ -80,7 +80,6 @@ int main() {
 
   auto res_f = cli_io.Request<ScanVerticesRequest, VerticesResponse>(srv_addr, req);
   auto res_rez = std::move(res_f).Wait();
-  // MG_ASSERT(res_rez.HasError());
   simulator.ShutDown();
   return 0;
 }

tests/unit/CMakeLists.txt

@@ -365,3 +365,7 @@ target_link_libraries(${test_prefix}websocket mg-communication Boost::headers)
 # Test storage-v3
 add_unit_test(storage_v3.cpp)
 target_link_libraries(${test_prefix}storage_v3 mg-storage-v3)
+
+# Test future
+add_unit_test(future.cpp)
+target_link_libraries(${test_prefix}future mg-io)

tests/unit/future.cpp

@@ -12,8 +12,9 @@
 #include <string>
 #include <thread>
 
+#include "gtest/gtest.h"
+
 #include "io/future.hpp"
-#include "utils/logging.hpp"
 
 using namespace memgraph::io;
 
@@ -21,11 +22,11 @@ void Fill(Promise<std::string> promise_1) { promise_1.Fill("success"); }
 
 void Wait(Future<std::string> future_1, Promise<std::string> promise_2) {
   std::string result_1 = std::move(future_1).Wait();
-  MG_ASSERT(result_1 == "success");
+  EXPECT_TRUE(result_1 == "success");
   promise_2.Fill("it worked");
 }
 
-int main() {
+TEST(Future, BasicLifecycle) {
   std::atomic_bool waiting = false;
 
   std::function<bool()> notifier = [&] {
@@ -50,7 +51,5 @@ int main() {
   t2.join();
 
   std::string result_2 = std::move(future_2).Wait();
-  MG_ASSERT(result_2 == "it worked");
-
-  return 0;
+  EXPECT_TRUE(result_2 == "it worked");
 }