Commit 177277d6 authored by Dominik Charousset's avatar Dominik Charousset

Fix linking with remote actors, closes #498

parent 37a6cd0b
......@@ -19,7 +19,7 @@
// This file is partially included in the manual, do not modify
// without updating the references in the *.tex files!
// Manual references: lines 32-91 (Error.tex)
// Manual references: lines 32-93 (Error.tex)
#ifndef CAF_SEC_HPP
#define CAF_SEC_HPP
......@@ -88,6 +88,8 @@ enum class sec : uint8_t {
no_proxy_registry,
/// An exception was thrown during message handling.
runtime_error,
/// Linking to a remote actor failed because actor no longer exists.
remote_linking_failed,
/// A function view was called without assigning an actor first.
bad_function_call
};
......
......@@ -80,7 +80,7 @@ bool forwarding_actor_proxy::link_impl(linking_operation op,
// causes remote actor to link to (proxy of) other
// receiving peer will call: this->local_link_to(other)
forward_msg(ctrl(), invalid_message_id,
make_message(link_atom::value, other->address()));
make_message(link_atom::value, other->ctrl()));
return true;
}
break;
......@@ -88,7 +88,7 @@ bool forwarding_actor_proxy::link_impl(linking_operation op,
if (remove_link_impl(other)) {
// causes remote actor to unlink from (proxy of) other
forward_msg(ctrl(), invalid_message_id,
make_message(unlink_atom::value, other->address()));
make_message(unlink_atom::value, other->ctrl()));
return true;
}
break;
......@@ -96,7 +96,7 @@ bool forwarding_actor_proxy::link_impl(linking_operation op,
if (establish_backlink_impl(other)) {
// causes remote actor to unlink from (proxy of) other
forward_msg(ctrl(), invalid_message_id,
make_message(link_atom::value, other->address()));
make_message(link_atom::value, other->ctrl()));
return true;
}
break;
......@@ -104,7 +104,7 @@ bool forwarding_actor_proxy::link_impl(linking_operation op,
if (remove_backlink_impl(other)) {
// causes remote actor to unlink from (proxy of) other
forward_msg(ctrl(), invalid_message_id,
make_message(unlink_atom::value, other->address()));
make_message(unlink_atom::value, other->ctrl()));
return true;
}
break;
......
......@@ -55,6 +55,7 @@ const char* sec_strings[] = {
"unknown_type",
"no_proxy_registry",
"runtime_error",
"remote_linking_failed",
"bad_function_call"
};
......
......@@ -233,6 +233,50 @@ void basp_broker_state::deliver(const node_id& src_nid, actor_id src_aid,
<< CAF_ARG(msg) << CAF_ARG(mid));
auto src = src_nid == this_node() ? system().registry().get(src_aid)
: proxies().get_or_put(src_nid, src_aid);
// Intercept link messages. Forwarding actor proxies signalize linking
// by sending link_atom/unlink_atom message with src = dest.
if (msg.type_token() == make_type_token<atom_value, strong_actor_ptr>()) {
switch (static_cast<uint64_t>(msg.get_as<atom_value>(0))) {
default:
break;
case link_atom::value.uint_value(): {
if (src_nid != this_node()) {
CAF_LOG_WARNING("received link message for an other node");
return;
}
auto ptr = msg.get_as<strong_actor_ptr>(1);
if (!ptr) {
CAF_LOG_WARNING("received link message with invalid target");
return;
}
if (!src) {
CAF_LOG_DEBUG("received link for invalid actor, report error");
anon_send(actor_cast<actor>(ptr),
make_error(sec::remote_linking_failed));
return;
}
static_cast<actor_proxy*>(ptr->get())->local_link_to(src->get());
return;
}
case unlink_atom::value.uint_value(): {
if (src_nid != this_node()) {
CAF_LOG_WARNING("received unlink message for an other node");
return;
}
auto ptr = msg.get_as<strong_actor_ptr>(1);
if (!ptr) {
CAF_LOG_DEBUG("received unlink message with invalid target");
return;
}
if (!src) {
CAF_LOG_DEBUG("received unlink for invalid actor, report error");
return;
}
static_cast<actor_proxy*>(ptr->get())->local_unlink_from(src->get());
return;
}
}
}
if (!dest) {
auto rsn = exit_reason::remote_link_unreachable;
CAF_LOG_INFO("cannot deliver message, destination not found");
......@@ -313,13 +357,6 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
// writing std::numeric_limits<actor_id>::max() is a hack to get
// this send-to-named-actor feature working with older CAF releases
instance.write(self->context(), path->wr_buf, hdr, &writer);
/*
basp::message_type::dispatch_message,
nullptr, static_cast<uint64_t>(atom("SpawnServ")),
this_node(), nid,
tmp.id(), std::numeric_limits<actor_id>::max(),
&writer);
*/
instance.flush(*path);
}
......
......@@ -30,30 +30,32 @@
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using namespace caf;
namespace {
constexpr char local_host[] = "127.0.0.1";
class config : public caf::actor_system_config {
class config : public actor_system_config {
public:
config() {
load<caf::io::middleman>();
load<io::middleman>();
add_message_type<std::vector<int>>("std::vector<int>");
actor_system_config::parse(caf::test::engine::argc(),
caf::test::engine::argv());
actor_system_config::parse(test::engine::argc(),
test::engine::argv());
}
};
struct fixture {
config server_side_config;
caf::actor_system server_side{server_side_config};
actor_system server_side{server_side_config};
config client_side_config;
caf::actor_system client_side{client_side_config};
caf::io::middleman& server_side_mm = server_side.middleman();
caf::io::middleman& client_side_mm = client_side.middleman();
actor_system client_side{client_side_config};
io::middleman& server_side_mm = server_side.middleman();
io::middleman& client_side_mm = client_side.middleman();
};
caf::behavior make_pong_behavior() {
behavior make_pong_behavior() {
return {
[](int val) -> int {
CAF_MESSAGE("pong with " << ++val);
......@@ -62,8 +64,7 @@ caf::behavior make_pong_behavior() {
};
}
caf::behavior make_ping_behavior(caf::event_based_actor* self,
caf::actor pong) {
behavior make_ping_behavior(event_based_actor* self, actor pong) {
CAF_MESSAGE("ping with " << 0);
self->send(pong, 0);
return {
......@@ -71,7 +72,7 @@ caf::behavior make_ping_behavior(caf::event_based_actor* self,
if (val == 3) {
CAF_MESSAGE("ping with exit");
self->send_exit(self->current_sender(),
caf::exit_reason::user_shutdown);
exit_reason::user_shutdown);
CAF_MESSAGE("ping quits");
self->quit();
}
......@@ -89,7 +90,7 @@ std::string to_string(const std::vector<int>& vec) {
return os.str();
}
caf::behavior make_sort_behavior() {
behavior make_sort_behavior() {
return {
[](std::vector<int>& vec) -> std::vector<int> {
CAF_MESSAGE("sorter received: " << to_string(vec));
......@@ -100,20 +101,39 @@ caf::behavior make_sort_behavior() {
};
}
caf::behavior make_sort_requester_behavior(caf::event_based_actor* self,
caf::actor sorter) {
behavior make_sort_requester_behavior(event_based_actor* self, actor sorter) {
self->send(sorter, std::vector<int>{5, 4, 3, 2, 1});
return {
[=](const std::vector<int>& vec) {
CAF_MESSAGE("sort requester received: " << to_string(vec));
for (size_t i = 1; i < vec.size(); ++i)
CAF_CHECK_EQUAL(static_cast<int>(i), vec[i - 1]);
self->send_exit(sorter, caf::exit_reason::user_shutdown);
self->send_exit(sorter, exit_reason::user_shutdown);
self->quit();
}
};
}
behavior fragile_mirror(event_based_actor* self) {
return {
[=](int i) {
self->quit(exit_reason::user_shutdown);
return i;
}
};
}
behavior linking_actor(event_based_actor* self, actor buddy) {
CAF_MESSAGE("link to mirror and send dummy message");
self->link_to(buddy);
self->send(buddy, 42);
return {
[](int i) {
CAF_CHECK_EQUAL(i, 42);
}
};
}
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(dynamic_remote_actor_tests, fixture)
......@@ -131,7 +151,7 @@ CAF_TEST(identity_semantics) {
CAF_EXP_THROW(server2, client_side_mm.remote_actor(local_host, port2));
CAF_CHECK_EQUAL(server1, client_side_mm.remote_actor(local_host, port1));
CAF_CHECK_EQUAL(server2, client_side_mm.remote_actor(local_host, port2));
caf::anon_send_exit(server, caf::exit_reason::user_shutdown);
anon_send_exit(server, exit_reason::user_shutdown);
}
CAF_TEST(ping_pong) {
......@@ -153,4 +173,18 @@ CAF_TEST(custom_message_type) {
client_side.spawn(make_sort_requester_behavior, sorter);
}
CAF_TEST(remote_link) {
// server side
CAF_EXP_THROW(port, server_side_mm.publish(server_side.spawn(fragile_mirror),
0, local_host));
// client side
CAF_EXP_THROW(mirror, client_side_mm.remote_actor(local_host, port));
auto linker = client_side.spawn(linking_actor, mirror);
scoped_actor self{client_side};
self->wait_for(linker);
CAF_MESSAGE("linker exited");
self->wait_for(mirror);
CAF_MESSAGE("mirror exited");
}
CAF_TEST_FIXTURE_SCOPE_END()
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