Commit eefbae4b authored by Joseph Noir's avatar Joseph Noir

Add addresse exchange when a proxy is created

parent 4658d45a
......@@ -86,6 +86,9 @@ public:
/// Called whenever BASP learns the ID of a remote node.
virtual void learned_new_node(const node_id& nid) = 0;
/// Get contact information for `nid` and establish communication.
virtual void establish_communication(const node_id& nid) = 0;
/// Called if a heartbeat was received from `nid`
virtual void handle_heartbeat(const node_id& nid) = 0;
......@@ -255,12 +258,12 @@ public:
uint16_t sequence_number = 0);
/// Writes the client handshake to `buf`.
static void write_client_handshake(execution_unit* ctx,
buffer_type& buf,
const node_id& remote_side,
const node_id& this_node,
const std::string& app_identifier,
uint16_t sequence_number = 0);
void write_client_handshake(execution_unit* ctx,
buffer_type& buf,
const node_id& remote_side,
const node_id& this_node,
const std::string& app_identifier,
uint16_t sequence_number = 0);
/// Writes the client handshake to `buf`.
void write_client_handshake(execution_unit* ctx,
......@@ -334,8 +337,6 @@ public:
return false;
}
// Close this connection if we already established communication.
// FIXME: Should we allow multiple "connections" over different
// transport protocols?
if (tbl_.lookup(hdr.source_node).hdl) {
CAF_LOG_INFO("close connection since we already have a "
"connection: " << CAF_ARG(hdr.source_node));
......@@ -345,7 +346,9 @@ public:
// Add this node to our contacts.
CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node));
tbl_.add(hdr.source_node, hdl);
tbl_.addresses(hdr.source_node, addrs);
auto config_server = system().registry().get(atom("ConfigServ"));
anon_send(actor_cast<actor>(config_server), put_atom::value,
to_string(hdr.source_node), make_message(addrs));
// TODO: Add addresses to share with other nodes?
// Write handshake as client in response.
if (tcp_based)
......@@ -376,8 +379,9 @@ public:
if (e)
return false;
}
auto new_node = (this_node() != hdr.source_node
&& !tbl_.lookup(hdr.source_node).known);
// Handshakes were only exchanged if `hdl` is set.
auto lr = tbl_.lookup(hdr.source_node);
auto new_node = (this_node() != hdr.source_node && !lr.hdl);
if (!new_node) {
if (tcp_based) {
CAF_LOG_INFO("received second client handshake:"
......@@ -387,8 +391,14 @@ public:
} else {
// Add this node to our contacts.
CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node));
tbl_.add(hdr.source_node, hdl);
tbl_.addresses(hdr.source_node, addrs);
// Either add a new node or add the handle to a known one.
if (lr.known)
tbl_.handle(hdr.source_node, hdl);
else
tbl_.add(hdr.source_node, hdl);
auto config_server = system().registry().get(atom("ConfigServ"));
anon_send(actor_cast<actor>(config_server), put_atom::value,
to_string(hdr.source_node), make_message(addrs));
}
// Since udp is unreliable we answer, maybe our message was lost.
if (!tcp_based) {
......@@ -398,6 +408,8 @@ public:
}
// We have to call this after `write_server_handshake` because
// `learned_new_node` expects there to be an entry in the routing table.
// TODO: Can we move this in the else block above with the changes to
// the routing table?
if (new_node) {
callee_.learned_new_node(hdr.source_node);
// TODO: Only send buffered messaged for new nodes?
......
......@@ -123,13 +123,6 @@ public:
/// or `none` if the node is unknown.
optional<endpoint_handle> handle(const node_id& nid);
/// Set `addrs` as the addresses to reach `nid`.
bool addresses(const node_id& nid, address_map addrs);
/// Set `addrs` as the addresses to reach `nid` with `proto`.
bool addresses(const node_id& nid, network::protocol::transport proto,
address_endpoint addrs);
/// Get the addresses to reach `nid` or `none` if the node is unknown.
optional<const address_map&> addresses(const node_id& nid);
......@@ -145,8 +138,6 @@ public:
struct node_info {
/// Handle for the node if communication is established.
optional<endpoint_handle> hdl;
/// Interfaces of the nodes for sharing with neighbors.
address_map addrs;
/// The endpoint who told us about the node.
optional<node_id> origin;
};
......
......@@ -82,6 +82,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// performs bookkeeping such as managing `spawn_servers`
void learned_new_node(const node_id& nid) override;
// get contact information for `nid` and establish communication
void establish_communication(const node_id& nid) override;
// inherited from basp::instance::callee
uint16_t next_sequence_number(connection_handle hdl) override;
......
......@@ -49,10 +49,11 @@ struct connection_helper_state {
behavior datagram_connection_broker(broker* self,
uint16_t port,
network::address_listing addresses,
actor system_broker);
actor system_broker,
basp::instance* instance);
behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b);
actor b, basp::instance* i);
} // namespace io
} // namespace caf
......@@ -99,15 +99,13 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
// how to contact B.
auto lr = instance.tbl().lookup(nid);
if (nid != this_context->id && !lr.known) {
// TODO: Try to establish communication with the new node.
CAF_CRITICAL("Not implemented.");
instance.tbl().add(nid, this_context->id);
establish_communication(nid);
}
// TODO: Everything below has to happen once we establish communication?
// I'm not sure yet, the functors can be attached earlier and we could
// send trigger an error message if we cannot contact the remote node.
// We need to tell remote side we are watching this actor now;
// use a direct route if possible, i.e., when talking to a third node.
// TODO: Should this communication already be established?
// TODO: Communication setup might still be in progress.
/*
if (lr.known && !lr.hdl) {
// This happens if and only if we don't have a path to `nid`
// and current_context_->hdl has been blacklisted.
......@@ -115,6 +113,7 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
"running on a node we don't have a route to");
return nullptr;
}
*/
// Create proxy and add functor that will be called if we
// receive a kill_proxy_instance message.
auto mm = &system().middleman();
......@@ -135,13 +134,21 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
CAF_LOG_INFO("successfully created proxy instance, "
"write announce_proxy_instance:"
<< CAF_ARG(nid) << CAF_ARG(aid));
auto& ctx = *this_context;
// Tell remote side we are monitoring this actor now.
instance.write_announce_proxy(self->context(),
get_buffer(this_context->hdl),
nid, aid,
ctx.requires_ordering ? ctx.seq_outgoing++ : 0);
instance.flush(*lr.hdl);
// TODO: Can it happen that things have changed here?
lr = instance.tbl().lookup(nid);
if (lr.hdl) {
auto& ctx = *this_context;
// Tell remote side we are monitoring this actor now.
instance.write_announce_proxy(self->context(),
get_buffer(nid),
nid, aid,
ctx.requires_ordering ? ctx.seq_outgoing++ : 0);
instance.flush(*lr.hdl);
} else {
instance.write_announce_proxy(self->context(),
get_buffer(nid),
nid, aid,0);
}
mm->notify<hook::new_remote_actor>(res);
return res;
}
......@@ -403,6 +410,61 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
}
}
void basp_broker_state::establish_communication(const node_id& nid) {
// TODO: Split this by functionality, address query & connecting?
CAF_ASSERT(this_context != nullptr);
CAF_LOG_TRACE(CAF_ARG(nid));
learned_new_node(nid);
if (!enable_automatic_connections)
return;
// this member function gets only called once, after adding a new
// indirect connection to the routing table; hence, spawning
// our helper here exactly once and there is no need to track
// in-flight connection requests
if (instance.tbl().lookup(nid).hdl) {
CAF_LOG_ERROR("establish_communication called with established connection");
return;
}
auto origin = instance.tbl().origin(nid);
if (!origin) {
CAF_LOG_ERROR("establish_communication called, but no node known "
"to ask for contact information");
return;
}
auto ehdl = instance.tbl().handle(*origin);
if (!ehdl) {
CAF_LOG_ERROR("establish_communication called, but node with contact "
"information is no longer reachable");
return;
}
auto hdl = std::move(*ehdl);
using namespace detail;
auto try_connect = [&](std::string item) {
auto tmp = system().config().middleman_detach_utility_actors
? system().spawn<detached + hidden>(connection_helper, self, &instance)
: system().spawn<hidden>(connection_helper, self, &instance);
system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp));
auto writer = make_callback([&item](serializer& sink) -> error {
auto name_atm = atom("ConfigServ");
std::vector<actor_id> stages;
auto msg = make_message(get_atom::value, std::move(item));
return sink(name_atm, stages, msg);
});
basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag,
0, 0, this_node(), nid, tmp.id(), invalid_actor_id,
visit(seq_num_visitor{this}, hdl)};
instance.write(self->context(), get_buffer(hdl),
hdr, &writer);
instance.flush(hdl);
};
auto item = to_string(nid);
if (enable_tcp)
try_connect(item);
if (enable_udp)
try_connect(item);
}
void basp_broker_state::set_context(connection_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
auto i = ctx_tcp.find(hdl);
......@@ -661,6 +723,8 @@ behavior basp_broker::make_behavior() {
if (res) {
auto port = res->second;
auto addrs = network::interfaces::list_addresses(false);
state.instance.tbl().local_addresses(network::protocol::tcp,
{port, addrs});
auto config_server = system().registry().get(atom("ConfigServ"));
send(actor_cast<actor>(config_server), put_atom::value,
"basp.default-connectivity-tcp",
......@@ -672,6 +736,8 @@ behavior basp_broker::make_behavior() {
if (res) {
auto port = res->second;
auto addrs = network::interfaces::list_addresses(false);
state.instance.tbl().local_addresses(network::protocol::udp,
{port, addrs});
auto config_server = system().registry().get(atom("ConfigServ"));
send(actor_cast<actor>(config_server), put_atom::value,
"basp.default-connectivity-udp",
......
......@@ -34,7 +34,8 @@ const char* connection_helper_state::name = "connection_helper";
behavior datagram_connection_broker(broker* self, uint16_t port,
network::address_listing addresses,
actor system_broker) {
actor system_broker,
basp::instance* instance) {
auto& mx = self->system().middleman().backend();
auto& this_node = self->system().node();
auto& app_id = self->system().config().middleman_app_identifier;
......@@ -44,10 +45,10 @@ behavior datagram_connection_broker(broker* self, uint16_t port,
if (eptr) {
auto hdl = (*eptr)->hdl();
self->add_datagram_servant(std::move(*eptr));
basp::instance::write_client_handshake(self->context(),
self->wr_buf(hdl),
none, this_node,
app_id);
instance->write_client_handshake(self->context(),
self->wr_buf(hdl),
none, this_node,
app_id);
}
}
}
......@@ -59,15 +60,35 @@ behavior datagram_connection_broker(broker* self, uint16_t port,
},
after(autoconnect_timeout) >> [=]() {
CAF_LOG_TRACE(CAF_ARG(""));
// nothing heard in about 10 minutes... just a call it a day, then
// nothing heard in about 10 minutes... just call it a day, then
CAF_LOG_INFO("aborted direct connection attempt after 10min");
self->quit(exit_reason::user_shutdown);
}
};
}
bool establish_stream_connection(stateful_actor<connection_helper_state>* self,
const actor& b, uint16_t port,
network::address_listing& addresses) {
auto& mx = self->system().middleman().backend();
for (auto& kvp : addresses) {
for (auto& addr : kvp.second) {
auto hdl = mx.new_tcp_scribe(addr, port);
if (hdl) {
// gotcha! send scribe to our BASP broker
// to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
self->send(b, connect_atom::value, *hdl, port);
return true;
}
}
}
return false;
}
behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b) {
actor b, basp::instance* i) {
CAF_LOG_TRACE(CAF_ARG(b));
self->monitor(b);
self->set_down_handler([=](down_msg& dm) {
......@@ -84,35 +105,43 @@ behavior connection_helper(stateful_actor<connection_helper_state>* self,
msg.apply({
[&](uint16_t port, network::address_listing& addresses) {
if (item == "basp.default-connectivity-tcp") {
auto& mx = self->system().middleman().backend();
for (auto& kvp : addresses) {
for (auto& addr : kvp.second) {
auto hdl = mx.new_tcp_scribe(addr, port);
if (hdl) {
// gotcha! send scribe to our BASP broker
// to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
self->send(b, connect_atom::value, *hdl, port);
return;
}
}
}
CAF_LOG_INFO("could not connect to node directly");
if (!establish_stream_connection(self, b, port, addresses))
CAF_LOG_INFO("could not connect to node");
} else if (item == "basp.default-connectivity-udp") {
// create new broker to try addresses for communication via UDP
if (self->system().config().middleman_detach_utility_actors) {
self->system().middleman().spawn_broker<detached + hidden>(
datagram_connection_broker, port, std::move(addresses), b
datagram_connection_broker, port, std::move(addresses), b, i
);
} else {
self->system().middleman().spawn_broker<hidden>(
datagram_connection_broker, port, std::move(addresses), b
datagram_connection_broker, port, std::move(addresses), b, i
);
}
} else {
CAF_LOG_INFO("aborted direct connection attempt, unknown item: "
<< CAF_ARG(item));
}
},
[&](basp::routing_table::address_map& addrs) {
if (addrs.count(network::protocol::tcp) > 0) {
auto eps = addrs[network::protocol::tcp];
if (!establish_stream_connection(self, b, eps.first, eps.second))
CAF_LOG_ERROR("could not connect to node ");
}
if (addrs.count(network::protocol::udp) > 0) {
auto eps = addrs[network::protocol::udp];
// create new broker to try addresses for communication via UDP
if (self->system().config().middleman_detach_utility_actors) {
self->system().middleman().spawn_broker<detached + hidden>(
datagram_connection_broker, eps.first, std::move(eps.second), b, i
);
} else {
self->system().middleman().spawn_broker<hidden>(
datagram_connection_broker, eps.first, std::move(eps.second), b, i
);
}
}
}
});
},
......
......@@ -59,13 +59,13 @@ void routing_table::add(const node_id& nid, const endpoint_handle& hdl) {
CAF_ASSERT(nid_by_hdl_.count(hdl) == 0);
CAF_ASSERT(node_information_base_.count(nid) == 0);
nid_by_hdl_.emplace(hdl, nid);
node_information_base_[nid] = node_info{hdl, {}, none};
node_information_base_[nid] = node_info{hdl, none};
parent_->parent().notify<hook::new_connection_established>(nid);
}
void routing_table::add(const node_id& nid, const node_id& origin) {
CAF_ASSERT(node_information_base_.count(nid) == 0);
node_information_base_[nid] = node_info{none, {}, origin};
node_information_base_[nid] = node_info{none, origin};
// TODO: Some new related hook?
//parent_->parent().notify<hook::new_connection_established>(nid);
}
......@@ -73,7 +73,7 @@ void routing_table::add(const node_id& nid, const node_id& origin) {
void routing_table::add(const node_id& nid) {
//CAF_ASSERT(hdl_by_nid_.count(nid) == 0);
CAF_ASSERT(node_information_base_.count(nid) == 0);
node_information_base_[nid] = node_info{none, {}, none};
node_information_base_[nid] = node_info{none, none};
// TODO: Some new related hook?
//parent_->parent().notify<hook::new_connection_established>(nid);
}
......@@ -119,33 +119,6 @@ routing_table::handle(const node_id& nid) {
return i->second.hdl;
}
bool routing_table::addresses(const node_id& nid,
routing_table::address_map addrs) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
i->second.addrs = addrs;
return true;
}
bool routing_table::addresses(const node_id& nid,
network::protocol::transport proto,
routing_table::address_endpoint addrs) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
i->second.addrs[proto] = addrs;
return true;
}
optional<const routing_table::address_map&>
routing_table::addresses(const node_id& nid) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return none;
return i->second.addrs;
}
void routing_table::local_addresses(network::protocol::transport key,
routing_table::address_endpoint addrs) {
local_addrs_[key] = addrs;
......
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