Commit cc66ef18 authored by Dominik Charousset's avatar Dominik Charousset

Add async publish/remote_actor replacements to MM

Add a managing typed actor to the MM offering asynchronous pub and get
operations. Put is used by `publish()`, meaning to "put" a port/actor mapping
to the MM. Get is used by `remote_actor()`, meaning to "get" a proxy instance
from the MM. Closes #209.
parent 1959b85c
......@@ -67,13 +67,11 @@ class basp_broker : public broker, public actor_namespace::backend {
message_id mid, const message& msg);
struct client_handshake_data {
id_type remote_id;
int64_t request_id;
actor client;
std::set<std::string> expected_ifs;
};
void init_client(connection_handle hdl, client_handshake_data* data);
inline actor_namespace& get_namespace() {
return m_namespace;
}
......@@ -96,14 +94,13 @@ class basp_broker : public broker, public actor_namespace::backend {
await_payload,
// connection is going to be shut down because of an error
close_connection
};
struct connection_context {
connection_state state;
connection_handle hdl;
id_type remote_id;
client_handshake_data* handshake_data;
optional<client_handshake_data> handshake_data;
basp::header hdr;
// keep a reference to the published actor of
// the remote node to prevent this particular
......@@ -111,7 +108,6 @@ class basp_broker : public broker, public actor_namespace::backend {
// a bug where re-using an "old" connection via
// remote_actor() could return an expired proxy
actor published_actor;
};
void read(binary_deserializer& bs, basp::header& msg);
......@@ -131,8 +127,7 @@ class basp_broker : public broker, public actor_namespace::backend {
void new_data(connection_context& ctx, buffer_type& buf);
void init_handshake_as_client(connection_context& ctx,
client_handshake_data* ptr);
void init_handshake_as_client(connection_context& ctx);
void init_handshake_as_sever(connection_context& ctx,
actor_addr published_actor);
......
......@@ -38,7 +38,36 @@ namespace caf {
namespace io {
/**
* Manages brokers.
* API for asynchronous networking operations.
*/
using middleman_actor =
typed_actor<
replies_to<put_atom, actor_addr, uint16_t, std::string, bool>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<put_atom, actor_addr, uint16_t, std::string>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<put_atom, actor_addr, uint16_t, bool>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<put_atom, actor_addr, uint16_t>
::with_either<ok_atom, uint16_t>
::or_else<error_atom, std::string>,
replies_to<get_atom, std::string, uint16_t, std::set<std::string>>
::with_either<ok_atom, actor_addr>
::or_else<error_atom, std::string>,
replies_to<get_atom, std::string, uint16_t>
::with_either<ok_atom, actor_addr>
::or_else<error_atom, std::string>>;
/**
* Returns a handle for asynchronous networking operations.
*/
middleman_actor get_middleman_actor();
/**
* Manages brokers and network backends.
*/
class middleman : public detail::abstract_singleton {
public:
......@@ -51,6 +80,11 @@ class middleman : public detail::abstract_singleton {
*/
static middleman* instance();
/**
* Returns a handle to the actor managing the middleman singleton.
*/
middleman_actor actor_handle();
/**
* Returns the broker associated with `name` or creates a
* new instance of type `Impl`.
......@@ -139,6 +173,8 @@ class middleman : public detail::abstract_singleton {
std::set<broker_ptr> m_brokers;
// user-defined hooks
hook_uptr m_hooks;
// actor offering asyncronous IO by managing this singleton instance
middleman_actor m_manager;
};
} // namespace io
......
......@@ -68,7 +68,7 @@ behavior basp_broker::make_behavior() {
CAF_REQUIRE(m_ctx.count(msg.handle) == 0);
auto& ctx = m_ctx[msg.handle];
ctx.hdl = msg.handle;
ctx.handshake_data = nullptr;
ctx.handshake_data = none;
ctx.state = await_client_handshake;
init_handshake_as_sever(ctx, m_acceptors[msg.source].first->address());
},
......@@ -80,7 +80,8 @@ behavior basp_broker::make_behavior() {
if (j != m_ctx.end()) {
auto hd = j->second.handshake_data;
if (hd) {
send(hd->client, atom("ERROR"), "disconnect during handshake");
send(hd->client, error_atom{}, hd->request_id,
"disconnect during handshake");
}
m_ctx.erase(j);
}
......@@ -140,8 +141,26 @@ behavior basp_broker::make_behavior() {
<< CAF_ARG(aid));
erase_proxy(nid, aid);
},
// received from middleman actor
[=](put_atom, network::native_socket fd, const actor_addr& whom, uint16_t port) {
auto hdl = add_tcp_doorman(fd);
add_published_actor(hdl, actor_cast<abstract_actor_ptr>(whom), port);
parent().notify<hook::actor_published>(whom, port);
},
[=](get_atom, network::native_socket fd, int64_t request_id, actor client, std::set<std::string>& expected_ifs) {
auto hdl = add_tcp_scribe(fd);
auto& ctx = m_ctx[hdl];
ctx.hdl = hdl;
ctx.handshake_data = client_handshake_data{};
auto& hdata = *ctx.handshake_data;
hdata.request_id = request_id;
hdata.client = client;
hdata.expected_ifs.swap(expected_ifs);
init_handshake_as_client(ctx);
},
// catch-all error handler
others() >> [=] {
CAF_LOG_INFO("received unexpected message: "
CAF_LOG_ERROR("received unexpected message: "
<< to_string(last_dequeued()));
}
};
......@@ -422,7 +441,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
}
case basp::server_handshake: {
CAF_REQUIRE(payload != nullptr);
if (ctx.handshake_data == nullptr) {
if (!ctx.handshake_data) {
CAF_LOG_INFO("received unexpected server handshake");
return close_connection;
}
......@@ -431,7 +450,6 @@ basp_broker::handle_basp_header(connection_context& ctx,
return close_connection;
}
ctx.remote_id = hdr.source_node;
ctx.handshake_data->remote_id = hdr.source_node;
binary_deserializer bd{payload->data(), payload->size(), &m_namespace};
auto remote_aid = bd.read<uint32_t>();
auto remote_ifs_size = bd.read<uint32_t>();
......@@ -441,6 +459,8 @@ basp_broker::handle_basp_header(connection_context& ctx,
remote_ifs.insert(std::move(str));
}
auto& ifs = ctx.handshake_data->expected_ifs;
auto hsclient = ctx.handshake_data->client;
auto hsid = ctx.handshake_data->request_id;
if (!std::includes(ifs.begin(), ifs.end(),
remote_ifs.begin(), remote_ifs.end())) {
auto tostr = [](const std::set<string>& what) -> string {
......@@ -481,15 +501,15 @@ basp_broker::handle_basp_header(connection_context& ctx,
+ iface_str;
}
// abort with error
send(ctx.handshake_data->client, atom("ERROR"), std::move(error_msg));
send(hsclient, error_atom{}, hsid, std::move(error_msg));
return close_connection;
}
auto nid = ctx.handshake_data->remote_id;
auto nid = hdr.source_node;
if (nid == node()) {
CAF_LOG_INFO("incoming connection from self: drop connection");
auto res = detail::singletons::get_actor_registry()->get(remote_aid);
send(ctx.handshake_data->client, atom("OK"), actor_cast<actor>(res));
ctx.handshake_data = nullptr;
send(hsclient, ok_atom{}, hsid, res->address());
ctx.handshake_data = none;
return close_connection;
}
if (!try_set_default_route(nid, ctx.hdl)) {
......@@ -497,21 +517,21 @@ basp_broker::handle_basp_header(connection_context& ctx,
<< " (re-use old one)");
auto proxy = m_namespace.get_or_put(nid, remote_aid);
// discard this peer; there's already an open connection
send(ctx.handshake_data->client, atom("OK"), actor_cast<actor>(proxy));
ctx.handshake_data = nullptr;
send(hsclient, ok_atom{}, hsid, proxy->address());
ctx.handshake_data = none;
return close_connection;
}
// finalize handshake
auto& buf = wr_buf(ctx.hdl);
binary_serializer bs(std::back_inserter(buf), &m_namespace);
write(bs, {node(), ctx.handshake_data->remote_id,
write(bs, {node(), nid,
invalid_actor_id, invalid_actor_id,
0, basp::client_handshake, 0});
// prepare to receive messages
auto proxy = m_namespace.get_or_put(nid, remote_aid);
ctx.published_actor = proxy;
send(ctx.handshake_data->client, atom("OK"), actor_cast<actor>(proxy));
ctx.handshake_data = nullptr;
send(hsclient, ok_atom{}, hsid, proxy->address());
ctx.handshake_data = none;
parent().notify<hook::new_connection_established>(nid);
break;
}
......@@ -622,19 +642,9 @@ bool basp_broker::try_set_default_route(const id_type& nid,
return false;
}
void basp_broker::init_client(connection_handle hdl,
client_handshake_data* data) {
CAF_LOG_TRACE("hdl = " << hdl.id());
auto& ctx = m_ctx[hdl];
ctx.hdl = hdl;
init_handshake_as_client(ctx, data);
}
void basp_broker::init_handshake_as_client(connection_context& ctx,
client_handshake_data* ptr) {
void basp_broker::init_handshake_as_client(connection_context& ctx) {
CAF_LOG_TRACE(CAF_ARG(this));
ctx.state = await_server_handshake;
ctx.handshake_data = ptr;
configure_read(ctx.hdl, receive_policy::exactly(basp::header_size));
}
......
......@@ -31,10 +31,13 @@
#include "caf/announce.hpp"
#include "caf/to_string.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/uniform_type_info.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/ripemd_160.hpp"
......@@ -173,6 +176,121 @@ void do_announce(const char* tname) {
using detail::make_counted;
using middleman_actor_base = middleman_actor::extend<
reacts_to<ok_atom, int64_t, actor_addr>,
reacts_to<error_atom, int64_t, std::string>
>::type;
class middleman_actor_impl : public middleman_actor_base::base {
public:
middleman_actor_impl(actor default_broker)
: m_broker(default_broker),
m_next_request_id(0) {
// nop
}
~middleman_actor_impl();
using get_op_result = either<ok_atom, actor_addr>::or_else<error_atom, std::string>;
using get_op_promise = typed_response_promise<get_op_result>;
middleman_actor_base::behavior_type make_behavior() {
return {
[=](put_atom, const actor_addr& whom, uint16_t port,
const std::string& addr, bool reuse_addr) {
return put(whom, port, addr.c_str(), reuse_addr);
},
[=](put_atom, const actor_addr& whom, uint16_t port,
const std::string& addr) {
return put(whom, port, addr.c_str());
},
[=](put_atom, const actor_addr& whom, uint16_t port, bool reuse_addr) {
return put(whom, port, nullptr, reuse_addr);
},
[=](put_atom, const actor_addr& whom, uint16_t port) {
return put(whom, port);
},
[=](get_atom, const std::string& host, uint16_t port,
std::set<std::string>& expected_ifs) {
return get(host, port, std::move(expected_ifs));
},
[=](get_atom, const std::string& host, uint16_t port) {
return get(host, port, std::set<std::string>());
},
[=](ok_atom ok, int64_t request_id, actor_addr result) {
auto i = m_pending_requests.find(request_id);
if (i == m_pending_requests.end()) {
CAF_LOG_ERROR("invalid request id: " << request_id);
return;
}
i->second.deliver(get_op_result{ok, result});
m_pending_requests.erase(i);
},
[=](error_atom error, int64_t request_id, std::string& reason) {
auto i = m_pending_requests.find(request_id);
if (i == m_pending_requests.end()) {
CAF_LOG_ERROR("invalid request id: " << request_id);
return;
}
i->second.deliver(get_op_result{error, std::move(reason)});
m_pending_requests.erase(i);
}
};
}
private:
either<ok_atom, uint16_t>::or_else<error_atom, std::string>
put(const actor_addr& whom, uint16_t port,
const char* in = nullptr, bool reuse_addr = false) {
network::native_socket fd;
uint16_t actual_port;
try {
// treat empty strings like nullptr
if (in != nullptr && in[0] == '\0') {
in = nullptr;
}
auto res = network::new_ipv4_acceptor_impl(port, in, reuse_addr);
fd = res.first;
actual_port = res.second;
}
catch (bind_failure& err) {
return {error_atom{}, std::string("bind_failure: ") + err.what()};
}
catch (network_error& err) {
return {error_atom{}, std::string("network_error: ") + err.what()};
}
send(m_broker, put_atom{}, fd, whom, actual_port);
return {ok_atom{}, actual_port};
}
get_op_promise get(const std::string& host, uint16_t port,
std::set<std::string> expected_ifs) {
get_op_promise result = make_response_promise();
try {
auto fd = network::new_ipv4_connection_impl(host, port);
auto req_id = m_next_request_id++;
send(m_broker, get_atom{}, fd, req_id, actor{this}, std::move(expected_ifs));
m_pending_requests.insert(std::make_pair(req_id, result));
}
catch (network_error& err) {
// fullfil promise immediately
std::string msg = "network_error: ";
msg += err.what();
result.deliver(get_op_result{error_atom{}, std::move(msg)});
}
return result;
}
actor m_broker;
int64_t m_next_request_id;
std::map<int64_t, get_op_promise> m_pending_requests;
};
middleman_actor_impl::~middleman_actor_impl() {
// nop
}
middleman* middleman::instance() {
auto sid = detail::singletons::middleman_plugin_id;
auto fac = [] { return new middleman; };
......@@ -205,6 +323,8 @@ void middleman::initialize() {
do_announce<connection_handle>("caf::io::connection_handle");
do_announce<new_connection_msg>("caf::io::new_connection_msg");
do_announce<new_data_msg>("caf::io::new_data_msg");
actor mgr = get_named_broker<basp_broker>(atom("_BASP"));
m_manager = spawn_typed<middleman_actor_impl, detached + hidden>(mgr);
}
void middleman::stop() {
......@@ -224,6 +344,14 @@ void middleman::stop() {
m_backend_supervisor.reset();
m_thread.join();
m_named_brokers.clear();
scoped_actor self(true);
self->monitor(m_manager);
self->send_exit(m_manager, exit_reason::user_shutdown);
self->receive(
[](const down_msg&) {
// nop
}
);
}
void middleman::dispose() {
......@@ -238,5 +366,13 @@ middleman::~middleman() {
// nop
}
middleman_actor middleman::actor_handle() {
return m_manager;
}
middleman_actor get_middleman_actor() {
return middleman::instance()->actor_handle();
}
} // namespace io
} // namespace caf
......@@ -35,39 +35,22 @@ namespace io {
uint16_t publish_impl(abstract_actor_ptr whom, uint16_t port,
const char* in, bool reuse_addr) {
using namespace detail;
auto mm = middleman::instance();
std::string str;
if (in != nullptr) {
str = in;
}
auto mm = get_middleman_actor();
scoped_actor self;
actor selfhdl = self;
mm->run_later([&] {
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
try {
auto hdl = mm->backend().add_tcp_doorman(bro.get(), port, in, reuse_addr);
bro->add_published_actor(std::move(hdl.first), whom, hdl.second);
mm->notify<hook::actor_published>(whom->address(), hdl.second);
anon_send(selfhdl, atom("OK"), hdl.second);
}
catch (bind_failure& e) {
anon_send(selfhdl, atom("BIND_FAIL"), e.what());
}
catch (network_error& e) {
anon_send(selfhdl, atom("ERROR"), e.what());
}
});
uint16_t bound_port = 0;
// block caller and re-throw exception here in case of an error
self->receive(
on(atom("OK"), arg_match) >> [&](uint16_t p) {
bound_port = p;
},
on(atom("BIND_FAIL"), arg_match) >> [](std::string& str) {
throw bind_failure(std::move(str));
uint16_t result;
self->sync_send(mm, put_atom{}, whom->address(), port, str, reuse_addr).await(
[&](ok_atom, uint16_t res) {
result = res;
},
on(atom("ERROR"), arg_match) >> [](std::string& str) {
throw network_error(std::move(str));
[&](error_atom, std::string& msg) {
throw network_error(std::move(msg));
}
);
return bound_port;
return result;
}
} // namespace io
......
......@@ -43,34 +43,15 @@ namespace io {
abstract_actor_ptr remote_actor_impl(std::set<std::string> ifs,
const std::string& host, uint16_t port) {
auto mm = middleman::instance();
auto mm = get_middleman_actor();
scoped_actor self;
actor selfhdl = self;
basp_broker::client_handshake_data hdata{invalid_node_id,
selfhdl, std::move(ifs)};
mm->run_later([&] {
std::string err;
try {
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
auto hdl = mm->backend().add_tcp_scribe(bro.get(), host, port);
bro->init_client(hdl, &hdata);
}
catch (std::exception& e) {
err = e.what();
}
// accessing variables from the outer scope inside the
// catch block triggers a silly compiler error on GCC 4.7
if (!err.empty()) {
anon_send(selfhdl, atom("ERROR"), std::move(err));
}
});
abstract_actor_ptr result;
self->receive(
on(atom("OK"), arg_match) >> [&](const actor& res) {
self->sync_send(mm, get_atom{}, std::move(host), port, std::move(ifs)).await(
[&](ok_atom, actor_addr res) {
result = actor_cast<abstract_actor_ptr>(res);
},
on(atom("ERROR"), arg_match) >> [](std::string& str) {
throw network_error(std::move(str));
[&](error_atom, std::string& msg) {
throw network_error(std::move(msg));
}
);
return result;
......
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