Commit 22b1bc45 authored by Dominik Charousset's avatar Dominik Charousset

Streamline API for remote spawning

parent 80c56b4f
...@@ -7,7 +7,8 @@ ...@@ -7,7 +7,8 @@
// Run client at the same host: // Run client at the same host:
// - remote_spawn -n localhost -p 4242 // - remote_spawn -n localhost -p 4242
// Manual refs: 250-262 () // Manual refs: 137-139, 144, 148 (ConfiguringActorApplications)
// 101-119 (RemoteSpawn)
#include <array> #include <array>
#include <vector> #include <vector>
...@@ -36,7 +37,6 @@ using calculator = typed_actor<replies_to<add_atom, int, int>::with<int>, ...@@ -36,7 +37,6 @@ using calculator = typed_actor<replies_to<add_atom, int, int>::with<int>,
replies_to<sub_atom, int, int>::with<int>>; replies_to<sub_atom, int, int>::with<int>>;
// function-based, statically typed, event-based API
calculator::behavior_type calculator_fun(calculator::pointer self) { calculator::behavior_type calculator_fun(calculator::pointer self) {
return { return {
[=](add_atom, int a, int b) -> int { [=](add_atom, int a, int b) -> int {
...@@ -108,7 +108,8 @@ void client(actor_system& system, const std::string& host, uint16_t port) { ...@@ -108,7 +108,8 @@ void client(actor_system& system, const std::string& host, uint16_t port) {
auto type = "calculator"; // type of the actor we wish to spawn auto type = "calculator"; // type of the actor we wish to spawn
auto args = make_message(); // arguments to construct the actor auto args = make_message(); // arguments to construct the actor
auto tout = std::chrono::seconds(30); // wait no longer than 30s auto tout = std::chrono::seconds(30); // wait no longer than 30s
auto worker = system.middleman().remote_spawn<calculator>(*node, type, args, tout); auto worker = system.middleman().remote_spawn<calculator>(*node, type,
args, tout);
if (! worker) { if (! worker) {
std::cerr << "*** remote spawn failed: " std::cerr << "*** remote spawn failed: "
<< system.render(worker.error()) << std::endl; << system.render(worker.error()) << std::endl;
...@@ -130,15 +131,10 @@ void server(actor_system& system, uint16_t port) { ...@@ -130,15 +131,10 @@ void server(actor_system& system, uint16_t port) {
std::cout << "*** running on port: " std::cout << "*** running on port: "
<< *res << std::endl << *res << std::endl
<< "*** press <enter> to shutdown server" << std::endl; << "*** press <enter> to shutdown server" << std::endl;
int dummy; getchar();
std::cin >> dummy;
} }
struct config : actor_system_config { struct config : actor_system_config {
uint16_t port = 0;
std::string host = "localhost";
bool server_mode = false;
config() { config() {
add_actor_type("calculator", calculator_fun); add_actor_type("calculator", calculator_fun);
opt_group{custom_options_, "global"} opt_group{custom_options_, "global"}
...@@ -146,6 +142,9 @@ struct config : actor_system_config { ...@@ -146,6 +142,9 @@ struct config : actor_system_config {
.add(host, "host,H", "set node (ignored in server mode)") .add(host, "host,H", "set node (ignored in server mode)")
.add(server_mode, "server-mode,s", "enable server mode"); .add(server_mode, "server-mode,s", "enable server mode");
} }
uint16_t port = 0;
std::string host = "localhost";
bool server_mode = false;
}; };
void caf_main(actor_system& system, const config& cfg) { void caf_main(actor_system& system, const config& cfg) {
......
...@@ -179,6 +179,10 @@ public: ...@@ -179,6 +179,10 @@ public:
return mpi{}; return mpi{};
} }
inline mpi message_types(detail::type_list<strong_actor_ptr>) const {
return mpi{};
}
template <class... Ts> template <class... Ts>
mpi message_types(detail::type_list<typed_actor<Ts...>>) const { mpi message_types(detail::type_list<typed_actor<Ts...>>) const {
static_assert(sizeof...(Ts) > 0, "empty typed actor handle given"); static_assert(sizeof...(Ts) > 0, "empty typed actor handle given");
...@@ -336,6 +340,26 @@ public: ...@@ -336,6 +340,26 @@ public:
return spawn_functor<Os>(cfg, fun, std::forward<Ts>(xs)...); return spawn_functor<Os>(cfg, fun, std::forward<Ts>(xs)...);
} }
/// Returns a new actor with run-time type `name`, constructed
/// with the arguments stored in `args`.
/// @experimental
template <class Handle,
class E = typename std::enable_if<is_handle<Handle>::value>::type>
expected<Handle> spawn(const std::string& name, message args,
execution_unit* ctx = nullptr,
bool check_interface = true,
const mpi* expected_ifs = nullptr) {
mpi tmp;
if (check_interface && ! expected_ifs) {
tmp = message_types<Handle>();
expected_ifs = &tmp;
}
auto res = dyn_spawn_impl(name, args, ctx, check_interface, expected_ifs);
if (! res)
return std::move(res.error());
return actor_cast<Handle>(std::move(*res));
}
template <class T, spawn_options Os = no_spawn_options, template <class T, spawn_options Os = no_spawn_options,
class Iter, class F, class... Ts> class Iter, class F, class... Ts>
infer_handle_from_class_t<T> infer_handle_from_class_t<T>
...@@ -445,6 +469,12 @@ private: ...@@ -445,6 +469,12 @@ private:
"Probably you have tried to spawn a broker or opencl actor."); "Probably you have tried to spawn a broker or opencl actor.");
} }
expected<strong_actor_ptr> dyn_spawn_impl(const std::string& name,
message& args,
execution_unit* ctx,
bool check_interface,
optional<const mpi&> expected_ifs);
template <class C, spawn_options Os, class... Ts> template <class C, spawn_options Os, class... Ts>
infer_handle_from_class_t<C> infer_handle_from_class_t<C>
spawn_impl(actor_config& cfg, Ts&&... xs) { spawn_impl(actor_config& cfg, Ts&&... xs) {
......
...@@ -259,8 +259,8 @@ bool operator!=(std::nullptr_t x, const function_view<T>& y) { ...@@ -259,8 +259,8 @@ bool operator!=(std::nullptr_t x, const function_view<T>& y) {
/// @relates function_view /// @relates function_view
/// @experimental /// @experimental
template <class T> template <class T>
function_view<T> make_function_view(const T& x) { function_view<T> make_function_view(const T& x, duration t = infinite) {
return {x}; return {x, t};
} }
} // namespace caf } // namespace caf
......
...@@ -209,6 +209,18 @@ struct infer_handle_from_state<T, false> { ...@@ -209,6 +209,18 @@ struct infer_handle_from_state<T, false> {
template <class T> template <class T>
using infer_handle_from_state_t = typename infer_handle_from_state<T>::type; using infer_handle_from_state_t = typename infer_handle_from_state<T>::type;
template <class T>
struct is_handle : std::false_type {};
template <>
struct is_handle<actor> : std::true_type {};
template <>
struct is_handle<strong_actor_ptr> : std::true_type {};
template <class... Ts>
struct is_handle<typed_actor<Ts...>> : std::true_type {};
} // namespace caf } // namespace caf
#endif // CAF_INFER_HANDLE_HPP #endif // CAF_INFER_HANDLE_HPP
...@@ -44,7 +44,7 @@ static constexpr none_t none = none_t{}; ...@@ -44,7 +44,7 @@ static constexpr none_t none = none_t{};
/// @relates none_t /// @relates none_t
inline std::string to_string(const none_t&) { inline std::string to_string(const none_t&) {
return "<none>"; return "none";
} }
} // namespace caf } // namespace caf
......
...@@ -176,6 +176,10 @@ class optional<T&> { ...@@ -176,6 +176,10 @@ class optional<T&> {
// nop // nop
} }
optional(T* x) : m_value(x) {
// nop
}
optional(const optional& other) = default; optional(const optional& other) = default;
optional& operator=(const optional& other) = default; optional& operator=(const optional& other) = default;
......
...@@ -75,10 +75,6 @@ public: ...@@ -75,10 +75,6 @@ public:
type_erased_value_ptr make_value(const std::type_info& ti) const; type_erased_value_ptr make_value(const std::type_info& ti) const;
actor_factory_result make_actor(const std::string& name,
actor_config& cfg,
message& msg) const;
/// Returns the portable name for given type information or `nullptr` /// Returns the portable name for given type information or `nullptr`
/// if no mapping was found. /// if no mapping was found.
const std::string* portable_name(uint16_t nr, const std::type_info* ti) const; const std::string* portable_name(uint16_t nr, const std::type_info* ti) const;
...@@ -90,8 +86,6 @@ public: ...@@ -90,8 +86,6 @@ public:
return portable_name(x.first, x.second); return portable_name(x.first, x.second);
} }
error_renderer renderer(atom_value x) const;
/// Returns the enclosing actor system. /// Returns the enclosing actor system.
inline actor_system& system() const { inline actor_system& system() const {
return system_; return system_;
......
...@@ -147,15 +147,12 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) { ...@@ -147,15 +147,12 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) {
behavior spawn_serv_impl(event_based_actor* self) { behavior spawn_serv_impl(event_based_actor* self) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
return { return {
[=](spawn_atom, const std::string& name, message& args) [=](spawn_atom, const std::string& name,
-> result<strong_actor_ptr, std::set<std::string>> { message& args, actor_system::mpi& xs)
-> expected<strong_actor_ptr> {
CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(args)); CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(args));
printf("%s %d -- %s\n", __FILE__, __LINE__, to_string(self->current_mailbox_element()->content()).c_str()); return self->system().spawn<strong_actor_ptr>(name, std::move(args),
actor_config cfg{self->context()}; self->context(), true, &xs);
auto res = self->system().types().make_actor(name, cfg, args);
if (! res.first)
return sec::cannot_spawn_actor_from_arguments;
return {res.first, res.second};
} }
}; };
} }
...@@ -307,8 +304,13 @@ const uniform_type_info_map& actor_system::types() const { ...@@ -307,8 +304,13 @@ const uniform_type_info_map& actor_system::types() const {
} }
std::string actor_system::render(const error& x) const { std::string actor_system::render(const error& x) const {
auto f = types().renderer(x.category()); if (! x)
return f ? f(x.code(), x.category(), x.context()) : to_string(x); return to_string(x);
auto& xs = config().error_renderers;
auto i = xs.find(x.category());
if (i != xs.end())
return i->second(x.code(), x.category(), x.context());
return to_string(x);
} }
group_manager& actor_system::groups() { group_manager& actor_system::groups() {
...@@ -368,4 +370,25 @@ void actor_system::await_detached_threads() { ...@@ -368,4 +370,25 @@ void actor_system::await_detached_threads() {
detached_cv.wait(guard); detached_cv.wait(guard);
} }
expected<strong_actor_ptr>
actor_system::dyn_spawn_impl(const std::string& name, message& args,
execution_unit* ctx, bool check_interface,
optional<const mpi&> expected_ifs) {
CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(args) << CAF_ARG(check_interface)
<< CAF_ARG(expected_ifs));
if (name.empty())
return sec::invalid_argument;
auto& fs = cfg_.actor_factories;
auto i = fs.find(name);
if (i == fs.end())
return sec::unknown_type;
actor_config cfg{ctx ? ctx : &dummy_execution_unit_};
auto res = i->second(cfg, args);
if (! res.first)
return sec::cannot_spawn_actor_from_arguments;
if (check_interface && !assignable(res.second, *expected_ifs))
return sec::unexpected_actor_messaging_interface;
return std::move(res.first);
}
} // namespace caf } // namespace caf
...@@ -154,27 +154,6 @@ uniform_type_info_map::portable_name(uint16_t nr, ...@@ -154,27 +154,6 @@ uniform_type_info_map::portable_name(uint16_t nr,
return nullptr; return nullptr;
} }
uniform_type_info_map::error_renderer
uniform_type_info_map::renderer(atom_value x) const {
auto& error_renderers = system().config().error_renderers;
auto i = error_renderers.find(x);
if (i != error_renderers.end())
return i->second;
return nullptr;
}
actor_factory_result uniform_type_info_map::make_actor(const std::string& name,
actor_config& cfg,
message& msg) const {
strong_actor_ptr res;
std::set<std::string> ifs;
auto& factories = system().config().actor_factories;
auto i = factories.find(name);
if (i != factories.end())
std::tie(res, ifs) = i->second(cfg, msg);
return std::make_pair(std::move(res), std::move(ifs));
}
uniform_type_info_map::uniform_type_info_map(actor_system& sys) : system_(sys) { uniform_type_info_map::uniform_type_info_map(actor_system& sys) : system_(sys) {
sorted_builtin_types list; sorted_builtin_types list;
fill_builtins(builtin_, list, 0); fill_builtins(builtin_, list, 0);
......
...@@ -45,15 +45,13 @@ struct fixture { ...@@ -45,15 +45,13 @@ struct fixture {
std::set<std::string> ifs; std::set<std::string> ifs;
scoped_execution_unit context{&system}; scoped_execution_unit context{&system};
actor_config actor_cfg{&context}; actor_config actor_cfg{&context};
std::tie(res, ifs) = system.types().make_actor("test_actor", actor_cfg, args); auto aut = system.spawn<actor>("test_actor", std::move(args));
if (expect_fail) { if (expect_fail) {
CAF_REQUIRE(! res); CAF_REQUIRE(! aut);
return; return;
} }
CAF_REQUIRE(res); CAF_REQUIRE(aut);
CAF_CHECK(ifs.empty()); self->wait_for(*aut);
auto aut = actor_cast<actor>(res);
self->wait_for(aut);
CAF_MESSAGE("aut done"); CAF_MESSAGE("aut done");
} }
}; };
......
...@@ -175,17 +175,11 @@ public: ...@@ -175,17 +175,11 @@ public:
duration timeout = std::chrono::seconds(60)) { duration timeout = std::chrono::seconds(60)) {
if (! nid || name.empty()) if (! nid || name.empty())
return sec::invalid_argument; return sec::invalid_argument;
auto res = remote_spawn_impl(nid, name, args, timeout); auto res = remote_spawn_impl(nid, name, args,
system().message_types<Handle>(), timeout);
if (! res) if (! res)
return std::move(res.error()); return std::move(res.error());
if (! system().assignable<Handle>(res->second)) { return actor_cast<Handle>(std::move(*res));
// kill remote actor immediately and return error on interfaces mismatch
anon_send_exit(res->first, exit_reason::kill);
return make_error(sec::unexpected_actor_messaging_interface,
system().message_types<Handle>(),
std::move(res->second));
}
return actor_cast<Handle>(res->first);
} }
/// Smart pointer for `network::multiplexer`. /// Smart pointer for `network::multiplexer`.
...@@ -297,9 +291,10 @@ private: ...@@ -297,9 +291,10 @@ private:
return system().spawn_class<Impl, Os>(cfg); return system().spawn_class<Impl, Os>(cfg);
} }
expected<std::pair<strong_actor_ptr, std::set<std::string>>> expected<strong_actor_ptr> remote_spawn_impl(const node_id& nid,
remote_spawn_impl(const node_id& nid, std::string& name, std::string& name, message& args,
message& args, duration timeout); std::set<std::string> ifs,
duration timeout);
expected<uint16_t> publish(const strong_actor_ptr& whom, expected<uint16_t> publish(const strong_actor_ptr& whom,
std::set<std::string> sigs, std::set<std::string> sigs,
......
...@@ -102,8 +102,8 @@ using middleman_actor = ...@@ -102,8 +102,8 @@ using middleman_actor =
reacts_to<close_atom, uint16_t>, reacts_to<close_atom, uint16_t>,
replies_to<spawn_atom, node_id, std::string, message> replies_to<spawn_atom, node_id, std::string, message, std::set<std::string>>
::with<strong_actor_ptr, std::set<std::string>>, ::with<strong_actor_ptr>,
replies_to<get_atom, node_id>::with<node_id, std::string, uint16_t>>; replies_to<get_atom, node_id>::with<node_id, std::string, uint16_t>>;
......
...@@ -107,17 +107,14 @@ middleman::middleman(actor_system& sys) ...@@ -107,17 +107,14 @@ middleman::middleman(actor_system& sys)
// nop // nop
} }
expected<std::pair<strong_actor_ptr, std::set<std::string>>> expected<strong_actor_ptr> middleman::remote_spawn_impl(const node_id& nid,
middleman::remote_spawn_impl(const node_id& nid, std::string& name, std::string& name,
message& args, duration timeout) { message& args,
auto f = make_function_view(actor_handle()); std::set<std::string> s,
f.timeout = timeout; duration timeout) {
auto res = f(spawn_atom::value, nid, std::move(name), std::move(args)); auto f = make_function_view(actor_handle(), timeout);
// why can't we assign a tuple<T1, T2> to pair<T1, T2>? No one knows! return f(spawn_atom::value, nid, std::move(name),
if (! res) std::move(args), std::move(s));
return std::move(res.error());
return std::make_pair(std::move(std::get<0>(*res)),
std::move(std::get<1>(*res)));
} }
expected<uint16_t> middleman::open(uint16_t port, const char* cstr, bool ru) { expected<uint16_t> middleman::open(uint16_t port, const char* cstr, bool ru) {
......
...@@ -160,12 +160,13 @@ public: ...@@ -160,12 +160,13 @@ public:
delegate(broker_, atm, p); delegate(broker_, atm, p);
return {}; return {};
}, },
[=](spawn_atom atm, node_id& nid, std::string& str, message& msg) [=](spawn_atom atm, node_id& nid, std::string& str,
-> delegated<strong_actor_ptr, mpi_set> { message& msg, std::set<std::string>& ifs)
-> delegated<strong_actor_ptr> {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
delegate(broker_, forward_atom::value, nid, atom("SpawnServ"), delegate(broker_, forward_atom::value, nid, atom("SpawnServ"),
make_message(atm, std::move(str), std::move(msg))); make_message(atm, std::move(str),
//delegate(broker_, atm, std::move(nid), std::move(str), std::move(msg)); std::move(msg), std::move(ifs)));
return {}; return {};
}, },
[=](get_atom atm, node_id nid) [=](get_atom atm, node_id nid)
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment