Commit ba460d07 authored by Dominik Charousset's avatar Dominik Charousset

Reduce connections by caching remote_actor result

Relates #390
parent 4e5d6544
......@@ -80,13 +80,20 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::listener
void learned_new_node_indirectly(const node_id& nid) override;
// stores meta information for open connections
struct connection_context {
// denotes what message we expect from the remote node next
basp::connection_state cstate;
// our currently processed BASP header
basp::header hdr;
// the connection handle for I/O operations
connection_handle hdl;
// network-agnostic node identifier
node_id id;
// connected port
uint16_t remote_port;
maybe<response_promise> callback;
// pending operations to be performed after handhsake completed
std::function<void (const maybe<message>&)> callback;
};
void set_context(connection_handle hdl);
......@@ -119,9 +126,31 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// routing paths by forming a mesh between all nodes
bool enable_automatic_connections = false;
// returns the node identifier of the underlying BASP instance
const node_id& this_node() const {
return instance.this_node();
}
// and endpoint is identified via host and port
using endpoint = std::pair<std::string, uint16_t>;
// stores the result of a `remote_actor()`
using remote_actor_res = std::tuple<node_id, actor_addr,
std::set<std::string>>;
// stores the state of connected endpoints, `i->second.empty()` indicates
// that a connection has been established but the handshake is still
// pending, in which case it is safe to attach to `ctx[i->first].callback`
using endpoint_data = std::pair<connection_handle, maybe<remote_actor_res>>;
using endpoint_map = std::map<endpoint, endpoint_data>;
// caches the result of all `remote_actor()` calls to avoid
// spamming connections for connecting to the same node multiple times
endpoint_map connected_endpoints;
// "reverse lookup" for finding credentials to a connection
endpoint_map::iterator find_endpoint(connection_handle x);
};
/// A broker implementation for the Binary Actor System Protocol (BASP).
......
......@@ -120,18 +120,21 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
auto& cb = this_context->callback;
if (! cb)
return;
actor_addr addr;
auto cleanup = detail::make_scope_guard([&] {
cb = none;
cb = nullptr;
auto e = connected_endpoints.end();
auto i = find_endpoint(this_context->hdl);
if (i == e)
return;
// store result of this `remote_actor()` call for later retrieval
i->second.second = std::make_tuple(nid, addr, sigs);
});
if (aid == invalid_actor_id) {
// can occur when connecting to the default port of a node
cb->deliver(make_message(ok_atom::value,
nid,
actor_addr{invalid_actor_addr},
std::move(sigs)));
cb(make_message(ok_atom::value, nid, addr, sigs));
return;
}
actor_addr addr;
if (nid == this_node()) {
// connected to self
addr = system().registry().get(aid).first;
......@@ -144,8 +147,7 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
}
if (addr.node() != system().node())
known_remotes.emplace(nid, std::make_pair(this_context->remote_port, addr));
cb->deliver(make_message(ok_atom::value, nid, addr, std::move(sigs)));
this_context->callback = none;
cb(make_message(ok_atom::value, nid, addr, sigs));
}
void basp_broker_state::purge_state(const node_id& nid) {
......@@ -353,7 +355,7 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
// gotcha! send scribe to our BASP broker
// to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
helper->send(bb, connect_atom::value, hdl, port);
helper->send(bb, connect_atom::value, hdl, addr, port);
return;
}
catch (...) {
......@@ -420,7 +422,7 @@ void basp_broker_state::set_context(connection_handle hdl) {
hdl,
invalid_node_id,
0,
none}).first;
nullptr}).first;
}
this_context = &i->second;
}
......@@ -433,12 +435,21 @@ bool basp_broker_state::erase_context(connection_handle hdl) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
ref.callback(sec::disconnect_during_handshake);
}
ctx.erase(i);
return true;
}
basp_broker_state::endpoint_map::iterator
basp_broker_state::find_endpoint(connection_handle y) {
auto pred = [&](const endpoint_map::value_type& x) {
return x.second.first == y;
};
auto e = connected_endpoints.end();
return std::find_if(connected_endpoints.begin(), e, pred);
}
/******************************************************************************
* basp_broker *
******************************************************************************/
......@@ -582,9 +593,41 @@ behavior basp_broker::make_behavior() {
state.instance.add_published_actor(port, whom, std::move(sigs));
},
// received from middleman actor (delegated)
[=](connect_atom, connection_handle hdl, uint16_t port) {
[=](connect_atom, connection_handle hdl,
std::string& hostname, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hdl.id()));
// store original context via response promise
auto rp = make_response_promise();
// check whether we already know these credentials
auto ep = std::make_pair(std::move(hostname), port);
auto i = state.connected_endpoints.find(ep);
if (i != state.connected_endpoints.end()) {
auto& x = i->second;
if (x.second.invalid()) {
rp.deliver(x.second.error());
} else if (x.second.empty()) {
// in this state, we have a connection but are waiting
// for the handshake to complete and we can attach to the
// respective callback
auto j = state.ctx.find(x.first);
CAF_ASSERT(j != ctx.end());
CAF_ASSERT(j->second.callback != nullptr);
auto f = j->second.callback; // store previous callback
j->second.callback = [f, rp](const maybe<message>& x) {
f(x); // run original callback(s) first
if (x)
rp.deliver(*x);
else if (x.invalid())
rp.deliver(x.error());
};
} else {
auto& tup = *(x.second);
rp.deliver(make_message(ok_atom::value, get<0>(tup),
get<1>(tup), get<2>(tup)));
}
return;
}
// initiate handshake in case we aren't connected to these credentials
try {
assign_tcp_scribe(hdl);
}
......@@ -597,7 +640,15 @@ behavior basp_broker::make_behavior() {
ctx.hdl = hdl;
ctx.remote_port = port;
ctx.cstate = basp::await_header;
ctx.callback = rp;
ctx.callback = [rp](const maybe<message>& res) {
if (res.valid())
rp.deliver(*res);
else if (res.invalid())
rp.deliver(res.error());
};
using endpoint_data = basp_broker_state::endpoint_data;
state.connected_endpoints.emplace(std::move(ep),
endpoint_data{hdl, none});
// await server handshake
configure_read(hdl, receive_policy::exactly(basp::header_size));
},
......
......@@ -105,7 +105,7 @@ public:
std::set<std::string> sigs;
return put(port, whom, sigs, addr.c_str(), reuse);
},
[=](connect_atom, const std::string& hostname, uint16_t port) -> get_res {
[=](connect_atom, std::string& hostname, uint16_t port) -> get_res {
CAF_LOG_TRACE(CAF_ARG(hostname) << CAF_ARG(port));
connection_handle hdl;
try {
......@@ -114,7 +114,8 @@ public:
// nop
}
if (hdl != invalid_connection_handle) {
delegate(broker_, connect_atom::value, hdl, port);
delegate(broker_, connect_atom::value, hdl,
std::move(hostname), port);
} else {
auto rp = make_response_promise();
rp.deliver(sec::cannot_connect_to_node);
......
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