Commit 29ab81f9 authored by Dominik Charousset's avatar Dominik Charousset

Implement remote spawn, close #486

parent a864e0a9
......@@ -47,69 +47,75 @@ template <class F, class T, class Bhvr, class R, class... Ts>
class fun_decorator<F, T, Bhvr, spawn_mode::function,
R, detail::type_list<Ts...>> {
public:
fun_decorator(const F& f, Bhvr& res, T*) : f_(f), res_(res) {
fun_decorator(const F& f, T*) : f_(f) {
// nop
}
void operator()(Ts... xs) {
behavior operator()(Ts... xs) {
detail::type_list<R> token;
apply(token, xs...);
return apply(token, xs...);
}
template <class U>
typename std::enable_if<
std::is_convertible<U, Bhvr>::value
std::is_convertible<U, Bhvr>::value,
behavior
>::type
apply(detail::type_list<U>, Ts... xs) {
res_ = f_(xs...);
auto bhvr = f_(xs...);
return std::move(bhvr.unbox());
}
template <class U>
typename std::enable_if<
! std::is_convertible<U, Bhvr>::value
! std::is_convertible<U, Bhvr>::value,
behavior
>::type
apply(detail::type_list<U>, Ts... xs) {
f_(xs...);
return {};
}
private:
F f_;
Bhvr& res_;
};
template <class F, class T, class Bhvr, class R, class... Ts>
class fun_decorator<F, T, Bhvr, spawn_mode::function_with_selfptr,
R, detail::type_list<T*, Ts...>> {
public:
fun_decorator(const F& f, Bhvr& res, T* ptr) : f_(f), ptr_(ptr), res_(res) {
fun_decorator(const F& f, T* ptr) : f_(f), ptr_(ptr) {
// nop
}
void operator()(Ts... xs) {
behavior operator()(Ts... xs) {
detail::type_list<R> token;
apply(token, xs...);
return apply(token, xs...);
}
template <class U>
typename std::enable_if<
std::is_convertible<U, Bhvr>::value
std::is_convertible<U, Bhvr>::value,
behavior
>::type
apply(detail::type_list<U>, Ts... xs) {
res_ = f_(ptr_, xs...);
auto bhvr = f_(ptr_, xs...);
return std::move(bhvr.unbox());
}
template <class U>
typename std::enable_if<
! std::is_convertible<U, Bhvr>::value
! std::is_convertible<U, Bhvr>::value,
behavior
>::type
apply(detail::type_list<U>, Ts... xs) {
f_(ptr_, xs...);
return {};
}
private:
F f_;
T* ptr_;
Bhvr& res_;
};
template <class Args>
......@@ -143,20 +149,23 @@ actor_factory make_actor_factory(F fun) {
message_verifier<typename trait::arg_types> mv;
if (! mv(msg, tk))
return {};
cfg.init_fun = [=](local_actor* x) -> behavior_t {
cfg.init_fun = [=](local_actor* x) -> behavior {
CAF_ASSERT(cfg.host);
using ctrait = typename detail::get_callable_trait<F>::type;
using fd = fun_decorator<F, impl, behavior_t, trait::mode,
typename ctrait::result_type,
typename ctrait::arg_types>;
behavior_t result;
fd f{fun, result, static_cast<impl*>(x)};
const_cast<message&>(msg).apply(f);
return result;
fd f{fun, static_cast<impl*>(x)};
empty_type_erased_tuple dummy_;
auto& ct = msg.empty() ? dummy_ : const_cast<message&>(msg).content();
auto opt = ct.apply(f);
if (! opt)
return {};
return std::move(*opt);
};
handle hdl = cfg.host->system().spawn_class<impl, no_spawn_options>(cfg);
return {actor_cast<strong_actor_ptr>(std::move(hdl)),
cfg.host->system().message_types(detail::type_list<handle>{})};
cfg.host->system().message_types<handle>()};
};
}
......@@ -177,9 +186,8 @@ actor_factory_result dyn_spawn_class(actor_config& cfg, message& msg) {
handle hdl{unsafe_actor_handle_init};
dyn_spawn_class_helper<handle, T, Ts...> factory{hdl, cfg};
msg.apply(factory);
detail::type_list<handle> token;
return {actor_cast<strong_actor_ptr>(std::move(hdl)),
cfg.host->system().message_types(token)};
cfg.host->system().message_types<handle>()};
}
template <class T, class... Ts>
......
......@@ -168,29 +168,30 @@ public:
virtual ~actor_system();
using message_types_set = std::set<std::string>;
/// A message passing interface (MPI) in run-time checkable representation.
using mpi = std::set<std::string>;
inline message_types_set message_types(detail::type_list<scoped_actor>) {
return message_types_set{};
inline mpi message_types(detail::type_list<scoped_actor>) const {
return mpi{};
}
inline message_types_set message_types(detail::type_list<actor>) {
return message_types_set{};
inline mpi message_types(detail::type_list<actor>) const {
return mpi{};
}
template <class... Ts>
message_types_set message_types(detail::type_list<typed_actor<Ts...>>) {
mpi message_types(detail::type_list<typed_actor<Ts...>>) const {
static_assert(sizeof...(Ts) > 0, "empty typed actor handle given");
message_types_set result{get_rtti_from_mpi<Ts>(types())...};
mpi result{get_rtti_from_mpi<Ts>(types())...};
return result;
}
inline message_types_set message_types(const actor&) {
return message_types_set{};
inline mpi message_types(const actor&) const {
return mpi{};
}
template <class... Ts>
message_types_set message_types(const typed_actor<Ts...>&) {
mpi message_types(const typed_actor<Ts...>&) const {
detail::type_list<typed_actor<Ts...>> token;
return message_types(token);
}
......@@ -198,11 +199,30 @@ public:
/// Returns a string representation of the messaging
/// interface using portable names;
template <class T>
message_types_set message_types() {
mpi message_types() const {
detail::type_list<T> token;
return message_types(token);
}
/// Returns whether actor handles described by `xs`
/// can be assigned to actor handles described by `ys`.
/// @experimental
inline bool assignable(const mpi& xs, const mpi& ys) const {
if (ys.empty())
return xs.empty();
if (xs.size() == ys.size())
return xs == ys;
return std::includes(xs.begin(), xs.end(), ys.begin(), ys.end());
}
/// Returns whether actor handles described by `xs`
/// can be assigned to actor handles of type `T`.
/// @experimental
template <class T>
bool assignable(const std::set<std::string>& xs) const {
return assignable(xs, message_types<T>());
}
/// Returns the host-local identifier for this system.
const node_id& node() const;
......
......@@ -130,6 +130,9 @@ public:
/// @experimental
template <class T, class... Ts>
actor_system_config& add_actor_type(std::string name) {
using handle = typename infer_handle_from_class<T>::type;
if (! std::is_same<handle, actor>::value)
add_message_type<handle>(name);
return add_actor_factory(std::move(name), make_actor_factory<T, Ts...>());
}
......@@ -138,6 +141,9 @@ public:
/// @experimental
template <class F>
actor_system_config& add_actor_type(std::string name, F f) {
using handle = typename infer_handle_from_fun<F>::type;
if (! std::is_same<handle, actor>::value)
add_message_type<handle>(name);
return add_actor_factory(std::move(name), make_actor_factory(std::move(f)));
}
......@@ -145,6 +151,7 @@ public:
template <class T>
actor_system_config& add_message_type(std::string name) {
static_assert(std::is_empty<T>::value
|| std::is_same<T, actor>::value // silence add_actor_type err
|| is_typed_actor<T>::value
|| (std::is_default_constructible<T>::value
&& std::is_copy_constructible<T>::value),
......
......@@ -402,6 +402,12 @@ private:
caf::error error_;
};
template <>
class expected<unit_t> : public expected<void> {
public:
using expected<void>::expected;
};
template <class T>
auto to_string(const expected<T>& x) -> decltype(to_string(*x)) {
if (x)
......
......@@ -135,11 +135,15 @@ class function_view {
public:
using type = Actor;
function_view() : impl_(unsafe_actor_handle_init) {
function_view(duration rel_timeout = infinite)
: timeout(rel_timeout),
impl_(unsafe_actor_handle_init) {
// nop
}
function_view(const type& impl) : impl_(impl) {
function_view(const type& impl, duration rel_timeout = infinite)
: timeout(rel_timeout),
impl_(impl) {
new_self(impl_);
}
......@@ -148,7 +152,9 @@ public:
self_.~scoped_actor();
}
function_view(function_view&& x) : impl_(std::move(x.impl_)) {
function_view(function_view&& x)
: timeout(x.timeout),
impl_(std::move(x.impl_)) {
if (! impl_.unsafe()) {
new (&self_) scoped_actor(impl_.home_system()); //(std::move(x.self_));
x.self_.~scoped_actor();
......@@ -156,6 +162,7 @@ public:
}
function_view& operator=(function_view&& x) {
timeout = x.timeout;
assign(x.impl_);
x.assign(unsafe_actor_handle_init);
return *this;
......@@ -178,7 +185,7 @@ public:
return sec::bad_function_call;
error err;
function_view_result<R> result;
self_->request(impl_, infinite, std::forward<Ts>(xs)...).receive(
self_->request(impl_, timeout, std::forward<Ts>(xs)...).receive(
[&](error& x) {
err = std::move(x);
},
......@@ -202,6 +209,8 @@ public:
return ! impl_.unsafe();
}
duration timeout;
private:
template <class T>
T&& flatten(T& x) {
......
......@@ -208,8 +208,15 @@ public:
/// @endcond
/// Returns a pointer to the sender of the current message.
inline strong_actor_ptr current_sender() {
return current_element_ ? current_element_->sender : nullptr;
/// @pre `current_mailbox_element() != nullptr`
inline strong_actor_ptr& current_sender() {
CAF_ASSERT(current_element_);
return current_element_->sender;
}
/// Returns a pointer to the currently processed mailbox element.
inline mailbox_element* current_mailbox_element() {
return current_element_;
}
/// Adds a unidirectional `monitor` to `whom`.
......
......@@ -101,10 +101,10 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) {
for (auto& kvp : self->state.data)
if (kvp.first != "*")
msgs.emplace_back(kvp.first, kvp.second.first);
return make_message(ok_atom::value, std::move(msgs));
return make_message(std::move(msgs));
}
auto i = self->state.data.find(key);
return make_message(ok_atom::value, std::move(key),
return make_message(std::move(key),
i != self->state.data.end() ? i->second.first
: make_message());
},
......@@ -147,14 +147,15 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) {
behavior spawn_serv_impl(event_based_actor* self) {
CAF_LOG_TRACE("");
return {
[=](get_atom, const std::string& name, message& args)
-> result<ok_atom, strong_actor_ptr, std::set<std::string>> {
[=](spawn_atom, const std::string& name, message& args)
-> result<strong_actor_ptr, std::set<std::string>> {
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());
actor_config cfg{self->context()};
auto res = self->system().types().make_actor(name, cfg, args);
if (! res.first)
return sec::cannot_spawn_actor_from_arguments;
return {ok_atom::value, res.first, res.second};
return {res.first, res.second};
}
};
}
......
......@@ -119,7 +119,7 @@ struct fixture {
// clear config
scoped_actor self{system};
self->request(config_server, infinite, get_atom::value, "*").receive(
[&](ok_atom, std::vector<std::pair<std::string, message>>& msgs) {
[&](std::vector<std::pair<std::string, message>>& msgs) {
for (auto& kvp : msgs)
self->send(config_server, put_atom::value, kvp.first, message{});
},
......@@ -160,7 +160,7 @@ struct fixture {
bool result = false;
scoped_actor self{system};
self->request(config_server, infinite, get_atom::value, key).receive(
[&](ok_atom, std::string&, message& msg) {
[&](std::string&, message& msg) {
msg.apply(
[&](type& val) {
result = detail::safe_equal(what, val);
......@@ -196,7 +196,7 @@ struct fixture {
size_t result = 0;
scoped_actor self{system};
self->request(config_server, infinite, get_atom::value, "*").receive(
[&](ok_atom, std::vector<std::pair<std::string, message>>& msgs) {
[&](std::vector<std::pair<std::string, message>>& msgs) {
for (auto& kvp : msgs)
if (! kvp.second.empty())
++result;
......
......@@ -26,6 +26,7 @@
#include <thread>
#include "caf/fwd.hpp"
#include "caf/send.hpp"
#include "caf/node_id.hpp"
#include "caf/expected.hpp"
#include "caf/actor_system.hpp"
......@@ -48,8 +49,20 @@ public:
~middleman();
/// Publishes `whom` at `port` and returns either an `error`
/// or the bound port.
/// Tries to open a port for other CAF instances to connect to.
/// @experimental
expected<uint16_t> open(uint16_t port, const char* in = nullptr,
bool ru = false);
/// Closes port `port` regardless of whether an actor is published to it.
expected<void> close(uint16_t port);
/// Tries to connect to given node.
/// @experimental
expected<node_id> connect(std::string host, uint16_t port);
/// Tries to publish `whom` at `port` and returns either an
/// `error` or the bound port.
/// @param whom Actor that should be published at `port`.
/// @param port Unused TCP port.
/// @param in The IP address to listen to or `INADDR_ANY` if `in == nullptr`.
......@@ -79,13 +92,13 @@ public:
return unpublish(whom.address(), port);
}
/// Establish a new connection to the typed actor at `host` on given `port`.
/// Establish a new connection to the actor at `host` on given `port`.
/// @param host Valid hostname or IP address.
/// @param port TCP port.
/// @returns An `actor` to the proxy instance representing
/// a remote actor or an `error`.
template <class ActorHandle>
expected<ActorHandle> typed_remote_actor(std::string host, uint16_t port) {
template <class ActorHandle = actor>
expected<ActorHandle> remote_actor(std::string host, uint16_t port) {
detail::type_list<ActorHandle> tk;
auto x = remote_actor(system().message_types(tk), std::move(host), port);
if (! x)
......@@ -94,15 +107,6 @@ public:
return actor_cast<ActorHandle>(std::move(*x));
}
/// Establish a new connection to the actor at `host` on given `port`.
/// @param host Valid hostname or IP address.
/// @param port TCP port.
/// @returns An `actor` to the proxy instance representing
/// a remote actor or an `error`.
inline expected<actor> remote_actor(std::string host, uint16_t port) {
return typed_remote_actor<actor>(std::move(host), port);
}
/// <group-name>@<host>:<port>
expected<group> remote_group(const std::string& group_uri);
......@@ -164,6 +168,26 @@ public:
/// or an error occurred.
strong_actor_ptr remote_lookup(atom_value name, const node_id& nid);
/// @experimental
template <class Handle>
expected<Handle> remote_spawn(const node_id& nid, std::string name,
message args,
duration timeout = std::chrono::seconds(60)) {
if (! nid || name.empty())
return sec::invalid_argument;
auto res = remote_spawn_impl(nid, name, args, timeout);
if (! res)
return std::move(res.error());
if (! system().assignable<Handle>(res->second)) {
// 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`.
using backend_pointer = std::unique_ptr<network::multiplexer>;
......@@ -273,10 +297,15 @@ private:
return system().spawn_class<Impl, Os>(cfg);
}
expected<std::pair<strong_actor_ptr, std::set<std::string>>>
remote_spawn_impl(const node_id& nid, std::string& name,
message& args, duration timeout);
expected<uint16_t> publish(const strong_actor_ptr& whom,
std::set<std::string> sigs,
uint16_t port, const char* in, bool ru);
expected<void> unpublish(const actor_addr& whom, uint16_t port);
expected<strong_actor_ptr> remote_actor(std::set<std::string> ifs,
......
......@@ -42,8 +42,7 @@ namespace io {
/// // reuse:_addr: Enables or disables SO_REUSEPORT option.
/// (publish_atom, uint16_t port, strong_actor_ptr whom,
/// set<string> ifs, string addr, bool reuse_addr)
/// -> either (ok_atom, uint16_t port)
/// or (error_atom, string error_string)
/// -> expected<uint16_t>
///
/// // Opens a new port other CAF instances can connect to. The
/// // difference between `PUBLISH` and `OPEN` is that no actor is mapped to
......@@ -53,7 +52,7 @@ namespace io {
/// // addr: IP address to listen to or empty for any.
/// // reuse:_addr: Enables or disables SO_REUSEPORT option.
/// (open_atom, uint16_t port, string addr, bool reuse_addr)
/// -> either (ok_atom, uint16_t port)
/// -> expected<uint16_t)
/// or (error_atom, string error_string)
///
/// // Queries a remote node and returns an ID to this node as well as
......@@ -63,22 +62,20 @@ namespace io {
/// // hostname: IP address or DNS hostname.
/// // port: TCP port.
/// (connect_atom, string hostname, uint16_t port)
/// -> either (ok_atom, node_id nid, strong_actor_ptr remote_actor, set<string> ifs)
/// -> either (node_id nid, strong_actor_ptr remote_actor, set<string> ifs)
/// or (error_atom, string error_string)
///
/// // Closes `port` if it is mapped to `whom`.
/// // whom: A published actor.
/// // port: Used TCP port.
/// (unpublish_atom, strong_actor_ptr whom, uint16_t port)
/// -> either (ok_atom)
/// or (error_atom, string error_string)
/// -> expected<void>
///
/// // Unconditionally closes `port`, removing any actor
/// // published at this port.
/// // port: Used TCP port.
/// (close_atom, uint16_t port)
/// -> either (ok_atom)
/// or (error_atom, string error_string)
/// -> expected<void>
///
/// // Spawns an actor on a remote node, initializing it using the arguments
/// // stored in `msg` and returns the address of the spawned actor and its
......@@ -87,8 +84,7 @@ namespace io {
/// // name: Announced type name of the actor.
/// // args: Initialization arguments for the actor.
/// (spawn_atom, node_id nid, string name, message args)
/// -> either (ok_atom, strong_actor_ptr, set<string>
/// or (error_atom, string error_string)
/// -> expected<strong_actor_ptr, set<string>>
///
/// }
/// ~~~
......@@ -96,20 +92,20 @@ using middleman_actor =
typed_actor<
replies_to<publish_atom, uint16_t, strong_actor_ptr,
std::set<std::string>, std::string, bool>
::with<ok_atom, uint16_t>,
::with<uint16_t>,
replies_to<open_atom, uint16_t, std::string, bool>
::with<ok_atom, uint16_t>,
::with<uint16_t>,
replies_to<connect_atom, std::string, uint16_t>
::with<ok_atom, node_id, strong_actor_ptr, std::set<std::string>>,
::with<node_id, strong_actor_ptr, std::set<std::string>>,
reacts_to<unpublish_atom, actor_addr, uint16_t>,
reacts_to<close_atom, uint16_t>,
replies_to<spawn_atom, node_id, std::string, message>
::with<ok_atom, strong_actor_ptr, std::set<std::string>>,
::with<strong_actor_ptr, std::set<std::string>>,
replies_to<get_atom, node_id>::with<node_id, std::string, uint16_t>>;
......
......@@ -129,7 +129,7 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
strong_actor_ptr ptr;
if (aid == invalid_actor_id) {
// can occur when connecting to the default port of a node
cb->deliver(ok_atom::value, nid, ptr, std::move(sigs));
cb->deliver(nid, ptr, std::move(sigs));
return;
}
if (nid == this_node()) {
......@@ -140,7 +140,7 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
ptr = namespace_.get_or_put(nid, aid);
CAF_LOG_ERROR_IF(! ptr, "creating actor in finalize_handshake failed");
}
cb->deliver(make_message(ok_atom::value, nid, ptr, std::move(sigs)));
cb->deliver(make_message(nid, ptr, std::move(sigs)));
this_context->callback = none;
}
......@@ -277,7 +277,7 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
tself->monitor(config_serv);
tself->become(
[=](spawn_atom, std::string& type, message& args)
-> delegated<ok_atom, strong_actor_ptr, std::set<std::string>> {
-> delegated<strong_actor_ptr, std::set<std::string>> {
CAF_LOG_TRACE(CAF_ARG(type) << CAF_ARG(args));
tself->delegate(actor_cast<actor>(std::move(config_serv)),
get_atom::value, std::move(type),
......@@ -350,7 +350,7 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
});
return {
// this config is send from the remote `ConfigServ`
[=](ok_atom, const std::string&, message& msg) {
[=](const std::string&, message& msg) {
CAF_LOG_TRACE(CAF_ARG(msg));
CAF_LOG_DEBUG("received requested config:" << CAF_ARG(msg));
// whatever happens, we are done afterwards
......@@ -472,10 +472,7 @@ behavior basp_broker::make_behavior() {
if (ctx.callback) {
CAF_LOG_WARNING("failed to handshake with remote node"
<< CAF_ARG(msg.handle));
ctx.callback->deliver(ok_atom::value,
node_id{none},
strong_actor_ptr{nullptr},
std::set<std::string>{});
ctx.callback->deliver(make_error(sec::disconnect_during_handshake));
}
close(msg.handle);
state.ctx.erase(msg.handle);
......@@ -508,15 +505,18 @@ behavior basp_broker::make_behavior() {
}
},
// received from some system calls like whereis
[=](forward_atom, strong_actor_ptr& src,
const node_id& dest_node, atom_value dest_name,
const message& msg) -> result<void> {
[=](forward_atom, const node_id& dest_node, atom_value dest_name,
const message& msg) -> result<message> {
auto cme = current_mailbox_element();
if (! cme)
return sec::invalid_argument;
auto& src = cme->sender;
CAF_LOG_TRACE(CAF_ARG(src)
<< ", " << CAF_ARG(dest_node)
<< ", " << CAF_ARG(dest_name)
<< ", " << CAF_ARG(msg));
if (! src)
return sec::cannot_forward_to_invalid_actor;
return sec::invalid_argument;
auto path = this->state.instance.tbl().lookup(dest_node);
if (! path) {
CAF_LOG_ERROR("no route to receiving node");
......@@ -525,16 +525,15 @@ behavior basp_broker::make_behavior() {
if (system().node() == src->node())
system().registry().put(src->id(), src);
auto writer = make_callback([&](serializer& sink) -> error {
std::vector<actor_addr> stages;
return sink(dest_name, stages, const_cast<message&>(msg));
return sink(dest_name, cme->stages, const_cast<message&>(msg));
});
basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag,
0, 0, state.this_node(), dest_node,
src->id(), invalid_actor_id};
0, cme->mid.integer_value(), state.this_node(),
dest_node, src->id(), invalid_actor_id};
state.instance.write(context(), path->wr_buf, hdr, &writer);
state.instance.flush(*path);
return unit;
return delegated<message>();
},
// received from underlying broker implementation
[=](const new_connection_msg& msg) {
......@@ -632,18 +631,6 @@ behavior basp_broker::make_behavior() {
}
return unit;
},
[=](spawn_atom, const node_id& nid, std::string& type, message& xs)
-> delegated<ok_atom, strong_actor_ptr, std::set<std::string>> {
CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(type) << CAF_ARG(xs));
auto i = state.spawn_servers.find(nid);
if (i == state.spawn_servers.end()) {
auto rp = make_response_promise();
rp.deliver(sec::no_route_to_receiving_node);
} else {
delegate(i->second, spawn_atom::value, std::move(type), std::move(xs));
}
return {};
},
[=](get_atom, const node_id& x)
-> std::tuple<node_id, std::string, uint16_t> {
std::string addr;
......
......@@ -34,6 +34,7 @@
#include "caf/actor_proxy.hpp"
#include "caf/make_counted.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/function_view.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/typed_event_based_actor.hpp"
......@@ -106,29 +107,52 @@ middleman::middleman(actor_system& sys)
// nop
}
expected<std::pair<strong_actor_ptr, std::set<std::string>>>
middleman::remote_spawn_impl(const node_id& nid, std::string& name,
message& args, duration timeout) {
auto f = make_function_view(actor_handle());
f.timeout = timeout;
auto res = f(spawn_atom::value, nid, std::move(name), std::move(args));
// why can't we assign a tuple<T1, T2> to pair<T1, T2>? No one knows!
if (! res)
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) {
std::string str;
if (cstr != nullptr)
str = cstr;
auto f = make_function_view(actor_handle());
return f(open_atom::value, port, std::move(str), ru);
}
expected<void> middleman::close(uint16_t port) {
auto f = make_function_view(actor_handle());
return f(close_atom::value, port);
}
expected<node_id> middleman::connect(std::string host, uint16_t port) {
auto f = make_function_view(actor_handle());
auto res = f(connect_atom::value, std::move(host), port);
if (! res)
return std::move(res.error());
return std::get<0>(*res);
}
expected<uint16_t> middleman::publish(const strong_actor_ptr& whom,
std::set<std::string> sigs,
uint16_t port, const char* in, bool ru) {
std::set<std::string> sigs, uint16_t port,
const char* cstr, bool ru) {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(sigs) << CAF_ARG(port)
<< CAF_ARG(in) << CAF_ARG(ru));
if (! whom)
return sec::cannot_publish_invalid_actor;
std::string str;
if (in != nullptr)
str = in;
auto mm = actor_handle();
scoped_actor self{system()};
expected<uint16_t> result{port};
self->request(mm, infinite, publish_atom::value, port,
std::move(whom), std::move(sigs), str, ru).receive(
[&](ok_atom, uint16_t res) {
result = res;
},
[&](error& err) {
result = std::move(err);
}
);
return result;
std::string in;
if (cstr)
in = cstr;
auto f = make_function_view(actor_handle());
return f(publish_atom::value, port, std::move(whom), std::move(sigs), in, ru);
}
expected<uint16_t> middleman::publish_local_groups(uint16_t port,
......@@ -154,56 +178,25 @@ expected<uint16_t> middleman::publish_local_groups(uint16_t port,
expected<void> middleman::unpublish(const actor_addr& whom, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(port));
scoped_actor self{system(), true};
expected<void> result{unit};
self->request(actor_handle(), infinite,
unpublish_atom::value, whom, port).receive(
[] {
// ok, basp_broker is done
},
[&](error& x) {
result = std::move(x);
}
);
return result;
auto f = make_function_view(actor_handle());
return f(unpublish_atom::value, whom, port);
}
expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs,
std::string host,
uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(ifs) << CAF_ARG(host) << CAF_ARG(port));
auto mm = actor_handle();
error err;
strong_actor_ptr result;
scoped_actor self{system(), true};
self->request(mm, infinite, connect_atom::value,
std::move(host), port).receive(
[&](ok_atom, const node_id&, strong_actor_ptr res, std::set<std::string>& xs) {
CAF_LOG_TRACE(CAF_ARG(res) << CAF_ARG(xs));
if (! res) {
err = make_error(sec::no_actor_published_at_port,
"no actor published at port", port);
return;
}
if (ifs.empty() != xs.empty()
|| ! std::includes(xs.begin(), xs.end(), ifs.begin(), ifs.end())) {
using kvpair = std::pair<std::string, std::set<std::string>>;
err = make_error(sec::unexpected_actor_messaging_interface,
kvpair("expected", ifs),
kvpair("found", xs));
return;
}
result.swap(res);
},
[&](error& msg) {
CAF_LOG_TRACE(CAF_ARG(msg));
err = std::move(msg);
}
);
CAF_ASSERT(result || err);
if (! result)
return err;
return result;
auto f = make_function_view(actor_handle());
auto res = f(connect_atom::value, std::move(host), port);
if (! res)
return std::move(res.error());
strong_actor_ptr ptr = std::move(std::get<1>(*res));
if (! ptr)
return make_error(sec::no_actor_published_at_port, port);
if (! system().assignable(std::get<2>(*res), ifs))
return make_error(sec::unexpected_actor_messaging_interface, std::move(ifs),
std::move(std::get<2>(*res)));
return ptr;
}
expected<group> middleman::remote_group(const std::string& group_uri) {
......
......@@ -71,11 +71,11 @@ public:
return "middleman_actor";
}
using put_res = result<ok_atom, uint16_t>;
using put_res = result<uint16_t>;
using mpi_set = std::set<std::string>;
using get_res = delegated<ok_atom, node_id, strong_actor_ptr, mpi_set>;
using get_res = delegated<node_id, strong_actor_ptr, mpi_set>;
using del_res = delegated<void>;
......@@ -105,7 +105,7 @@ public:
auto x = cached(key);
if (x) {
CAF_LOG_DEBUG("found cached entry" << CAF_ARG(*x));
rp.deliver(ok_atom::value, get<0>(*x), get<1>(*x), get<2>(*x));
rp.deliver(get<0>(*x), get<1>(*x), get<2>(*x));
return {};
}
// attach this promise to a pending request if possible
......@@ -125,7 +125,7 @@ public:
std::vector<response_promise> tmp{std::move(rp)};
pending_.emplace(key, std::move(tmp));
request(broker_, infinite, connect_atom::value, hdl, port).then(
[=](ok_atom, node_id& nid, strong_actor_ptr& addr, mpi_set& sigs) {
[=](node_id& nid, strong_actor_ptr& addr, mpi_set& sigs) {
auto i = pending_.find(key);
if (i == pending_.end())
return;
......@@ -133,8 +133,8 @@ public:
monitor(addr);
cached_.emplace(key, std::make_tuple(nid, addr, sigs));
}
auto res = make_message(ok_atom::value, std::move(nid),
std::move(addr), std::move(sigs));
auto res = make_message(std::move(nid), std::move(addr),
std::move(sigs));
for (auto& promise : i->second)
promise.deliver(res);
pending_.erase(i);
......@@ -161,9 +161,12 @@ public:
return {};
},
[=](spawn_atom atm, node_id& nid, std::string& str, message& msg)
-> delegated<ok_atom, strong_actor_ptr, mpi_set> {
-> delegated<strong_actor_ptr, mpi_set> {
CAF_LOG_TRACE("");
delegate(broker_, atm, std::move(nid), std::move(str), std::move(msg));
printf("%s %d -- %s\n", __FILE__, __LINE__, to_string(current_mailbox_element()->content()).c_str());
delegate(broker_, forward_atom::value, nid, atom("SpawnServ"),
make_message(atm, std::move(str), std::move(msg)));
//delegate(broker_, atm, std::move(nid), std::move(str), std::move(msg));
return {};
},
[=](get_atom atm, node_id nid)
......@@ -194,7 +197,7 @@ private:
actual_port = res->second;
anon_send(broker_, publish_atom::value, hdl, actual_port,
std::move(whom), std::move(sigs));
return {ok_atom::value, actual_port};
return actual_port;
}
optional<endpoint_data&> cached(const endpoint& ep) {
......
......@@ -615,7 +615,7 @@ CAF_TEST(remote_actor_and_send) {
invalid_actor_id, jupiter().dummy_actor->id());
CAF_MESSAGE("BASP broker should've send the proxy");
f.receive(
[&](ok_atom, node_id nid, strong_actor_ptr res, std::set<std::string> ifs) {
[&](node_id nid, strong_actor_ptr res, std::set<std::string> ifs) {
CAF_REQUIRE(res);
auto aptr = actor_cast<abstract_actor*>(res);
CAF_REQUIRE(dynamic_cast<forwarding_actor_proxy*>(aptr) != nullptr);
......@@ -815,7 +815,7 @@ CAF_TEST(automatic_connection) {
this_node(), this_node(),
invalid_actor_id, connection_helper},
std::vector<actor_id>{},
make_message(ok_atom::value, "basp.default-connectivity",
make_message("basp.default-connectivity",
make_message(uint16_t{8080}, std::move(res))));
// our connection helper should now connect to jupiter and
// send the scribe handle over to the BASP broker
......
......@@ -36,94 +36,71 @@ using namespace caf;
namespace {
behavior mirror(event_based_actor* self) {
self->set_default_handler(reflect);
return {
[] {
// nop
using add_atom = atom_constant<atom("add")>;
using sub_atom = atom_constant<atom("sub")>;
using calculator = typed_actor<replies_to<add_atom, int, int>::with<int>,
replies_to<sub_atom, int, int>::with<int>>;
// function-based, dynamically typed, event-based API
behavior calculator_fun(event_based_actor*) {
return behavior{
[](add_atom, int a, int b) {
return a + b;
},
[](sub_atom, int a, int b) {
return a - b;
}
};
}
behavior client(event_based_actor* self, actor serv) {
self->send(serv, ok_atom::value);
// function-based, statically typed, event-based API
calculator::behavior_type typed_calculator_fun() {
return {
[] {
// nop
[](add_atom, int a, int b) {
return a + b;
},
[](sub_atom, int a, int b) {
return a - b;
}
};
}
struct server_state {
actor client; // the spawn we connect to the server in our main
actor aut; // our mirror
server_state()
: client(unsafe_actor_handle_init),
aut(unsafe_actor_handle_init) {
// nop
struct config : actor_system_config {
config(int argc, char** argv) {
parse(argc, argv);
load<io::middleman>();
add_actor_type("calculator", calculator_fun);
add_actor_type("typed_calculator", typed_calculator_fun);
}
};
behavior server(stateful_actor<server_state>* self) {
CAF_LOG_TRACE("");
return {
[=](ok_atom) {
CAF_LOG_TRACE("");
auto s = self->current_sender();
CAF_REQUIRE(s != nullptr);
CAF_REQUIRE(self->node() != s->node());
auto opt = actor_cast<actor>(s);
//CAF_REQUIRE(opt);
self->state.client = opt;
auto mm = self->system().middleman().actor_handle();
self->request(mm, infinite, spawn_atom::value,
s->node(), "mirror", make_message()).then(
[=](ok_atom, const strong_actor_ptr& ptr,
const std::set<std::string>& ifs) {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(ifs));
CAF_REQUIRE(ptr);
CAF_CHECK(ifs.empty());
self->state.aut = actor_cast<actor>(ptr);
self->send(self->state.aut, "hello mirror");
self->become(
[=](const std::string& str) {
CAF_CHECK_EQUAL(self->current_sender(),
self->state.aut.address());
CAF_CHECK_EQUAL(str, "hello mirror");
self->send_exit(self->state.aut, exit_reason::kill);
self->send_exit(self->state.client, exit_reason::kill);
self->quit();
}
);
}
);
},
[=](const error& err) {
CAF_LOG_TRACE("");
CAF_ERROR("Error: " << self->system().render(err));
}
};
}
void run_client(int argc, char** argv, uint16_t port) {
actor_system_config cfg;
cfg.parse(argc, argv)
.load<io::middleman>()
.add_actor_type("mirror", mirror);
config cfg{argc, argv};
actor_system system{cfg};
CAF_EXP_THROW(serv, system.middleman().remote_actor("localhost", port));
system.spawn(client, serv);
auto& mm = system.middleman();
auto nid = mm.connect("localhost", port);
CAF_REQUIRE(nid);
CAF_REQUIRE_NOT_EQUAL(system.node(), *nid);
auto calc = mm.remote_spawn<calculator>(*nid, "calculator", make_message());
CAF_REQUIRE_EQUAL(calc, sec::unexpected_actor_messaging_interface);
calc = mm.remote_spawn<calculator>(*nid, "typed_calculator", make_message());
CAF_REQUIRE(calc);
auto f1 = make_function_view(*calc);
CAF_REQUIRE_EQUAL(f1(add_atom::value, 10, 20), 30);
CAF_REQUIRE_EQUAL(f1(sub_atom::value, 10, 20), -10);
f1.assign(unsafe_actor_handle_init);
anon_send_exit(*calc, exit_reason::kill);
mm.close(port);
}
void run_server(int argc, char** argv) {
actor_system_config cfg;
actor_system system{cfg.load<io::middleman>().parse(argc, argv)};
auto serv = system.spawn(server);
CAF_EXP_THROW(port, system.middleman().publish(serv, 0));
CAF_REQUIRE(port != 0);
CAF_MESSAGE("published server at port " << port);
std::thread child([=] { run_client(argc, argv, port); });
config cfg{argc, argv};
actor_system system{cfg};
CAF_EXP_THROW(port, system.middleman().open(0));
std::thread child{[=] { run_client(argc, argv, port); }};
child.join();
printf("%s %d -- %d\n", __FILE__, __LINE__, (int) system.registry().running());
}
} // namespace <anonymous>
......
......@@ -89,7 +89,7 @@ void run_client(int argc, char** argv, uint16_t port) {
CAF_MESSAGE(system.render(res.error()));
CAF_MESSAGE("connect to typed_remote_actor");
CAF_EXP_THROW(serv,
system.middleman().typed_remote_actor<server_type>("127.0.0.1",
system.middleman().remote_actor<server_type>("127.0.0.1",
port));
auto f = make_function_view(serv);
CAF_CHECK_EQUAL(f(ping{42}), pong{42});
......
......@@ -75,7 +75,7 @@ struct fixture {
scoped_actor self{system, true};
self->request(system.middleman().actor_handle(), infinite,
connect_atom::value, hostname, port).receive(
[&](ok_atom, node_id&, strong_actor_ptr& res, std::set<std::string>& xs) {
[&](node_id&, strong_actor_ptr& res, std::set<std::string>& xs) {
CAF_REQUIRE(xs.empty());
if (res)
result = actor_cast<actor>(std::move(res));
......
......@@ -220,7 +220,7 @@ void bootstrap(actor_system& system,
std::string slaveslist;
for (size_t i = 0; i < slaves.size(); ++i) {
self->receive(
[&](ok_atom, const string& host, uint16_t slave_port) {
[&](const string& host, uint16_t slave_port) {
if (! slaveslist.empty())
slaveslist += ',';
slaveslist += host;
......
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