From c90823bde9aea6b8f4c9737a0d8d5cd17560e8b9 Mon Sep 17 00:00:00 2001 From: "Claudius \"keldu\" Holeksa" Date: Sat, 7 Nov 2026 15:10:17 +0100 Subject: Finish Loopback --- modules/remote/c++/remote.hpp | 8 ++ modules/remote/c++/remote_loopback.hpp | 196 +++++++++++++++++++++++++--- modules/remote/c++/remote_loopback_base.hpp | 36 ++++- modules/remote/c++/transfer_loopback.hpp | 52 ++++++-- modules/remote/tests/remote_loopback.cpp | 45 +++++++ 5 files changed, 304 insertions(+), 33 deletions(-) diff --git a/modules/remote/c++/remote.hpp b/modules/remote/c++/remote.hpp index 7e846d3..83e6042 100644 --- a/modules/remote/c++/remote.hpp +++ b/modules/remote/c++/remote.hpp @@ -90,6 +90,14 @@ template struct tmpl_group_to_data_server_ptr_std_tuple, Encoding, Remote> { using type = std::tuple>...>; }; + +template +struct tmpl_group_to_data_client_std_tuple; + +template +struct tmpl_group_to_data_client_std_tuple, Encoding, Remote> { + using type = std::tuple...>; +}; } template class i_rpc_server { diff --git a/modules/remote/c++/remote_loopback.hpp b/modules/remote/c++/remote_loopback.hpp index 0d25641..d8cd7fa 100644 --- a/modules/remote/c++/remote_loopback.hpp +++ b/modules/remote/c++/remote_loopback.hpp @@ -1,5 +1,6 @@ #pragma once +#include #include #include @@ -9,56 +10,169 @@ namespace saw { +/** + * Handle for data which is stored on the remote. + * It is the result of an rpc call and resolves to the actual data by + * going through the data transfer layer again. + */ template class remote_data { private: id id_; + ptr> srv_; public: - remote_data(const id& id): - id_{id} + remote_data(const id& id__, ptr> srv__): + id_{id__}, + srv_{srv__} {} /** - * Wait until data arrives + * Id of the data on the remote. + */ + const id& get_id() const { + return id_; + } + + /** + * Wait until data arrives. + * The loopback backend answers calls inline, so the data is already + * present if the call succeeded. */ - error_or> wait(wait_scope& wait); + error_or> wait(wait_scope& wait){ + (void) wait; + return srv_().receive(id_); + } /** * Asynchronously wait for a result */ - conveyor> on_receive(); + conveyor> on_receive(){ + auto eov = srv_().receive(id_); + if(eov.is_error()){ + return std::move(eov.get_error()); + } + + return std::move(eov.get_value()); + } }; /** - * Client RPC reference structure + * Client RPC reference structure. + * + * This is the overlay on top of the data transfer layer. Parameters and + * return values are moved by the data clients/servers. The rpc layer only + * ties ids to a function call on the remote. */ template class rpc_client { +public: + using ServerT = rpc_server; + using TypeGroup = typename schema_iface_type_group::Type; + using DataClients = typename impl::tmpl_group_to_data_client_std_tuple::type; +private: + ptr srv_; + DataClients dat_clients_; + + /** + * Every data client is just a handle to the data server the rpc server + * is already bound to. + */ + static DataClients create_data_clients(const typename ServerT::DataServers& dat_srvs){ + return std::apply([](const auto&... srv){ + return DataClients{srv...}; + }, dat_srvs); + } +public: + rpc_client(ptr srv__): + srv_{srv__}, + dat_clients_{create_data_clients(srv__().get_data_servers())} + {} + + /** + * Access the data client of one of the interface types. + * Useful if parameters are meant to be reused in multiple calls. + */ + template + data_client& get_data_client(){ + return std::get>(dat_clients_); + } + /** - * request the data from the remote + * Request the data from the remote */ template - remote_data request_data(id data); - - /** @todo - * Determine type based on Name + remote_data request_data(id dat_id){ + return {dat_id, std::get>>(srv_().get_data_servers())}; + } + + /** + * Call a function on the remote. + * + * The parameter is either already present on the remote, in which case + * only its id is handed over, or it is sent through the data layer + * first. */ - /* template error_or< - id< - typename schema_member_type::type - > - > call(data_or_id inp); - */ + remote_data::type::ResponseT, Encoding, rmt::Loopback> + > call(data_or_id::type::RequestT, Encoding> input){ + using Req = typename schema_member_type::type::RequestT; + using Resp = typename schema_member_type::type::ResponseT; + + auto& req_client = get_data_client(); + auto& resp_client = get_data_client(); + + /** + * Resolve the parameter to an id on the remote. + */ + auto eo_req_id = [&]() -> error_or> { + if(input.is_id()){ + return input.get_id(); + } + + return req_client.send(input.get_data()); + }(); + if(eo_req_id.is_error()){ + return std::move(eo_req_id.get_error()); + } + + /** + * Reserve the slot the response is supposed to end up in. + */ + auto eo_resp_id = resp_client.next_free_id(); + if(eo_resp_id.is_error()){ + return std::move(eo_resp_id.get_error()); + } + auto resp_id = eo_resp_id.get_value(); + + { + auto eov = srv_().template call(eo_req_id.get_value(), resp_id); + if(eov.is_error()){ + return std::move(eov.get_error()); + } + } + + return request_data(resp_id); + } }; +/** + * Server side of the rpc overlay. + * It translates a call on an id to a call on the local interface by + * retrieving the parameter from the data server and storing the response in + * the data server again. + */ template -class rpc_server { +class rpc_server final : public i_rpc_server { public: using InterfaceT = interface; using TypeGroup = typename schema_iface_type_group::Type; using DataServers = typename impl::tmpl_group_to_data_server_ptr_std_tuple::type; + + /** + * Identifies the interface this server provides. + */ + static constexpr std::pair class_id{schema_hash::apply(), schema_hash::apply()}; private: ptr> remote_; ptr> addr_; @@ -71,7 +185,31 @@ public: addr_{addr__}, iface_{std::move(iface__)}, dat_srvs_{dat_srvs__} - {} + { + remote_().register_rpc_server(addr_(), *this); + } + + ~rpc_server(){ + remote_().deregister_rpc_server(addr_(), *this); + } + + SAW_FORBID_COPY(rpc_server); + SAW_FORBID_MOVE(rpc_server); + + /** + * Return the interface id + */ + std::pair get_class_id() const override { + return class_id; + } + + /** + * The data servers this rpc server moves its parameters and responses + * through. + */ + const DataServers& get_data_servers() const { + return dat_srvs_; + } template error_or call(id::type::RequestT> id_param, id::type::ResponseT> id_resp){ @@ -103,4 +241,24 @@ public: } }; +template +conveyor> remote::rpc_connect(const remote_address& addr){ + constexpr auto class_id = rpc_server::class_id; + key_t key; + key.data = std::array{addr.get_address_id().get(), class_id.first, class_id.second}; + + auto find = registered_rpc_servers_.find(key); + if(find == registered_rpc_servers_.end()){ + return make_error("Rpc server not found"); + } + + auto eo_rpc_srv = find->second().template cast_to>(); + if(eo_rpc_srv.is_error()){ + auto& err = eo_rpc_srv.get_error(); + return std::move(err); + } + + return rpc_client{eo_rpc_srv.get_value()}; +} + } diff --git a/modules/remote/c++/remote_loopback_base.hpp b/modules/remote/c++/remote_loopback_base.hpp index 5224d95..f2076c6 100644 --- a/modules/remote/c++/remote_loopback_base.hpp +++ b/modules/remote/c++/remote_loopback_base.hpp @@ -108,6 +108,9 @@ public: /** * Connect to a remote + * + * Defined in remote_loopback.hpp, since the rpc client needs to be + * a complete type at that point. */ template conveyor> rpc_connect(const remote_address& addr); @@ -175,9 +178,9 @@ public: auto class_id = srv.get_class_id(); key_t key; key.data = std::array{addr.get_address_id().get(), class_id.first, class_id.second}; - + auto insert = registered_data_servers_.emplace(std::make_pair(key, ptr>{srv})); - if(insert.second){ + if(!insert.second){ return make_error("Server already bound to this address"); } @@ -195,5 +198,34 @@ public: return make_void(); } + + /** + * Internal function + */ + error_or register_rpc_server(const remote_address& addr, i_rpc_server& srv){ + auto class_id = srv.get_class_id(); + key_t key; + key.data = std::array{addr.get_address_id().get(), class_id.first, class_id.second}; + + auto insert = registered_rpc_servers_.emplace(std::make_pair(key, ptr>{srv})); + if(!insert.second){ + return make_error("Server already bound to this address"); + } + + return make_void(); + } + + error_or deregister_rpc_server(const remote_address& addr, i_rpc_server& srv){ + auto class_id = srv.get_class_id(); + key_t key; + key.data = std::array{addr.get_address_id().get(), class_id.first, class_id.second}; + + auto erase = registered_rpc_servers_.erase(key); + if(erase == 0u){ + return make_error("Server not found"); + } + + return make_void(); + } }; } diff --git a/modules/remote/c++/transfer_loopback.hpp b/modules/remote/c++/transfer_loopback.hpp index e45abb3..1cd7e91 100644 --- a/modules/remote/c++/transfer_loopback.hpp +++ b/modules/remote/c++/transfer_loopback.hpp @@ -13,12 +13,18 @@ class remote; template class data_server final : public i_data_server { private: - //typename + //typename using type = std::unordered_map>; type values_; ptr> remote_; remote_address rmt_address_; + + /** + * Counter for handing out ids. Ids are owned by the server, since it is + * the only instance knowing about every client. + */ + uint64_t next_id_ = 0u; public: static constexpr std::pair class_id{schema_hash::apply(), schema_hash::apply()}; @@ -42,7 +48,20 @@ public: std::pair get_class_id() const override { return class_id; } - + + /** + * Hand out the next id which is not in use yet. + * The id is not reserved by this call. It is only in use as soon as + * something is stored under it. + */ + id next_free_id(){ + while(values_.find(next_id_) != values_.end()){ + ++next_id_; + } + + return id{next_id_}; + } + error_or send(ref> dat, id store_id){ try { auto insert_res = values_.emplace(std::make_pair(store_id.get_value(), dat())); @@ -233,40 +252,49 @@ template class data_client final { private: ptr> srv_; - - uint64_t next_id_; public: data_client(ptr> srv__): - srv_{srv__}, - next_id_{0u} + srv_{srv__} {} - + + /** + * Ask the remote for an id which is not in use yet. + */ + error_or> next_free_id(){ + return srv_().next_free_id(); + } + /** * Send data to the remote. */ error_or> send(const data& dat){ - id dat_id{next_id_}; + id dat_id = srv_().next_free_id(); auto eov = srv_().send(dat, dat_id); if(eov.is_error()){ auto& err = eov.get_error(); return std::move(err); } - ++next_id_; return dat_id; } - + + /** + * Send data to the remote and store it under a specific id. + */ + error_or send(const data& dat, id dat_id){ + return srv_().send(dat, dat_id); + } + /** * Preallocate data */ error_or> allocate(const data::MetaSchema, Encoding>& meta){ - id dat_id{next_id_}; + id dat_id = srv_().next_free_id(); auto eov = srv_().allocate(meta, dat_id); if(eov.is_error()){ auto& err = eov.get_error(); return std::move(err); } - ++next_id_; return dat_id; } diff --git a/modules/remote/tests/remote_loopback.cpp b/modules/remote/tests/remote_loopback.cpp index bd24384..dbdd0f1 100644 --- a/modules/remote/tests/remote_loopback.cpp +++ b/modules/remote/tests/remote_loopback.cpp @@ -65,6 +65,51 @@ SAW_TEST("Remote Loopback RPC"){ } } +SAW_TEST("Remote Loopback RPC Client"){ + using namespace saw; + + event_loop loop; + wait_scope wait{loop}; + + remote rmt; + + auto eov = rmt.parse_address(0u); + SAW_EXPECT(eov.is_value(), "Didn't parse correctly"); + auto& addr = eov.get_value(); + + auto eo_u32_srv = rmt.template data_listen(*addr); + SAW_EXPECT(eo_u32_srv.is_value(), std::string{"Couldn't listen: "} + std::string{eo_u32_srv.get_error().get_category()}); + + auto eo_i64_srv = rmt.template data_listen(*addr); + SAW_EXPECT(eo_i64_srv.is_value(), std::string{"Couldn't listen: "} + std::string{eo_i64_srv.get_error().get_category()}); + + auto foo_func = [](data input) -> error_or> { + data rv; + rv.set(input.get() * 2); + return rv; + }; + + auto iface = interface_factory::create(std::move(foo_func)); + + auto eo_rpc_srv = rmt.template rpc_listen(*addr, std::move(iface)); + SAW_EXPECT(eo_rpc_srv.is_value(), std::string{"Couldn't listen: "} + std::string{eo_rpc_srv.get_error().get_category()}); + + auto cvr_client = rmt.template rpc_connect(*addr); + auto eo_client = cvr_client.take(); + SAW_EXPECT(eo_client.is_value(), "Couldn't connect."); + auto& client = eo_client.get_value(); + + data inp{21}; + + auto eo_rmt_dat = client.template call<"foo">(std::move(inp)); + SAW_EXPECT(eo_rmt_dat.is_value(), (std::string{"Failed call: "} + std::string{eo_rmt_dat.get_error().get_category()})); + auto& rmt_dat = eo_rmt_dat.get_value(); + + auto eo_resp = rmt_dat.wait(wait); + SAW_EXPECT(eo_resp.is_value(), "Failed receive."); + SAW_EXPECT((eo_resp.get_value().get() == 42), (std::string{"Wrong value received: "} + std::to_string(eo_resp.get_value().get()))); +} + SAW_TEST("Remote Loopback Data"){ using namespace saw; -- cgit v1.3.1