diff --git a/src/io/v3/address.hpp b/src/io/v3/address.hpp index 4738d3042..4e1a80c75 100644 --- a/src/io/v3/address.hpp +++ b/src/io/v3/address.hpp @@ -11,6 +11,8 @@ #pragma once +#include + #include #include @@ -31,4 +33,16 @@ struct Address { ret.last_known_port = port; return ret; } + + bool operator==(const Address &other) const { + return (last_known_ip == other.last_known_ip) && (last_known_port == other.last_known_port); + } + + bool operator<(const Address &other) const { + 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; + } + } }; diff --git a/src/io/v3/future.hpp b/src/io/v3/future.hpp index 48fc1f9bc..4b0e188db 100644 --- a/src/io/v3/future.hpp +++ b/src/io/v3/future.hpp @@ -159,6 +159,8 @@ class MgPromise { friend std::pair, MgPromise> FuturePromisePairWithNotifier(std::function); public: + MgPromise(std::shared_ptr> shared) : shared_(shared) {} + MgPromise(MgPromise &&old) { shared_ = std::move(old.shared_); MG_ASSERT(!old.filled_or_moved_, "MgPromise moved from after already being moved from or filled."); @@ -183,9 +185,16 @@ class MgPromise { bool IsAwaited() { return shared_->IsAwaited(); } - private: - MgPromise(std::shared_ptr> shared) : shared_(shared) {} + /// Moves this MgPromise into a unique_ptr. + std::unique_ptr> ToUnique() && { + std::unique_ptr> up = std::make_unique>(std::move(shared_)); + filled_or_moved_ = true; + + return up; + } + + private: std::shared_ptr> shared_; bool filled_or_moved_ = false; }; diff --git a/src/io/v3/simulator.hpp b/src/io/v3/simulator.hpp index e5522c813..40fef63ef 100644 --- a/src/io/v3/simulator.hpp +++ b/src/io/v3/simulator.hpp @@ -41,12 +41,12 @@ class SimulatorTransport { return std::move(future); } - /* - template - RequestResult Receive(uint64_t timeout_microseconds) { - return simulator_handle_->template Receive(timeout_microseconds); - } + template + RequestResult Receive(uint64_t timeout_microseconds) { + return simulator_handle_->template Receive(timeout_microseconds); + } + /* template void Send(Address address, uint64_t request_id, M message) { return simulator_handle_->template Send(address, request_id, message); @@ -69,5 +69,5 @@ class Simulator { } private: - std::shared_ptr simulator_handle_; + std::shared_ptr simulator_handle_{std::make_shared()}; }; diff --git a/src/io/v3/simulator_handle.hpp b/src/io/v3/simulator_handle.hpp index 9e0e73d9b..e6b761fe7 100644 --- a/src/io/v3/simulator_handle.hpp +++ b/src/io/v3/simulator_handle.hpp @@ -11,6 +11,7 @@ #pragma once +#include #include #include @@ -28,11 +29,71 @@ struct PromiseKey { Address requester; uint64_t request_id; Address replier; + + public: + bool operator<(const PromiseKey &other) const { + if (requester == other.requester) { + return request_id < other.request_id; + } else { + return requester < other.requester; + } + } }; -struct OpaquePromise { +// TODO delete copy ctor & copy assignment operator if possible +class OpaquePromise { + public: + OpaquePromise(OpaquePromise &&old) : ti_(old.ti_) { + ptr_ = old.ptr_; + old.ptr_ = nullptr; + } + + OpaquePromise &operator=(OpaquePromise &&old) { + MG_ASSERT(this != &old); + + ptr_ = old.ptr_; + ti_ = old.ti_; + 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); + + MgPromise *ptr = static_cast *>(ptr_); + + ptr_ = nullptr; + + return std::unique_ptr(ptr); + } + + template + OpaquePromise(std::unique_ptr> promise) + : ti_(&typeid(T)), + ptr_((void *)promise.release()), + dtor_([](void *ptr) { static_cast *>(ptr)->~MgPromise(); }) {} + + ~OpaquePromise() { + if (nullptr != ptr_) { + dtor_(ptr_); + } + } + + private: + const std::type_info *ti_; + void *ptr_; + std::function dtor_; +}; + +struct DeadlineAndOpaquePromise { uint64_t deadline; - std::any promise; + OpaquePromise promise; }; class SimulatorHandle { @@ -58,12 +119,10 @@ class SimulatorHandle { 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))); - */ + OpaquePromise opaque_promise(std::move(promise).ToUnique()); + PromiseKey pk{.requester = from_addr, .request_id = request_id, .replier = to_addr}; + DeadlineAndOpaquePromise op{.deadline = deadline, .promise = std::move(opaque_promise)}; + promises_.emplace(std::move(pk), std::move(op)); stats_.total_messages_++; stats_.total_requests_++; @@ -71,11 +130,11 @@ class SimulatorHandle { return; } + template + RequestResult Receive(uint64_t timeout_microseconds) { + std::terminate(); + } /* - template - RequestResult Receive(uint64_t timeout_microseconds) { - std::abort(); - } template void Send(Address address, uint64_t request_id, M message) { @@ -84,11 +143,11 @@ class SimulatorHandle { */ private: - std::mutex mu_; + std::mutex mu_{}; std::condition_variable cv_sim_; std::condition_variable cv_srv_; std::vector> in_flight_; - std::map promises_; + std::map promises_; std::map can_receive_; uint64_t cluster_wide_time_microseconds_ = 0; bool shut_down_; diff --git a/src/io/v3/transport.hpp b/src/io/v3/transport.hpp index 348529e74..40f3cd125 100644 --- a/src/io/v3/transport.hpp +++ b/src/io/v3/transport.hpp @@ -9,6 +9,8 @@ // by the Apache License, Version 2.0, included in the file // licenses/APL.txt. +// TODO chrono::microseconds instead of std::time_t + #pragma once #include @@ -61,13 +63,13 @@ 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. + /// Set the default timeout for all requests that are issued + /// without an explicit timeout set. void SetDefaultTimeoutMicroseconds(uint64_t timeout_microseconds) { default_timeout_microseconds_ = timeout_microseconds; } - /// Issue a request with an explicit time-out in microseconds provided. + /// Issue a request with an explicit timeout in microseconds provided. template ResponseFuture RequestWithTimeout(Address address, Request request, uint64_t timeout_microseconds) { uint64_t request_id = ++request_id_counter_; diff --git a/tests/simulation/basic_request.cpp b/tests/simulation/basic_request.cpp index b4dd7b532..40a756a71 100644 --- a/tests/simulation/basic_request.cpp +++ b/tests/simulation/basic_request.cpp @@ -14,8 +14,6 @@ #include #include -#include - #include "io/v3/simulator.hpp" #include "utils/terminate_handler.hpp" @@ -47,8 +45,8 @@ int main() { RequestMsg cli_req; ResponseFuture response_future = cli_io.template Request(srv_addr, cli_req); - // // receive request - // RequestResult request_result = sim_io_2.template ReceiveTimeout(); + // receive request + RequestResult request_result = srv_io.template Receive(); // auto req_envelope = request_result.GetValue(); // Request req = std::get(req_envelope.message); //