diff --git a/src/io/v3/CMakeLists.txt b/src/io/v3/CMakeLists.txt index 08fd55582..e6a750431 100644 --- a/src/io/v3/CMakeLists.txt +++ b/src/io/v3/CMakeLists.txt @@ -4,7 +4,8 @@ set(io_v3_sources future.hpp transport.hpp simulator.hpp - simulator_handle.hpp) + simulator_handle.hpp + simulator_stats.hpp) find_package(fmt REQUIRED) find_package(Threads REQUIRED) diff --git a/src/io/v3/simulator.hpp b/src/io/v3/simulator.hpp index 6a09981e5..e5522c813 100644 --- a/src/io/v3/simulator.hpp +++ b/src/io/v3/simulator.hpp @@ -19,14 +19,6 @@ #include "simulator_handle.hpp" #include "transport.hpp" -struct SimulatorStats { - uint64_t total_messages_; - uint64_t dropped_messages_; - uint64_t total_requests_; - uint64_t total_responses_; - uint64_t simulator_ticks_; -}; - struct SimulatorConfig { uint8_t drop_percent_; uint64_t rng_seed_; @@ -38,20 +30,21 @@ class SimulatorTransport { : simulator_handle_(simulator_handle), address_(address) {} template - ResponseFuture RequestTimeout(Address address, uint64_t request_id, Request request, - uint64_t timeout_microseconds) { + ResponseFuture Request(Address address, uint64_t request_id, Request request, + uint64_t timeout_microseconds) { std::function notifier = [=] { simulator_handle_->NotifySimulator(); }; auto [future, promise] = FuturePromisePairWithNotifier>(notifier); - simulator_handle_->SubmitRequest(address, request_id, request, timeout_microseconds, std::move(promise)); + simulator_handle_->SubmitRequest(address, address_, request_id, std::move(request), timeout_microseconds, + std::move(promise)); return std::move(future); } /* template - RequestResult ReceiveTimeout(uint64_t timeout_microseconds) { - return simulator_handle_->template ReceiveTimeout(timeout_microseconds); + RequestResult Receive(uint64_t timeout_microseconds) { + return simulator_handle_->template Receive(timeout_microseconds); } template diff --git a/src/io/v3/simulator_handle.hpp b/src/io/v3/simulator_handle.hpp index 660171834..9e0e73d9b 100644 --- a/src/io/v3/simulator_handle.hpp +++ b/src/io/v3/simulator_handle.hpp @@ -12,14 +12,16 @@ #pragma once #include +#include #include "address.hpp" +#include "simulator_stats.hpp" #include "transport.hpp" struct OpaqueMessage { Address address; uint64_t request_id; - std::unique_ptr message; + std::any message; }; struct PromiseKey { @@ -29,8 +31,8 @@ struct PromiseKey { }; struct OpaquePromise { - time_t deadline; - std::unique_ptr promise; + uint64_t deadline; + std::any promise; }; class SimulatorHandle { @@ -46,14 +48,32 @@ class SimulatorHandle { } template - void SubmitRequest(Address address, uint64_t request_id, Request request, uint64_t timeout_microseconds, - MgPromise> promise) { + void SubmitRequest(Address to_addr, Address from_addr, uint64_t request_id, Request &&request, + uint64_t timeout_microseconds, MgPromise> &&promise) { std::unique_lock lock(mu_); + + uint64_t deadline = cluster_wide_time_microseconds_ + timeout_microseconds; + + std::any message(std::move(request)); + OpaqueMessage om{.address = from_addr, .request_id = request_id, .message = std::move(message)}; + in_flight_.emplace_back(std::make_pair(std::move(to_addr), std::move(om))); + + /* + std::any opaque_promise(std::move(promise)); + PromiseKey pk { .requester=from_addr, .request_id=request_id, .replier=to_addr }; + OpaquePromise op { .deadline=deadline, .promise=std::move(opaque_promise) }; + promises_.insert(std::make_pair(std::move(pk), std::move(op))); + */ + + stats_.total_messages_++; + stats_.total_requests_++; + + return; } /* template - RequestResult ReceiveTimeout(uint64_t timeout_microseconds) { + RequestResult Receive(uint64_t timeout_microseconds) { std::abort(); } @@ -67,8 +87,10 @@ class SimulatorHandle { std::mutex mu_; std::condition_variable cv_sim_; std::condition_variable cv_srv_; - std::map> in_flight_; - std::map promises; - std::map can_receive; + std::vector> in_flight_; + std::map promises_; + std::map can_receive_; + uint64_t cluster_wide_time_microseconds_ = 0; bool shut_down_; + SimulatorStats stats_; }; diff --git a/src/io/v3/simulator_stats.hpp b/src/io/v3/simulator_stats.hpp new file mode 100644 index 000000000..e15ef3ec4 --- /dev/null +++ b/src/io/v3/simulator_stats.hpp @@ -0,0 +1,20 @@ +// 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 + +struct SimulatorStats { + uint64_t total_messages_; + uint64_t dropped_messages_; + uint64_t total_requests_; + uint64_t total_responses_; + uint64_t simulator_ticks_; +}; diff --git a/src/io/v3/transport.hpp b/src/io/v3/transport.hpp index 93baaf131..348529e74 100644 --- a/src/io/v3/transport.hpp +++ b/src/io/v3/transport.hpp @@ -61,43 +61,54 @@ class Io { public: Io(I io, Address address) : implementation_(io), address_(address) {} + /// Set the default time-out for all requests that are issued + /// without an explicit time-out set. void SetDefaultTimeoutMicroseconds(uint64_t timeout_microseconds) { default_timeout_microseconds_ = timeout_microseconds; } + /// Issue a request with an explicit time-out in microseconds provided. template - ResponseFuture RequestTimeout(Address address, Request request, uint64_t timeout_microseconds) { + ResponseFuture RequestWithTimeout(Address address, Request request, uint64_t timeout_microseconds) { uint64_t request_id = ++request_id_counter_; - return implementation_.template RequestTimeout(address, request_id, request, - timeout_microseconds); + return implementation_.template Request(address, request_id, request, timeout_microseconds); } + /// Issue a request that times out after the default timeout. template - ResponseFuture RequestTimeout(Address address, Request request) { + ResponseFuture Request(Address address, Request request) { uint64_t request_id = ++request_id_counter_; uint64_t timeout_microseconds = default_timeout_microseconds_; - return implementation_.template RequestTimeout(address, request_id, request, - timeout_microseconds); + return implementation_.template Request(address, request_id, request, timeout_microseconds); } + /// Wait for an explicit number of microseconds for a request of one of the + /// provided types to arrive. template - RequestResult ReceiveTimeout(uint64_t timeout_microseconds) { - return implementation_.template ReceiveTimeout(timeout_microseconds); + RequestResult ReceiveWithTimeout(uint64_t timeout_microseconds) { + return implementation_.template Receive(timeout_microseconds); } + /// Wait the default number of microseconds for a request of one of the + /// provided types to arrive. template - RequestResult ReceiveTimeout() { + RequestResult Receive() { uint64_t timeout_microseconds = default_timeout_microseconds_; - return implementation_.template ReceiveTimeout(timeout_microseconds); + return implementation_.template Receive(timeout_microseconds); } + /// 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. template void Send(Address address, uint64_t request_id, M message) { return implementation_.template Send(address, request_id, message); } + /// The current system time. This time source should be preferred over + /// any other time source. std::time_t Now() { return implementation_.Now(); } + /// Returns true of the system should shut-down. bool ShouldShutDown() { return implementation_.ShouldShutDown(); } private: diff --git a/tests/simulation/basic_request.cpp b/tests/simulation/basic_request.cpp index d7f47af39..b4dd7b532 100644 --- a/tests/simulation/basic_request.cpp +++ b/tests/simulation/basic_request.cpp @@ -19,20 +19,20 @@ #include "io/v3/simulator.hpp" #include "utils/terminate_handler.hpp" -struct Request { +struct RequestMsg { std::string data; std::vector Serialize() { return std::vector(); } - static Request Deserialize(uint8_t *ptr, size_t len) { return Request{}; } + static RequestMsg Deserialize(uint8_t *ptr, size_t len) { return RequestMsg{}; } }; -struct Response { +struct ResponseMsg { std::string data; std::vector Serialize() { return std::vector(); } - static Response Deserialize(uint8_t *ptr, size_t len) { return Response{}; } + static ResponseMsg Deserialize(uint8_t *ptr, size_t len) { return ResponseMsg{}; } }; int main() { @@ -41,18 +41,18 @@ int main() { auto srv_addr = Address::TestAddress(2); Io cli_io = simulator.Register(cli_addr, true); - // Io srv_io = simulator.Register(srv_addr, true); + Io srv_io = simulator.Register(srv_addr, true); // send request - Request cli_req; - ResponseFuture response_future = cli_io.template RequestTimeout(srv_addr, cli_req); + RequestMsg cli_req; + ResponseFuture response_future = cli_io.template Request(srv_addr, cli_req); // // receive request // RequestResult request_result = sim_io_2.template ReceiveTimeout(); // auto req_envelope = request_result.GetValue(); // Request req = std::get(req_envelope.message); // - // auto srv_res = Response{req.data}; + // auto srv_res = ResponseMsg{req.data}; // // // send response // sim_io_2.Send(req_envelope.from, req_envelope.request_id, srv_res);