Commit 3a7ecac8 authored by Dominik Charousset's avatar Dominik Charousset

Add whereis for remote actor lookup

parent 433ad433
...@@ -26,8 +26,15 @@ ...@@ -26,8 +26,15 @@
namespace caf { namespace caf {
namespace experimental { namespace experimental {
/// Returns the actor with the (locally) registered name.
actor whereis(atom_value registered_name); actor whereis(atom_value registered_name);
/// Returns the actor with the registered name located at `nid` or
/// `invalid_actor` if `nid` is not connected or does not
/// respond to name lookups.
/// @warning Blocks the caller until `nid` responded to the lookup.
actor whereis(atom_value registered_name, node_id nid);
} // namespace experimental } // namespace experimental
} // namespace caf } // namespace caf
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include "caf/experimental/whereis.hpp" #include "caf/experimental/whereis.hpp"
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/detail/singletons.hpp" #include "caf/detail/singletons.hpp"
#include "caf/detail/actor_registry.hpp" #include "caf/detail/actor_registry.hpp"
...@@ -31,5 +32,30 @@ actor whereis(atom_value registered_name) { ...@@ -31,5 +32,30 @@ actor whereis(atom_value registered_name) {
return detail::singletons::get_actor_registry()->get_named(registered_name); return detail::singletons::get_actor_registry()->get_named(registered_name);
} }
actor whereis(atom_value registered_name, node_id nid) {
auto basp = whereis(atom("BASP"));
if (basp == invalid_actor)
return invalid_actor;
actor result;
scoped_actor self;
try {
self->send(basp, forward_atom::value, self.address(), nid, registered_name,
make_message(sys_atom::value, get_atom::value, "info"));
self->receive(
[&](ok_atom, const std::string& /* key == "info" */,
const actor_addr& addr, const std::string& /* name */) {
result = actor_cast<actor>(addr);
},
after(std::chrono::minutes(5)) >> [] {
// nop
}
);
} catch (...) {
// nop
}
return result;
}
} // namespace experimental } // namespace experimental
} // namespace caf } // namespace caf
...@@ -183,7 +183,7 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) { ...@@ -183,7 +183,7 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
mm->backend().dispatch([=] { mm->backend().dispatch([=] {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
// ... to make sure this is safe // ... to make sure this is safe
if (bptr == mm->get_named_broker<basp_broker>(atom("_BASP")) if (bptr == mm->get_named_broker<basp_broker>(atom("BASP"))
&& bptr->exit_reason() == exit_reason::not_exited) && bptr->exit_reason() == exit_reason::not_exited)
send_kill_proxy_instance(reason); send_kill_proxy_instance(reason);
}); });
...@@ -476,6 +476,38 @@ behavior basp_broker::make_behavior() { ...@@ -476,6 +476,38 @@ behavior basp_broker::make_behavior() {
srb(sender, mid); srb(sender, mid);
} }
}, },
// received from some system calls like whereis
[=](forward_atom, const actor_addr& sender,
const node_id& receiving_node, atom_value receiver_name,
const message& msg) -> optional<message> {
CAF_LOG_TRACE(CAF_TSARG(sender) << ", " << CAF_TSARG(receiving_node)
<< ", " << CAF_TSARG(receiver_name)
<< ", " << CAF_MARG(mid, integer_value)
<< ", " << CAF_TSARG(msg));
if (sender == invalid_actor_addr)
return make_message(error_atom::value, "sender == invalid_actor");
if (! sender.is_remote())
detail::singletons::get_actor_registry()->put(sender->id(), sender);
auto writer = make_callback([&](serializer& sink) {
msg.serialize(sink);
});
auto path = this->state.instance.tbl().lookup(receiving_node);
if (! path) {
CAF_LOG_ERROR("no route to receiving node");
return make_message(error_atom::value, "no route to receiving node");
}
// writing std::numeric_limits<actor_id>::max() is a hack to get
// this send-to-named-actor feature working with older CAF releases
this->state.instance.write(path->wr_buf,
basp::message_type::dispatch_message,
nullptr, static_cast<uint64_t>(receiver_name),
state.this_node(), receiving_node,
sender.id(),
std::numeric_limits<actor_id>::max(),
&writer);
state.instance.flush(*path);
return none;
},
// received from underlying broker implementation // received from underlying broker implementation
[=](const new_connection_msg& msg) { [=](const new_connection_msg& msg) {
CAF_LOG_TRACE("handle = " << msg.handle.id()); CAF_LOG_TRACE("handle = " << msg.handle.id());
...@@ -514,7 +546,7 @@ behavior basp_broker::make_behavior() { ...@@ -514,7 +546,7 @@ behavior basp_broker::make_behavior() {
std::set<std::string>& sigs) { std::set<std::string>& sigs) {
CAF_LOG_TRACE(CAF_ARG(hdl.id()) << ", "<< CAF_TSARG(whom) CAF_LOG_TRACE(CAF_ARG(hdl.id()) << ", "<< CAF_TSARG(whom)
<< ", " << CAF_ARG(port)); << ", " << CAF_ARG(port));
if (hdl.invalid() || whom == invalid_actor_addr) if (hdl.invalid())
return; return;
try { try {
assign_tcp_doorman(hdl); assign_tcp_doorman(hdl);
...@@ -523,7 +555,8 @@ behavior basp_broker::make_behavior() { ...@@ -523,7 +555,8 @@ behavior basp_broker::make_behavior() {
CAF_LOG_DEBUG("failed to assign doorman from handle"); CAF_LOG_DEBUG("failed to assign doorman from handle");
return; return;
} }
detail::singletons::get_actor_registry()->put(whom->id(), whom); if (whom != invalid_actor_addr)
detail::singletons::get_actor_registry()->put(whom->id(), whom);
state.instance.add_published_actor(port, whom, std::move(sigs)); state.instance.add_published_actor(port, whom, std::move(sigs));
}, },
// received from middleman actor (delegated) // received from middleman actor (delegated)
...@@ -610,8 +643,7 @@ behavior basp_broker::make_behavior() { ...@@ -610,8 +643,7 @@ behavior basp_broker::make_behavior() {
} }
}, },
// catch-all error handler // catch-all error handler
others >> others >> [=] {
[=] {
CAF_LOGF_ERROR("received unexpected message: " CAF_LOGF_ERROR("received unexpected message: "
<< to_string(current_message())); << to_string(current_message()));
}}; }};
......
...@@ -276,7 +276,8 @@ void middleman::initialize() { ...@@ -276,7 +276,8 @@ void middleman::initialize() {
do_announce<connection_handle>("caf::io::connection_handle"); do_announce<connection_handle>("caf::io::connection_handle");
do_announce<new_connection_msg>("caf::io::new_connection_msg"); do_announce<new_connection_msg>("caf::io::new_connection_msg");
do_announce<new_data_msg>("caf::io::new_data_msg"); do_announce<new_data_msg>("caf::io::new_data_msg");
actor mgr = get_named_broker<basp_broker>(atom("_BASP")); actor mgr = get_named_broker<basp_broker>(atom("BASP"));
detail::singletons::get_actor_registry()->put_named(atom("BASP"), mgr);
manager_ = spawn<middleman_actor_impl, detached + hidden>(*this, mgr); manager_ = spawn<middleman_actor_impl, detached + hidden>(*this, mgr);
} }
......
...@@ -111,7 +111,7 @@ public: ...@@ -111,7 +111,7 @@ public:
mpx_ = new network::test_multiplexer; mpx_ = new network::test_multiplexer;
set_middleman(mpx_); set_middleman(mpx_);
auto mm = middleman::instance(); auto mm = middleman::instance();
aut_ = mm->get_named_broker<basp_broker>(atom("_BASP")); aut_ = mm->get_named_broker<basp_broker>(atom("BASP"));
this_node_ = detail::singletons::get_node_id(); this_node_ = detail::singletons::get_node_id();
CAF_MESSAGE("this node: " << to_string(this_node_)); CAF_MESSAGE("this node: " << to_string(this_node_));
self_.reset(new scoped_actor); self_.reset(new scoped_actor);
......
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