diff options
Diffstat (limited to 'modules')
| -rw-r--r-- | modules/remote/c++/remote.hpp | 8 | ||||
| -rw-r--r-- | modules/remote/c++/remote_loopback.hpp | 196 | ||||
| -rw-r--r-- | modules/remote/c++/remote_loopback_base.hpp | 36 | ||||
| -rw-r--r-- | modules/remote/c++/transfer_loopback.hpp | 52 | ||||
| -rw-r--r-- | 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<typename... T, typename Encoding, typename Remote> struct tmpl_group_to_data_server_ptr_std_tuple<tmpl_group<T...>, Encoding, Remote> { using type = std::tuple<ptr<data_server<T,Encoding,Remote>>...>; }; + +template<typename T, typename Encoding, typename Remote> +struct tmpl_group_to_data_client_std_tuple; + +template<typename... T, typename Encoding, typename Remote> +struct tmpl_group_to_data_client_std_tuple<tmpl_group<T...>, Encoding, Remote> { + using type = std::tuple<data_client<T,Encoding,Remote>...>; +}; } template<typename Remote> 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 <tuple> #include <variant> #include <forstio/codec/interface.hpp> @@ -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<typename T, typename Encoding> class remote_data<T, Encoding, rmt::Loopback> { private: id<T> id_; + ptr<data_server<T, Encoding, rmt::Loopback>> srv_; public: - remote_data(const id<T>& id): - id_{id} + remote_data(const id<T>& id__, ptr<data_server<T, Encoding, rmt::Loopback>> srv__): + id_{id__}, + srv_{srv__} {} /** - * Wait until data arrives + * Id of the data on the remote. + */ + const id<T>& 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<data<T, Encoding>> wait(wait_scope& wait); + error_or<data<T, Encoding>> wait(wait_scope& wait){ + (void) wait; + return srv_().receive(id_); + } /** * Asynchronously wait for a result */ - conveyor<data<T, Encoding>> on_receive(); + conveyor<data<T, Encoding>> 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<typename Iface, typename Encoding> class rpc_client<Iface, Encoding, rmt::Loopback> { +public: + using ServerT = rpc_server<Iface, Encoding, rmt::Loopback>; + using TypeGroup = typename schema_iface_type_group<Iface>::Type; + using DataClients = typename impl::tmpl_group_to_data_client_std_tuple<TypeGroup, Encoding, rmt::Loopback>::type; +private: + ptr<ServerT> 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<ServerT> 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<typename IdT> + data_client<IdT, Encoding, rmt::Loopback>& get_data_client(){ + return std::get<data_client<IdT, Encoding, rmt::Loopback>>(dat_clients_); + } + /** - * request the data from the remote + * Request the data from the remote */ template<typename IdT> - remote_data<IdT, Encoding, rmt::Loopback> request_data(id<IdT> data); - - /** @todo - * Determine type based on Name + remote_data<IdT, Encoding, rmt::Loopback> request_data(id<IdT> dat_id){ + return {dat_id, std::get<ptr<data_server<IdT, Encoding, rmt::Loopback>>>(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<string_literal Name> error_or< - id< - typename schema_member_type<Name, Iface>::type - > - > call(data_or_id<Input> inp); - */ + remote_data<typename schema_member_type<Name, Iface>::type::ResponseT, Encoding, rmt::Loopback> + > call(data_or_id<typename schema_member_type<Name, Iface>::type::RequestT, Encoding> input){ + using Req = typename schema_member_type<Name, Iface>::type::RequestT; + using Resp = typename schema_member_type<Name, Iface>::type::ResponseT; + + auto& req_client = get_data_client<Req>(); + auto& resp_client = get_data_client<Resp>(); + + /** + * Resolve the parameter to an id on the remote. + */ + auto eo_req_id = [&]() -> error_or<id<Req>> { + 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<Name>(eo_req_id.get_value(), resp_id); + if(eov.is_error()){ + return std::move(eov.get_error()); + } + } + + return request_data<Resp>(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<typename Iface, typename Encode> -class rpc_server<Iface, Encode, rmt::Loopback> { +class rpc_server<Iface, Encode, rmt::Loopback> final : public i_rpc_server<rmt::Loopback> { public: using InterfaceT = interface<Iface, Encode>; using TypeGroup = typename schema_iface_type_group<Iface>::Type; using DataServers = typename impl::tmpl_group_to_data_server_ptr_std_tuple<TypeGroup, Encode, rmt::Loopback>::type; + + /** + * Identifies the interface this server provides. + */ + static constexpr std::pair<uint32_t,uint32_t> class_id{schema_hash<Iface>::apply(), schema_hash<Encode>::apply()}; private: ptr<remote<rmt::Loopback>> remote_; ptr<remote_address<rmt::Loopback>> 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<uint32_t,uint32_t> 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<string_literal Lit> error_or<void> call(id<typename schema_member_type<Lit, Iface>::type::RequestT> id_param, id<typename schema_member_type<Lit,Iface>::type::ResponseT> id_resp){ @@ -103,4 +241,24 @@ public: } }; +template<typename Iface, typename Encode> +conveyor<rpc_client<Iface, Encode, rmt::Loopback>> remote<rmt::Loopback>::rpc_connect(const remote_address<rmt::Loopback>& addr){ + constexpr auto class_id = rpc_server<Iface, Encode, rmt::Loopback>::class_id; + key_t key; + key.data = std::array<uint64_t,3>{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<err::not_found>("Rpc server not found"); + } + + auto eo_rpc_srv = find->second().template cast_to<rpc_server<Iface, Encode, rmt::Loopback>>(); + if(eo_rpc_srv.is_error()){ + auto& err = eo_rpc_srv.get_error(); + return std::move(err); + } + + return rpc_client<Iface, Encode, rmt::Loopback>{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<typename Iface, typename Encode> conveyor<rpc_client<Iface, Encode, rmt::Loopback>> rpc_connect(const remote_address<rmt::Loopback>& addr); @@ -175,9 +178,9 @@ public: auto class_id = srv.get_class_id(); key_t key; key.data = std::array<uint64_t,3>{addr.get_address_id().get(), class_id.first, class_id.second}; - + auto insert = registered_data_servers_.emplace(std::make_pair(key, ptr<i_data_server<rmt::Loopback>>{srv})); - if(insert.second){ + if(!insert.second){ return make_error<err::already_exists>("Server already bound to this address"); } @@ -195,5 +198,34 @@ public: return make_void(); } + + /** + * Internal function + */ + error_or<void> register_rpc_server(const remote_address<rmt::Loopback>& addr, i_rpc_server<rmt::Loopback>& srv){ + auto class_id = srv.get_class_id(); + key_t key; + key.data = std::array<uint64_t,3>{addr.get_address_id().get(), class_id.first, class_id.second}; + + auto insert = registered_rpc_servers_.emplace(std::make_pair(key, ptr<i_rpc_server<rmt::Loopback>>{srv})); + if(!insert.second){ + return make_error<err::already_exists>("Server already bound to this address"); + } + + return make_void(); + } + + error_or<void> deregister_rpc_server(const remote_address<rmt::Loopback>& addr, i_rpc_server<rmt::Loopback>& srv){ + auto class_id = srv.get_class_id(); + key_t key; + key.data = std::array<uint64_t,3>{addr.get_address_id().get(), class_id.first, class_id.second}; + + auto erase = registered_rpc_servers_.erase(key); + if(erase == 0u){ + return make_error<err::not_found>("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<rmt::Loopback>; template<typename Schema, typename Encoding> class data_server<Schema, Encoding, rmt::Loopback> final : public i_data_server<rmt::Loopback> { private: - //typename + //typename using type = std::unordered_map<uint64_t, data<Schema, Encoding>>; type values_; ptr<remote<rmt::Loopback>> remote_; remote_address<rmt::Loopback> 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<uint32_t,uint32_t> class_id{schema_hash<Schema>::apply(), schema_hash<Encoding>::apply()}; @@ -42,7 +48,20 @@ public: std::pair<uint32_t,uint32_t> 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<Schema> next_free_id(){ + while(values_.find(next_id_) != values_.end()){ + ++next_id_; + } + + return id<Schema>{next_id_}; + } + error_or<void> send(ref<data<Schema, Encoding>> dat, id<Schema> store_id){ try { auto insert_res = values_.emplace(std::make_pair(store_id.get_value(), dat())); @@ -233,40 +252,49 @@ template<typename Schema, typename Encoding> class data_client<Schema, Encoding, rmt::Loopback> final { private: ptr<data_server<Schema, Encoding, rmt::Loopback>> srv_; - - uint64_t next_id_; public: data_client(ptr<data_server<Schema, Encoding, rmt::Loopback>> srv__): - srv_{srv__}, - next_id_{0u} + srv_{srv__} {} - + + /** + * Ask the remote for an id which is not in use yet. + */ + error_or<id<Schema>> next_free_id(){ + return srv_().next_free_id(); + } + /** * Send data to the remote. */ error_or<id<Schema>> send(const data<Schema, Encoding>& dat){ - id<Schema> dat_id{next_id_}; + id<Schema> 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<void> send(const data<Schema, Encoding>& dat, id<Schema> dat_id){ + return srv_().send(dat, dat_id); + } + /** * Preallocate data */ error_or<id<Schema>> allocate(const data<typename meta_schema<Schema>::MetaSchema, Encoding>& meta){ - id<Schema> dat_id{next_id_}; + id<Schema> 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::Loopback> 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<sch::UInt32, encode::Native>(*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<sch::Int64, encode::Native>(*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<sch::UInt32, encode::Native> input) -> error_or<data<sch::Int64, encode::Native>> { + data<sch::Int64, encode::Native> rv; + rv.set(input.get() * 2); + return rv; + }; + + auto iface = interface_factory<sch::TestInterface, encode::Native>::create(std::move(foo_func)); + + auto eo_rpc_srv = rmt.template rpc_listen<sch::TestInterface, encode::Native>(*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<sch::TestInterface, encode::Native>(*addr); + auto eo_client = cvr_client.take(); + SAW_EXPECT(eo_client.is_value(), "Couldn't connect."); + auto& client = eo_client.get_value(); + + data<sch::UInt32, encode::Native> 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; |