From 66ef5b20727e5a338f3bb2a1fc4b7279430b89bf Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 08:26:09 +0000 Subject: [PATCH 01/31] Make certain temporaries const --- src/io/transport.hpp | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/io/transport.hpp b/src/io/transport.hpp index 0a0531fd6..a7f6e99ff 100644 --- a/src/io/transport.hpp +++ b/src/io/transport.hpp @@ -74,15 +74,15 @@ class Io { /// Issue a request with an explicit timeout in microseconds provided. template ResponseFuture 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(address, request_id, request, timeout); } /// Issue a request that times out after the default timeout. template ResponseFuture 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(address, request_id, std::move(request), timeout); } @@ -97,7 +97,7 @@ class Io { /// provided types to arrive. template requires(sizeof...(Ms) > 0) RequestResult Receive() { - Duration timeout = default_timeout_; + const Duration timeout = default_timeout_; return implementation_.template Receive(timeout); } From b127f6f34594c6c28a069d6911c14cd766f3dd9b Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 10:23:25 +0000 Subject: [PATCH 02/31] Use spdlog and create a formatter for Address --- src/io/address.hpp | 7 +++++++ src/io/simulator/simulator_handle.hpp | 3 ++- src/io/transport.hpp | 2 +- 3 files changed, 10 insertions(+), 2 deletions(-) diff --git a/src/io/address.hpp b/src/io/address.hpp index 3506f0523..d649dd451 100644 --- a/src/io/address.hpp +++ b/src/io/address.hpp @@ -13,8 +13,10 @@ #include +#include #include #include +#include namespace memgraph::io { struct Address { @@ -52,5 +54,10 @@ struct Address { return unique_id < other.unique_id; } } + + 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 diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index 33757ee9b..d740fb151 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -260,7 +260,8 @@ class SimulatorHandle { 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; + spdlog::debug("timing out request from requester {} to replier {}.", promise_key.requester_address.ToString(), + promise_key.replier_address.ToString()); DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key)); promises_.erase(promise_key); diff --git a/src/io/transport.hpp b/src/io/transport.hpp index a7f6e99ff..4928b5f58 100644 --- a/src/io/transport.hpp +++ b/src/io/transport.hpp @@ -115,7 +115,7 @@ class Io { Time Now() { return implementation_.Now(); } /// Returns true of the system should shut-down. - bool ShouldShutDown() { return implementation_.ShouldShutDown(); } + const bool ShouldShutDown() { return implementation_.ShouldShutDown(); } /// Returns a random number within the specified distribution. template , class Return = uint64_t> From 9b915be1aa6c5722ccb4b6268402938a5a68a473 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 10:30:37 +0000 Subject: [PATCH 03/31] Make some methods const. Remove outdated TODO --- src/io/future.hpp | 4 ++-- src/io/simulator/simulator_handle.hpp | 5 ++--- src/io/simulator/simulator_transport.hpp | 6 +++--- src/io/transport.hpp | 4 ++-- 4 files changed, 9 insertions(+), 10 deletions(-) diff --git a/src/io/future.hpp b/src/io/future.hpp index 83503f28d..f935f4146 100644 --- a/src/io/future.hpp +++ b/src/io/future.hpp @@ -30,8 +30,8 @@ namespace memgraph::io { namespace { template class Shared { - std::condition_variable cv_; - std::mutex mu_; + mutable std::condition_variable cv_; + mutable std::mutex mu_; std::optional item_; bool consumed_ = false; bool waiting_ = false; diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index d740fb151..9819322d3 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -205,8 +205,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> in_flight_; @@ -258,7 +258,6 @@ class SimulatorHandle { 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) { spdlog::debug("timing out request from requester {} to replier {}.", promise_key.requester_address.ToString(), promise_key.replier_address.ToString()); diff --git a/src/io/simulator/simulator_transport.hpp b/src/io/simulator/simulator_transport.hpp index 6cc82ebeb..5f4c52a3d 100644 --- a/src/io/simulator/simulator_transport.hpp +++ b/src/io/simulator/simulator_transport.hpp @@ -25,7 +25,7 @@ using memgraph::io::Time; class SimulatorTransport { std::shared_ptr simulator_handle_; - Address address_; + const Address address_; std::mt19937 rng_{}; public: @@ -53,9 +53,9 @@ class SimulatorTransport { return simulator_handle_->template Send(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 Return = uint64_t> Return Rand(D distrib) { diff --git a/src/io/transport.hpp b/src/io/transport.hpp index 4928b5f58..7d079685e 100644 --- a/src/io/transport.hpp +++ b/src/io/transport.hpp @@ -112,10 +112,10 @@ class Io { /// 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. - const bool ShouldShutDown() { return implementation_.ShouldShutDown(); } + bool ShouldShutDown() const { return implementation_.ShouldShutDown(); } /// Returns a random number within the specified distribution. template , class Return = uint64_t> From c0d6cec9ab5863d58e36f003d859e4be890422ef Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 11:06:08 +0000 Subject: [PATCH 04/31] Split simulator_handle.cpp into a source and header file --- src/io/CMakeLists.txt | 6 +- src/io/simulator/CMakeLists.txt | 5 +- src/io/simulator/message_conversion.hpp | 171 +++++++++++++ src/io/simulator/simulator_handle.cpp | 175 ++++++++++++++ src/io/simulator/simulator_handle.hpp | 305 +----------------------- 5 files changed, 360 insertions(+), 302 deletions(-) create mode 100644 src/io/simulator/message_conversion.hpp create mode 100644 src/io/simulator/simulator_handle.cpp diff --git a/src/io/CMakeLists.txt b/src/io/CMakeLists.txt index 7ba068c8b..128e87114 100644 --- a/src/io/CMakeLists.txt +++ b/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) diff --git a/src/io/simulator/CMakeLists.txt b/src/io/simulator/CMakeLists.txt index 758659386..1cb61d8d9 100644 --- a/src/io/simulator/CMakeLists.txt +++ b/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) diff --git a/src/io/simulator/message_conversion.hpp b/src/io/simulator/message_conversion.hpp new file mode 100644 index 000000000..11a1322ea --- /dev/null +++ b/src/io/simulator/message_conversion.hpp @@ -0,0 +1,171 @@ +// 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 type that holds the + /// full parameter pack without interfering with recursive + /// narrowing expansion. + template + std::optional Unpack(std::any &&a) { + if (typeid(Head) == a.type()) { + Head concrete = std::any_cast(std::move(a)); + return concrete; + } + + if constexpr (sizeof...(Rest) > 0) { + return Unpack(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 + requires(sizeof...(Ms) > 0) std::optional> VariantFromAny(std::any &&a) { + return Unpack, Ms...>(std::move(a)); + } + + template + requires(sizeof...(Ms) > 0) std::optional> Take() { + std::optional> m_opt = VariantFromAny(std::move(message)); + + if (m_opt) { + return RequestEnvelope{ + .message = std::move(*m_opt), + .request_id = request_id, + .from_address = from_address, + }; + } else { + return std::nullopt; + } + } +}; + +class OpaquePromise { + const std::type_info *ti_; + void *ptr_; + std::function dtor_; + std::function is_awaited_; + std::function fill_; + std::function 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 + std::unique_ptr> Take() { + MG_ASSERT(typeid(T) == *ti_); + MG_ASSERT(ptr_ != nullptr); + + ResponsePromise *ptr = static_cast *>(ptr_); + + ptr_ = nullptr; + + return std::unique_ptr(ptr); + } + + template + explicit OpaquePromise(std::unique_ptr> promise) + : ti_(&typeid(T)), + ptr_(static_cast(promise.release())), + dtor_([](void *ptr) { static_cast *>(ptr)->~ResponsePromise(); }), + is_awaited_([](void *ptr) { return static_cast *>(ptr)->IsAwaited(); }), + fill_([](void *this_ptr, OpaqueMessage opaque_message) { + T message = std::any_cast(std::move(opaque_message.message)); + auto response_envelope = ResponseEnvelope{.message = std::move(message), + .request_id = opaque_message.request_id, + .from_address = opaque_message.from_address}; + ResponsePromise *promise = static_cast *>(this_ptr); + auto unique_promise = std::unique_ptr>(promise); + unique_promise->Fill(std::move(response_envelope)); + }), + time_out_([](void *ptr) { + ResponsePromise *promise = static_cast *>(ptr); + auto unique_promise = std::unique_ptr>(promise); + ResponseResult 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_); + } + } +}; + +} // namespace memgraph::io::simulator diff --git a/src/io/simulator/simulator_handle.cpp b/src/io/simulator/simulator_handle.cpp new file mode 100644 index 000000000..5dfd99b67 --- /dev/null +++ b/src/io/simulator/simulator_handle.cpp @@ -0,0 +1,175 @@ +// 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::TimeoutPromisesPastDeadline() { + const Time now = cluster_wide_time_microseconds_; + + for (auto &[promise_key, dop] : promises_) { + if (dop.deadline < now) { + spdlog::debug("timing out request from requester {} to replier {}.", promise_key.requester_address.ToString(), + promise_key.replier_address.ToString()); + DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key)); + promises_.erase(promise_key); + + stats_.timed_out_requests++; + + dop.promise.TimeOut(); + } + } +} + +void SimulatorHandle::ShutDown() { + std::unique_lock lock(mu_); + should_shut_down_ = true; + cv_.notify_all(); +} + +bool SimulatorHandle::ShouldShutDown() { + std::unique_lock lock(mu_); + return should_shut_down_; +} + +void SimulatorHandle::IncrementServerCountAndWaitForQuiescentState(Address address) { + std::unique_lock 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); + } +} + +bool SimulatorHandle::MaybeTickSimulator() { + std::unique_lock lock(mu_); + + 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++; + } + } + } + + 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 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 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()); + om_vec->second.emplace_back(std::move(opaque_message)); + } + + return true; +} + +Time SimulatorHandle::Now() { + std::unique_lock lock(mu_); + return cluster_wide_time_microseconds_; +} + +SimulatorStats SimulatorHandle::Stats() { + std::unique_lock lock(mu_); + return stats_; +} +} // namespace memgraph::io::simulator diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index 9819322d3..111448b59 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -25,6 +25,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,60 +36,6 @@ 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 type that holds the - /// full parameter pack without interfering with recursive - /// narrowing expansion. - template - std::optional Unpack(std::any &&a) { - if (typeid(Head) == a.type()) { - Head concrete = std::any_cast(std::move(a)); - return concrete; - } - - if constexpr (sizeof...(Rest) > 0) { - return Unpack(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 - requires(sizeof...(Ms) > 0) std::optional> VariantFromAny(std::any &&a) { - return Unpack, Ms...>(std::move(a)); - } - - template - requires(sizeof...(Ms) > 0) std::optional> Take() { - std::optional> m_opt = VariantFromAny(std::move(message)); - - if (m_opt) { - return RequestEnvelope{ - .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; @@ -104,101 +51,6 @@ struct PromiseKey { } }; -class OpaquePromise { - const std::type_info *ti_; - void *ptr_; - std::function dtor_; - std::function is_awaited_; - std::function fill_; - std::function 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 - std::unique_ptr> Take() { - MG_ASSERT(typeid(T) == *ti_); - MG_ASSERT(ptr_ != nullptr); - - ResponsePromise *ptr = static_cast *>(ptr_); - - ptr_ = nullptr; - - return std::unique_ptr(ptr); - } - - template - explicit OpaquePromise(std::unique_ptr> promise) - : ti_(&typeid(T)), - ptr_(static_cast(promise.release())), - dtor_([](void *ptr) { static_cast *>(ptr)->~ResponsePromise(); }), - is_awaited_([](void *ptr) { return static_cast *>(ptr)->IsAwaited(); }), - fill_([](void *this_ptr, OpaqueMessage opaque_message) { - T message = std::any_cast(std::move(opaque_message.message)); - auto response_envelope = ResponseEnvelope{.message = std::move(message), - .request_id = opaque_message.request_id, - .from_address = opaque_message.from_address}; - ResponsePromise *promise = static_cast *>(this_ptr); - auto unique_promise = std::unique_ptr>(promise); - unique_promise->Fill(std::move(response_envelope)); - }), - time_out_([](void *ptr) { - ResponsePromise *promise = static_cast *>(ptr); - auto unique_promise = std::unique_ptr>(promise); - ResponseResult 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_); - } - } -}; - struct DeadlineAndOpaquePromise { Time deadline; OpaquePromise promise; @@ -229,146 +81,19 @@ class SimulatorHandle { explicit SimulatorHandle(SimulatorConfig config) : cluster_wide_time_microseconds_(config.start_time), rng_(config.rng_seed), config_(config) {} - void IncrementServerCountAndWaitForQuiescentState(Address address) { - std::unique_lock lock(mu_); - server_addresses_.insert(address); + void IncrementServerCountAndWaitForQuiescentState(Address address); - while (true) { - size_t blocked_servers = blocked_on_receive_; + void TimeoutPromisesPastDeadline(); - 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(); - bool all_servers_blocked = blocked_servers == server_addresses_.size(); + void ShutDown(); - if (all_servers_blocked) { - return; - } - - cv_.wait(lock); - } - } - - void TimeoutPromisesPastDeadline() { - const Time now = cluster_wide_time_microseconds_; - - for (auto &[promise_key, dop] : promises_) { - if (dop.deadline < now) { - spdlog::debug("timing out request from requester {} to replier {}.", promise_key.requester_address.ToString(), - promise_key.replier_address.ToString()); - DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key)); - promises_.erase(promise_key); - - stats_.timed_out_requests++; - - dop.promise.TimeOut(); - } - } - } - - bool MaybeTickSimulator() { - std::unique_lock lock(mu_); - - 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++; - } - } - } - - 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 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 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()); - om_vec->second.emplace_back(std::move(opaque_message)); - } - - return true; - } - - void ShutDown() { - std::unique_lock lock(mu_); - should_shut_down_ = true; - cv_.notify_all(); - } - - bool ShouldShutDown() { - std::unique_lock lock(mu_); - return should_shut_down_; - } + bool ShouldShutDown(); template void SubmitRequest(Address to_address, Address from_address, uint64_t request_id, Request &&request, Duration timeout, @@ -444,10 +169,7 @@ class SimulatorHandle { cv_.notify_all(); } - Time Now() { - std::unique_lock lock(mu_); - return cluster_wide_time_microseconds_; - } + Time Now(); template , class Return = uint64_t> Return Rand(D distrib) { @@ -455,9 +177,6 @@ class SimulatorHandle { return distrib(rng_); } - SimulatorStats Stats() { - std::unique_lock lock(mu_); - return stats_; - } + SimulatorStats Stats(); }; }; // namespace memgraph::io::simulator From 69ea79a75ec191cb62834bc88a3a83a593a3f47f Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 11:07:39 +0000 Subject: [PATCH 05/31] Make a couple more things const --- src/io/simulator/simulator_handle.cpp | 2 +- src/io/simulator/simulator_handle.hpp | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/io/simulator/simulator_handle.cpp b/src/io/simulator/simulator_handle.cpp index 5dfd99b67..77dcef083 100644 --- a/src/io/simulator/simulator_handle.cpp +++ b/src/io/simulator/simulator_handle.cpp @@ -163,7 +163,7 @@ bool SimulatorHandle::MaybeTickSimulator() { return true; } -Time SimulatorHandle::Now() { +Time SimulatorHandle::Now() const { std::unique_lock lock(mu_); return cluster_wide_time_microseconds_; } diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index 111448b59..0db9d2231 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -169,7 +169,7 @@ class SimulatorHandle { cv_.notify_all(); } - Time Now(); + Time Now() const; template , class Return = uint64_t> Return Rand(D distrib) { From 0c2cbb5461f84abc9c40fcdb910110ae794fa0c1 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 11:08:17 +0000 Subject: [PATCH 06/31] Make a couple more things const --- src/io/simulator/simulator_handle.cpp | 2 +- src/io/simulator/simulator_handle.hpp | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/io/simulator/simulator_handle.cpp b/src/io/simulator/simulator_handle.cpp index 77dcef083..ba328eb75 100644 --- a/src/io/simulator/simulator_handle.cpp +++ b/src/io/simulator/simulator_handle.cpp @@ -45,7 +45,7 @@ void SimulatorHandle::ShutDown() { cv_.notify_all(); } -bool SimulatorHandle::ShouldShutDown() { +bool SimulatorHandle::ShouldShutDown() const { std::unique_lock lock(mu_); return should_shut_down_; } diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index 0db9d2231..cee649fc1 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -93,7 +93,7 @@ class SimulatorHandle { void ShutDown(); - bool ShouldShutDown(); + bool ShouldShutDown() const; template void SubmitRequest(Address to_address, Address from_address, uint64_t request_id, Request &&request, Duration timeout, From eb1b6c3ac8a2538b4ea8be6372854b6e4f09e79d Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 11:10:00 +0000 Subject: [PATCH 07/31] Make SimulatorHandle::TimeoutPromisesPastDeadline private --- src/io/simulator/simulator_handle.cpp | 17 ----------------- src/io/simulator/simulator_handle.hpp | 19 +++++++++++++++++-- 2 files changed, 17 insertions(+), 19 deletions(-) diff --git a/src/io/simulator/simulator_handle.cpp b/src/io/simulator/simulator_handle.cpp index ba328eb75..5d9a92ec5 100644 --- a/src/io/simulator/simulator_handle.cpp +++ b/src/io/simulator/simulator_handle.cpp @@ -22,23 +22,6 @@ namespace memgraph::io::simulator { using memgraph::io::Duration; using memgraph::io::Time; -void SimulatorHandle::TimeoutPromisesPastDeadline() { - const Time now = cluster_wide_time_microseconds_; - - for (auto &[promise_key, dop] : promises_) { - if (dop.deadline < now) { - spdlog::debug("timing out request from requester {} to replier {}.", promise_key.requester_address.ToString(), - promise_key.replier_address.ToString()); - DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key)); - promises_.erase(promise_key); - - stats_.timed_out_requests++; - - dop.promise.TimeOut(); - } - } -} - void SimulatorHandle::ShutDown() { std::unique_lock lock(mu_); should_shut_down_ = true; diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index cee649fc1..bd0c8fe63 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -77,14 +77,29 @@ class SimulatorHandle { std::mt19937 rng_; SimulatorConfig config_; + void TimeoutPromisesPastDeadline() { + const Time now = cluster_wide_time_microseconds_; + + for (auto &[promise_key, dop] : promises_) { + if (dop.deadline < now) { + spdlog::debug("timing out request from requester {} to replier {}.", promise_key.requester_address.ToString(), + promise_key.replier_address.ToString()); + DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key)); + promises_.erase(promise_key); + + stats_.timed_out_requests++; + + dop.promise.TimeOut(); + } + } + } + public: explicit SimulatorHandle(SimulatorConfig config) : cluster_wide_time_microseconds_(config.start_time), rng_(config.rng_seed), config_(config) {} void IncrementServerCountAndWaitForQuiescentState(Address address); - void TimeoutPromisesPastDeadline(); - /// 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 From b2b11f3a306e2dd9f892d48213106e7e7221896b Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 11:24:21 +0000 Subject: [PATCH 08/31] Run simulation tests in CI --- .github/workflows/diff.yaml | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/.github/workflows/diff.yaml b/.github/workflows/diff.yaml index bf6a39147..ce2caea75 100644 --- a/.github/workflows/diff.yaml +++ b/.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. From dbd744470b2e1d1f438d57101d6d832c8a8ecae7 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 11:48:21 +0000 Subject: [PATCH 09/31] Add future benchmark --- tests/benchmark/CMakeLists.txt | 3 +++ tests/benchmark/future.cpp | 30 ++++++++++++++++++++++++++++++ 2 files changed, 33 insertions(+) create mode 100644 tests/benchmark/future.cpp diff --git a/tests/benchmark/CMakeLists.txt b/tests/benchmark/CMakeLists.txt index 4bf8374b0..31f0eebc0 100644 --- a/tests/benchmark/CMakeLists.txt +++ b/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) diff --git a/tests/benchmark/future.cpp b/tests/benchmark/future.cpp new file mode 100644 index 000000000..abbe3fb98 --- /dev/null +++ b/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 + +#include "io/future.hpp" + +static void FuturePairFillWait(benchmark::State &state) { + uint64_t counter = 0; + while (state.KeepRunning()) { + auto [future, promise] = memgraph::io::FuturePromisePair(); + promise.Fill(1); + std::move(future).Wait(); + + ++counter; + } + state.SetItemsProcessed(counter); +} + +BENCHMARK(FuturePairFillWait)->Unit(benchmark::kNanosecond)->UseRealTime(); + +BENCHMARK_MAIN(); From b8487da3924f651ec552905b476783fe42c25fce Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 11:58:54 +0000 Subject: [PATCH 10/31] Move future test to unit tests --- tests/simulation/CMakeLists.txt | 2 -- tests/unit/CMakeLists.txt | 4 ++++ tests/{simulation => unit}/future.cpp | 0 3 files changed, 4 insertions(+), 2 deletions(-) rename tests/{simulation => unit}/future.cpp (100%) diff --git a/tests/simulation/CMakeLists.txt b/tests/simulation/CMakeLists.txt index 67f073418..142657401 100644 --- a/tests/simulation/CMakeLists.txt +++ b/tests/simulation/CMakeLists.txt @@ -25,8 +25,6 @@ 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(trial_query_storage/query_storage_test.cpp address) diff --git a/tests/unit/CMakeLists.txt b/tests/unit/CMakeLists.txt index 319627715..d8af2c201 100644 --- a/tests/unit/CMakeLists.txt +++ b/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) diff --git a/tests/simulation/future.cpp b/tests/unit/future.cpp similarity index 100% rename from tests/simulation/future.cpp rename to tests/unit/future.cpp From 7d33bb193799c33fcf708bb888dea068cc0d6d31 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 12:42:37 +0000 Subject: [PATCH 11/31] Update Future unit test to use gtest --- tests/unit/future.cpp | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/tests/unit/future.cpp b/tests/unit/future.cpp index 861769ea5..490e19bbc 100644 --- a/tests/unit/future.cpp +++ b/tests/unit/future.cpp @@ -12,8 +12,9 @@ #include #include +#include "gtest/gtest.h" + #include "io/future.hpp" -#include "utils/logging.hpp" using namespace memgraph::io; @@ -21,11 +22,11 @@ void Fill(Promise promise_1) { promise_1.Fill("success"); } void Wait(Future future_1, Promise 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 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"); } From ca638db509fe815951ae7ca8b6caf1ffea6779aa Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 12:45:45 +0000 Subject: [PATCH 12/31] Remove extra blank line --- src/io/simulator/simulator_handle.hpp | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index bd0c8fe63..f268d0e18 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -11,9 +11,8 @@ #pragma once -#include - #include +#include #include #include #include From 4f4eb9ea1374331fdaacce4ff20a5266e6f072fc Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 14:47:13 +0200 Subject: [PATCH 13/31] Update src/io/future.hpp MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: János Benjamin Antal --- src/io/future.hpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/io/future.hpp b/src/io/future.hpp index f935f4146..c585668f4 100644 --- a/src/io/future.hpp +++ b/src/io/future.hpp @@ -87,7 +87,7 @@ class Shared { break; } } - if (!simulator_progressed) { + if (!simulator_progressed) [[likely]] { cv_.wait(lock); } MG_ASSERT(!consumed_, "Future consumed twice!"); From a3f3e05fc254cbc3fdb1d1a8c6399330aaee75f2 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 12:50:48 +0000 Subject: [PATCH 14/31] Include replier address in operator< for PromiseKey --- src/io/simulator/simulator_handle.hpp | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index f268d0e18..afc66c6f5 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -38,12 +38,18 @@ using memgraph::io::Time; 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; + if (request_id == other.request_id) { + return replier_address < other.replier_address; + } else { + return request_id < other.request_id; + } } else { return requester_address < other.requester_address; } From 102d9972884bdb4b743f1c952007cc3f047c986f Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 14:52:06 +0200 Subject: [PATCH 15/31] Update src/io/simulator/simulator_handle.hpp MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: János Benjamin Antal --- src/io/simulator/simulator_handle.hpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index afc66c6f5..c32aa1fc0 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -89,7 +89,7 @@ class SimulatorHandle { if (dop.deadline < now) { spdlog::debug("timing out request from requester {} to replier {}.", promise_key.requester_address.ToString(), promise_key.replier_address.ToString()); - DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key)); + DeadlineAndOpaquePromise dop = std::move(dop); promises_.erase(promise_key); stats_.timed_out_requests++; From 649b5437b05c35e81e3ae49cf24bd4cc3bf3f39b Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 12:58:55 +0000 Subject: [PATCH 16/31] Improve docs around Io interface --- src/io/transport.hpp | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/src/io/transport.hpp b/src/io/transport.hpp index 7d079685e..36df67c66 100644 --- a/src/io/transport.hpp +++ b/src/io/transport.hpp @@ -71,14 +71,15 @@ 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 ResponseFuture RequestWithTimeout(Address address, Request request, Duration timeout) { const uint64_t request_id = ++request_id_counter_; return implementation_.template Request(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 ResponseFuture Request(Address address, Request request) { const uint64_t request_id = ++request_id_counter_; @@ -87,34 +88,34 @@ class Io { } /// 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 RequestResult ReceiveWithTimeout(Duration timeout) { return implementation_.template Receive(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 requires(sizeof...(Ms) > 0) RequestResult Receive() { const Duration timeout = default_timeout_; return implementation_.template Receive(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 void Send(Address address, uint64_t request_id, M message) { return implementation_.template Send(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() const { return implementation_.Now(); } - /// Returns true of the system should shut-down. + /// Returns true if the system should shut-down. bool ShouldShutDown() const { return implementation_.ShouldShutDown(); } /// Returns a random number within the specified distribution. From cacb0dac8023c126e436c97bf25499a348dc7d08 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 14:12:07 +0000 Subject: [PATCH 17/31] Avoid unnamed namespace in future.hpp --- src/io/future.hpp | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/src/io/future.hpp b/src/io/future.hpp index c585668f4..f694e22dc 100644 --- a/src/io/future.hpp +++ b/src/io/future.hpp @@ -27,7 +27,7 @@ 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 class Shared { mutable std::condition_variable cv_; @@ -131,15 +131,15 @@ class Shared { return waiting_; } }; -} // namespace +} // namespace details template class Future { bool consumed_or_moved_ = false; - std::shared_ptr> shared_; + std::shared_ptr> shared_; public: - explicit Future(std::shared_ptr> shared) : shared_(shared) {} + explicit Future(std::shared_ptr> shared) : shared_(shared) {} Future() = delete; Future(Future &&old) { @@ -198,11 +198,11 @@ class Future { template class Promise { - std::shared_ptr> shared_; + std::shared_ptr> shared_; bool filled_or_moved_ = false; public: - explicit Promise(std::shared_ptr> shared) : shared_(shared) {} + explicit Promise(std::shared_ptr> shared) : shared_(shared) {} Promise() = delete; Promise(Promise &&old) { @@ -242,7 +242,7 @@ class Promise { template std::pair, Promise> FuturePromisePair() { - std::shared_ptr> shared = std::make_shared>(); + std::shared_ptr> shared = std::make_shared>(); Future future = Future(shared); Promise promise = Promise(shared); @@ -252,7 +252,7 @@ std::pair, Promise> FuturePromisePair() { template std::pair, Promise> FuturePromisePairWithNotifier(std::function simulator_notifier) { - std::shared_ptr> shared = std::make_shared>(simulator_notifier); + std::shared_ptr> shared = std::make_shared>(simulator_notifier); Future future = Future(shared); Promise promise = Promise(shared); From 5bb3361a2db63e90eff812188dcb890068ec379a Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 14:32:07 +0000 Subject: [PATCH 18/31] Use std::chrono::microseconds explicitly for Duration. Fix compiler warning related to timeouts --- src/io/simulator/simulator_handle.hpp | 4 +--- src/io/time.hpp | 2 +- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index c32aa1fc0..140fd9111 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -89,12 +89,10 @@ class SimulatorHandle { if (dop.deadline < now) { spdlog::debug("timing out request from requester {} to replier {}.", promise_key.requester_address.ToString(), promise_key.replier_address.ToString()); - DeadlineAndOpaquePromise dop = std::move(dop); + std::move(dop).promise.TimeOut(); promises_.erase(promise_key); stats_.timed_out_requests++; - - dop.promise.TimeOut(); } } } diff --git a/src/io/time.hpp b/src/io/time.hpp index 6ab1f1184..57f58cab1 100644 --- a/src/io/time.hpp +++ b/src/io/time.hpp @@ -15,7 +15,7 @@ namespace memgraph::io { -using Duration = std::chrono::duration>; +using Duration = std::chrono::microseconds; using Time = std::chrono::time_point; } // namespace memgraph::io From 9576eea0511130d50262fcbd5c417c3d22e479b9 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 15:37:16 +0000 Subject: [PATCH 19/31] Address some feedback from clang-tidy --- src/io/address.hpp | 6 ++---- src/io/future.hpp | 12 ++++++------ src/io/simulator/message_conversion.hpp | 22 +++++++++++----------- src/io/simulator/simulator.hpp | 2 +- src/io/simulator/simulator_handle.hpp | 12 ++++-------- src/io/simulator/simulator_stats.hpp | 2 ++ src/io/transport.hpp | 3 ++- tests/simulation/basic_request.cpp | 2 +- 8 files changed, 29 insertions(+), 32 deletions(-) diff --git a/src/io/address.hpp b/src/io/address.hpp index d649dd451..5ad76fb7e 100644 --- a/src/io/address.hpp +++ b/src/io/address.hpp @@ -47,12 +47,10 @@ struct Address { 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 { - return unique_id < other.unique_id; + return last_known_ip < other.last_known_ip; } + return unique_id < other.unique_id; } std::string ToString() const { diff --git a/src/io/future.hpp b/src/io/future.hpp index f694e22dc..bc4adb9f6 100644 --- a/src/io/future.hpp +++ b/src/io/future.hpp @@ -108,9 +108,9 @@ class Shared { if (item_) { return Take(lock); - } else { - return std::nullopt; } + + return std::nullopt; } void Fill(T item) { @@ -142,14 +142,14 @@ class Future { explicit Future(std::shared_ptr> 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; @@ -205,13 +205,13 @@ class Promise { explicit Promise(std::shared_ptr> 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; diff --git a/src/io/simulator/message_conversion.hpp b/src/io/simulator/message_conversion.hpp index 11a1322ea..ab57bd18c 100644 --- a/src/io/simulator/message_conversion.hpp +++ b/src/io/simulator/message_conversion.hpp @@ -67,9 +67,9 @@ struct OpaqueMessage { .request_id = request_id, .from_address = from_address, }; - } else { - return std::nullopt; } + + return std::nullopt; } }; @@ -82,17 +82,17 @@ class OpaquePromise { std::function time_out_; public: - OpaquePromise(OpaquePromise &&old) + OpaquePromise(OpaquePromise &&old) noexcept : ti_(old.ti_), ptr_(old.ptr_), - dtor_(old.dtor_), - is_awaited_(old.is_awaited_), - fill_(old.fill_), - time_out_(old.time_out_) { + dtor_(std::move(old.dtor_)), + is_awaited_(std::move(old.is_awaited_)), + fill_(std::move(old.fill_)), + time_out_(std::move(old.time_out_)) { old.ptr_ = nullptr; } - OpaquePromise &operator=(OpaquePromise &&old) { + OpaquePromise &operator=(OpaquePromise &&old) noexcept { MG_ASSERT(this != &old); ptr_ = old.ptr_; @@ -115,7 +115,7 @@ class OpaquePromise { MG_ASSERT(typeid(T) == *ti_); MG_ASSERT(ptr_ != nullptr); - ResponsePromise *ptr = static_cast *>(ptr_); + auto ptr = static_cast *>(ptr_); ptr_ = nullptr; @@ -133,12 +133,12 @@ class OpaquePromise { auto response_envelope = ResponseEnvelope{.message = std::move(message), .request_id = opaque_message.request_id, .from_address = opaque_message.from_address}; - ResponsePromise *promise = static_cast *>(this_ptr); + auto promise = static_cast *>(this_ptr); auto unique_promise = std::unique_ptr>(promise); unique_promise->Fill(std::move(response_envelope)); }), time_out_([](void *ptr) { - ResponsePromise *promise = static_cast *>(ptr); + auto promise = static_cast *>(ptr); auto unique_promise = std::unique_ptr>(promise); ResponseResult result = TimedOut{}; unique_promise->Fill(std::move(result)); diff --git a/src/io/simulator/simulator.hpp b/src/io/simulator/simulator.hpp index 8e3ad85c0..354aae6ac 100644 --- a/src/io/simulator/simulator.hpp +++ b/src/io/simulator/simulator.hpp @@ -33,7 +33,7 @@ class Simulator { Io Register(Address address) { std::uniform_int_distribution 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) { diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index 140fd9111..cd5d5565d 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -47,12 +47,10 @@ struct PromiseKey { if (requester_address == other.requester_address) { if (request_id == other.request_id) { return replier_address < other.replier_address; - } else { - return request_id < other.request_id; } - } else { - return requester_address < other.requester_address; + return request_id < other.request_id; } + return requester_address < other.requester_address; } }; @@ -120,9 +118,9 @@ class SimulatorHandle { const Time deadline = cluster_wide_time_microseconds_ + timeout; - std::any message(std::move(request)); + std::any message(std::forward(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()); @@ -133,8 +131,6 @@ class SimulatorHandle { stats_.total_requests++; cv_.notify_all(); - - return; } template diff --git a/src/io/simulator/simulator_stats.hpp b/src/io/simulator/simulator_stats.hpp index 1f016c2fb..7f529a456 100644 --- a/src/io/simulator/simulator_stats.hpp +++ b/src/io/simulator/simulator_stats.hpp @@ -11,6 +11,8 @@ #pragma once +#include + namespace memgraph::io::simulator { struct SimulatorStats { uint64_t total_messages = 0; diff --git a/src/io/transport.hpp b/src/io/transport.hpp index 36df67c66..a9e550434 100644 --- a/src/io/transport.hpp +++ b/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. diff --git a/tests/simulation/basic_request.cpp b/tests/simulation/basic_request.cpp index 6e87b8e41..ac3190ad7 100644 --- a/tests/simulation/basic_request.cpp +++ b/tests/simulation/basic_request.cpp @@ -30,7 +30,7 @@ struct CounterResponse { }; void run_server(Io io) { - uint64_t highest_seen; + uint64_t highest_seen = 0; while (!io.ShouldShutDown()) { std::cout << "[SERVER] Is receiving..." << std::endl; From ace5f2b6394509c16f01c991af359bfbf8ace543 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 15:44:34 +0000 Subject: [PATCH 20/31] Complete clang-tidy cleanup --- src/io/future.hpp | 7 +++---- src/io/simulator/simulator_transport.hpp | 2 +- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/src/io/future.hpp b/src/io/future.hpp index bc4adb9f6..63f3989de 100644 --- a/src/io/future.hpp +++ b/src/io/future.hpp @@ -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 &) { + T Take() { MG_ASSERT(item_, "Take called without item_ being present"); MG_ASSERT(!consumed_, "Take called on already-consumed Future"); @@ -95,7 +94,7 @@ class Shared { waiting_ = false; - return Take(lock); + return Take(); } bool IsReady() { @@ -107,7 +106,7 @@ class Shared { std::unique_lock lock(mu_); if (item_) { - return Take(lock); + return Take(); } return std::nullopt; diff --git a/src/io/simulator/simulator_transport.hpp b/src/io/simulator/simulator_transport.hpp index 5f4c52a3d..b67371ff0 100644 --- a/src/io/simulator/simulator_transport.hpp +++ b/src/io/simulator/simulator_transport.hpp @@ -26,7 +26,7 @@ using memgraph::io::Time; class SimulatorTransport { std::shared_ptr simulator_handle_; const Address address_; - std::mt19937 rng_{}; + std::mt19937 rng_; public: SimulatorTransport(std::shared_ptr simulator_handle, Address address, uint64_t seed) From 997d25d5369508150e9885b5e1fb5110154ab4b7 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Mon, 1 Aug 2022 15:53:37 +0000 Subject: [PATCH 21/31] Remove buggy usage of std::forward --- src/io/simulator/simulator_handle.hpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index cd5d5565d..0d4ab1585 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -118,7 +118,7 @@ class SimulatorHandle { const Time deadline = cluster_wide_time_microseconds_ + timeout; - std::any message(std::forward(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(to_address, std::move(om))); From dd93b594bc1bb0bdc05349c635ba8fc9ad6f01bf Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Tue, 2 Aug 2022 07:01:25 +0000 Subject: [PATCH 22/31] Remove dead code --- tests/simulation/trial_query_storage/query_storage_test.cpp | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/simulation/trial_query_storage/query_storage_test.cpp b/tests/simulation/trial_query_storage/query_storage_test.cpp index 79c11da60..9cdff4ee6 100644 --- a/tests/simulation/trial_query_storage/query_storage_test.cpp +++ b/tests/simulation/trial_query_storage/query_storage_test.cpp @@ -80,7 +80,6 @@ int main() { auto res_f = cli_io.Request(srv_addr, req); auto res_rez = std::move(res_f).Wait(); - // MG_ASSERT(res_rez.HasError()); simulator.ShutDown(); return 0; } From e935a9a7b13b86d2434aca98f06b82fb8056cf16 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Tue, 2 Aug 2022 07:02:17 +0000 Subject: [PATCH 23/31] Make parenthesization less confusing --- src/io/address.hpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/io/address.hpp b/src/io/address.hpp index 5ad76fb7e..65ae2cf52 100644 --- a/src/io/address.hpp +++ b/src/io/address.hpp @@ -38,7 +38,7 @@ 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); } From 4d8f9ea821ea031c5b830fcef3484e2416fc9b53 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Tue, 2 Aug 2022 07:03:12 +0000 Subject: [PATCH 24/31] Remove unnecessary comment --- tests/simulation/trial_query_storage/messages.hpp | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/simulation/trial_query_storage/messages.hpp b/tests/simulation/trial_query_storage/messages.hpp index 3dfbdc33a..8db78a54c 100644 --- a/tests/simulation/trial_query_storage/messages.hpp +++ b/tests/simulation/trial_query_storage/messages.hpp @@ -15,8 +15,6 @@ #include #include -// header - namespace memgraph::tests::simulation { struct Vertex { From 3b44ef70b6d702a3f828b0578020d7d24f850421 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Tue, 2 Aug 2022 07:06:43 +0000 Subject: [PATCH 25/31] Denest operator< for clarity --- src/io/address.hpp | 12 +++++++----- src/io/simulator/simulator_handle.hpp | 12 +++++++----- 2 files changed, 14 insertions(+), 10 deletions(-) diff --git a/src/io/address.hpp b/src/io/address.hpp index 65ae2cf52..ad2c92b57 100644 --- a/src/io/address.hpp +++ b/src/io/address.hpp @@ -44,13 +44,15 @@ struct Address { /// 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; - } + 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 unique_id < other.unique_id; + + return last_known_port < other.last_known_port; } std::string ToString() const { diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index 0d4ab1585..5280b9ed5 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -44,13 +44,15 @@ struct PromiseKey { public: bool operator<(const PromiseKey &other) const { - if (requester_address == other.requester_address) { - if (request_id == other.request_id) { - return replier_address < other.replier_address; - } + if (requester_address != other.requester_address) { + return requester_address < other.requester_address; + } + + if (request_id != other.request_id) { return request_id < other.request_id; } - return requester_address < other.requester_address; + + return replier_address < other.replier_address; } }; From 902a46d14fb6b2abdf352c535da4dfd96dcb1ee0 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Tue, 2 Aug 2022 07:12:05 +0000 Subject: [PATCH 26/31] Make dynamic message Take messages rvalue methods --- src/io/simulator/message_conversion.hpp | 4 ++-- src/io/simulator/simulator_handle.hpp | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/io/simulator/message_conversion.hpp b/src/io/simulator/message_conversion.hpp index ab57bd18c..41551c4eb 100644 --- a/src/io/simulator/message_conversion.hpp +++ b/src/io/simulator/message_conversion.hpp @@ -58,7 +58,7 @@ struct OpaqueMessage { } template - requires(sizeof...(Ms) > 0) std::optional> Take() { + requires(sizeof...(Ms) > 0) std::optional> Take() && { std::optional> m_opt = VariantFromAny(std::move(message)); if (m_opt) { @@ -111,7 +111,7 @@ class OpaquePromise { OpaquePromise &operator=(const OpaquePromise &) = delete; template - std::unique_ptr> Take() { + std::unique_ptr> Take() && { MG_ASSERT(typeid(T) == *ti_); MG_ASSERT(ptr_ != nullptr); diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index 5280b9ed5..f35a9faa0 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -152,7 +152,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(); + auto m_opt = std::move(message).Take(); blocked_on_receive_ -= 1; From ad0a8c4942077d233e52ec3afe3657bb068a846e Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Tue, 2 Aug 2022 13:36:40 +0000 Subject: [PATCH 27/31] Use simple virtual inheritance instead of bespoke vtable for OpaquePromise --- src/io/simulator/message_conversion.hpp | 93 ++++++++++++------------- 1 file changed, 45 insertions(+), 48 deletions(-) diff --git a/src/io/simulator/message_conversion.hpp b/src/io/simulator/message_conversion.hpp index 41551c4eb..017cf8e6f 100644 --- a/src/io/simulator/message_conversion.hpp +++ b/src/io/simulator/message_conversion.hpp @@ -73,37 +73,54 @@ struct OpaqueMessage { } }; +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; +}; + +template +class OpaquePromiseTrait : public OpaquePromiseTraitBase { + public: + ~OpaquePromiseTrait() = default; + + const std::type_info *TypeInfo() const override { return &typeid(T); }; + + bool IsAwaited(void *ptr) const override { return static_cast *>(ptr)->IsAwaited(); }; + + void Fill(void *ptr, OpaqueMessage &&opaque_message) const override { + T message = std::any_cast(std::move(opaque_message.message)); + auto response_envelope = ResponseEnvelope{.message = std::move(message), + .request_id = opaque_message.request_id, + .from_address = opaque_message.from_address}; + auto promise = static_cast *>(ptr); + auto unique_promise = std::unique_ptr>(promise); + unique_promise->Fill(std::move(response_envelope)); + }; + + void TimeOut(void *ptr) const override { + auto promise = static_cast *>(ptr); + auto unique_promise = std::unique_ptr>(promise); + ResponseResult result = TimedOut{}; + unique_promise->Fill(std::move(result)); + } +}; + class OpaquePromise { - const std::type_info *ti_; void *ptr_; - std::function dtor_; - std::function is_awaited_; - std::function fill_; - std::function time_out_; + std::unique_ptr trait_; public: - OpaquePromise(OpaquePromise &&old) noexcept - : ti_(old.ti_), - ptr_(old.ptr_), - dtor_(std::move(old.dtor_)), - is_awaited_(std::move(old.is_awaited_)), - fill_(std::move(old.fill_)), - time_out_(std::move(old.time_out_)) { - old.ptr_ = nullptr; - } + 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_; - ti_ = old.ti_; - dtor_ = old.dtor_; - is_awaited_ = old.is_awaited_; - fill_ = old.fill_; - time_out_ = old.time_out_; - + trait_ = std::move(old.trait_); old.ptr_ = nullptr; - return *this; } @@ -112,7 +129,7 @@ class OpaquePromise { template std::unique_ptr> Take() && { - MG_ASSERT(typeid(T) == *ti_); + MG_ASSERT(typeid(T) == *trait_->TypeInfo()); MG_ASSERT(ptr_ != nullptr); auto ptr = static_cast *>(ptr_); @@ -124,47 +141,27 @@ class OpaquePromise { template explicit OpaquePromise(std::unique_ptr> promise) - : ti_(&typeid(T)), - ptr_(static_cast(promise.release())), - dtor_([](void *ptr) { static_cast *>(ptr)->~ResponsePromise(); }), - is_awaited_([](void *ptr) { return static_cast *>(ptr)->IsAwaited(); }), - fill_([](void *this_ptr, OpaqueMessage opaque_message) { - T message = std::any_cast(std::move(opaque_message.message)); - auto response_envelope = ResponseEnvelope{.message = std::move(message), - .request_id = opaque_message.request_id, - .from_address = opaque_message.from_address}; - auto promise = static_cast *>(this_ptr); - auto unique_promise = std::unique_ptr>(promise); - unique_promise->Fill(std::move(response_envelope)); - }), - time_out_([](void *ptr) { - auto promise = static_cast *>(ptr); - auto unique_promise = std::unique_ptr>(promise); - ResponseResult result = TimedOut{}; - unique_promise->Fill(std::move(result)); - }) {} + : ptr_(static_cast(promise.release())), trait_(std::make_unique>()) {} bool IsAwaited() { MG_ASSERT(ptr_ != nullptr); - return is_awaited_(ptr_); + return trait_->IsAwaited(ptr_); } void TimeOut() { MG_ASSERT(ptr_ != nullptr); - time_out_(ptr_); + trait_->TimeOut(ptr_); ptr_ = nullptr; } void Fill(OpaqueMessage &&opaque_message) { MG_ASSERT(ptr_ != nullptr); - fill_(ptr_, std::move(opaque_message)); + trait_->Fill(ptr_, std::move(opaque_message)); ptr_ = nullptr; } ~OpaquePromise() { - if (nullptr != ptr_) { - dtor_(ptr_); - } + MG_ASSERT(ptr_ == nullptr, "OpaquePromise destroyed without being explicitly timed out or filled"); } }; From 4f06eb0f2f11f4d4f6035dd9779414fedc2ffbdd Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Tue, 2 Aug 2022 14:24:59 +0000 Subject: [PATCH 28/31] Address rule-of-five for abstract inhertance in OpaquePromise --- src/io/simulator/message_conversion.hpp | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/io/simulator/message_conversion.hpp b/src/io/simulator/message_conversion.hpp index 017cf8e6f..379fc47e3 100644 --- a/src/io/simulator/message_conversion.hpp +++ b/src/io/simulator/message_conversion.hpp @@ -79,14 +79,18 @@ class OpaquePromiseTraitBase { 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 class OpaquePromiseTrait : public OpaquePromiseTraitBase { public: - ~OpaquePromiseTrait() = default; - const std::type_info *TypeInfo() const override { return &typeid(T); }; bool IsAwaited(void *ptr) const override { return static_cast *>(ptr)->IsAwaited(); }; From 54369958d1d85ec15c69140e8b13668121718064 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Tue, 2 Aug 2022 14:30:12 +0000 Subject: [PATCH 29/31] Extract BlockedServers functionality from MaybeTickSimulator and IncrementServerCountAndWaitForQuiescentState --- src/io/simulator/simulator_handle.cpp | 22 ++++++++++------------ src/io/simulator/simulator_handle.hpp | 8 ++++++++ 2 files changed, 18 insertions(+), 12 deletions(-) diff --git a/src/io/simulator/simulator_handle.cpp b/src/io/simulator/simulator_handle.cpp index 5d9a92ec5..27a917aa5 100644 --- a/src/io/simulator/simulator_handle.cpp +++ b/src/io/simulator/simulator_handle.cpp @@ -38,15 +38,7 @@ void SimulatorHandle::IncrementServerCountAndWaitForQuiescentState(Address addre 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++; - } - } - } + size_t blocked_servers = BlockedServers(); bool all_servers_blocked = blocked_servers == server_addresses_.size(); @@ -58,9 +50,7 @@ void SimulatorHandle::IncrementServerCountAndWaitForQuiescentState(Address addre } } -bool SimulatorHandle::MaybeTickSimulator() { - std::unique_lock lock(mu_); - +size_t SimulatorHandle::BlockedServers() { size_t blocked_servers = blocked_on_receive_; for (auto &[promise_key, opaque_promise] : promises_) { @@ -71,6 +61,14 @@ bool SimulatorHandle::MaybeTickSimulator() { } } + return blocked_servers; +} + +bool SimulatorHandle::MaybeTickSimulator() { + std::unique_lock lock(mu_); + + 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 diff --git a/src/io/simulator/simulator_handle.hpp b/src/io/simulator/simulator_handle.hpp index f35a9faa0..668d2389b 100644 --- a/src/io/simulator/simulator_handle.hpp +++ b/src/io/simulator/simulator_handle.hpp @@ -82,6 +82,14 @@ class SimulatorHandle { std::mt19937 rng_; SimulatorConfig config_; + /// 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_; From b2a8063a96ad88b8603cf58955b111fb639ba307 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Tue, 2 Aug 2022 14:33:05 +0000 Subject: [PATCH 30/31] Make more temporaries const --- src/io/simulator/simulator_handle.cpp | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/io/simulator/simulator_handle.cpp b/src/io/simulator/simulator_handle.cpp index 27a917aa5..46575e277 100644 --- a/src/io/simulator/simulator_handle.cpp +++ b/src/io/simulator/simulator_handle.cpp @@ -38,9 +38,9 @@ void SimulatorHandle::IncrementServerCountAndWaitForQuiescentState(Address addre server_addresses_.insert(address); while (true) { - size_t blocked_servers = BlockedServers(); + const size_t blocked_servers = BlockedServers(); - bool all_servers_blocked = blocked_servers == server_addresses_.size(); + const bool all_servers_blocked = blocked_servers == server_addresses_.size(); if (all_servers_blocked) { return; @@ -67,7 +67,7 @@ size_t SimulatorHandle::BlockedServers() { bool SimulatorHandle::MaybeTickSimulator() { std::unique_lock lock(mu_); - size_t blocked_servers = BlockedServers(); + const size_t blocked_servers = BlockedServers(); if (blocked_servers < server_addresses_.size()) { // we only need to advance the simulator when all @@ -100,7 +100,7 @@ bool SimulatorHandle::MaybeTickSimulator() { if (config_.scramble_messages) { // scramble messages std::uniform_int_distribution swap_distrib(0, in_flight_.size() - 1); - size_t swap_index = swap_distrib(rng_); + const size_t swap_index = swap_distrib(rng_); std::swap(in_flight_[swap_index], in_flight_.back()); } @@ -108,8 +108,8 @@ bool SimulatorHandle::MaybeTickSimulator() { in_flight_.pop_back(); std::uniform_int_distribution drop_distrib(0, 99); - int drop_threshold = drop_distrib(rng_); - bool should_drop = drop_threshold < config_.drop_percent; + const int drop_threshold = drop_distrib(rng_); + const bool should_drop = drop_threshold < config_.drop_percent; if (should_drop) { stats_.dropped_messages++; From 0a43afdec1637a548a2f1143633b12b0f795a842 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Thu, 4 Aug 2022 13:28:52 +0000 Subject: [PATCH 31/31] Update raft implementation to use std::chrono like upstream --- src/io/rsm/raft.hpp | 33 +++++++++++++++++++++++++-------- tests/simulation/CMakeLists.txt | 2 ++ tests/simulation/raft.cpp | 15 +++++++++------ 3 files changed, 36 insertions(+), 14 deletions(-) diff --git a/src/io/rsm/raft.hpp b/src/io/rsm/raft.hpp index d3b9de49d..3798c78fd 100644 --- a/src/io/rsm/raft.hpp +++ b/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 @@ -132,14 +132,14 @@ struct PendingClientRequest { struct Leader { std::map followers; std::deque pending_client_requests; - Time last_broadcast = 0; + Time last_broadcast = Time::min(); void Print() { std::cout << "\tLeader \t"; } }; struct Candidate { std::map successful_votes; - Time election_began = 0; + Time election_began = Time::min(); std::set
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(min).count(); + auto max_micros = std::chrono::duration_cast(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 void Log(Ts &&...args) { const Time now = io_.Now(); + auto micros = std::chrono::duration_cast(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 Cron(Candidate &candidate) { const auto now = io_.Now(); const Duration election_timeout = RandomTimeout(100000, 200000); + auto election_timeout_us = std::chrono::duration_cast(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{ diff --git a/tests/simulation/CMakeLists.txt b/tests/simulation/CMakeLists.txt index 142657401..e868ca0ef 100644 --- a/tests/simulation/CMakeLists.txt +++ b/tests/simulation/CMakeLists.txt @@ -27,4 +27,6 @@ endfunction(add_simulation_test) add_simulation_test(basic_request.cpp address) +add_simulation_test(raft.cpp address) + add_simulation_test(trial_query_storage/query_storage_test.cpp address) diff --git a/tests/simulation/raft.cpp b/tests/simulation/raft.cpp index 7cf6144fd..7998eb08e 100644 --- a/tests/simulation/raft.cpp +++ b/tests/simulation/raft.cpp @@ -9,6 +9,7 @@ // by the Apache License, Version 2.0, included in the file // licenses/APL.txt. +#include #include #include #include @@ -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> cas_response_future = - cli_io.RequestWithTimeout, WriteResponse>(leader, cli_req, 50000); + cli_io.Request, WriteResponse>(leader, cli_req); // receive cas_response - ResponseResult> cas_response_result = cas_response_future.Wait(); + ResponseResult> 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> get_response_future = - cli_io.RequestWithTimeout, ReadResponse>(leader, read_req, 50000); + cli_io.Request, ReadResponse>(leader, read_req); // receive response - ResponseResult> get_response_result = get_response_future.Wait(); + ResponseResult> 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;