Commit 7bedcd63 authored by Joseph Noir's avatar Joseph Noir

Rework automatic connections to new nodes

parent adc4041d
......@@ -80,14 +80,8 @@ public:
std::vector<strong_actor_ptr>& forwarding_stack,
message& msg) = 0;
/// Called whenever BASP learns the ID of a remote node
/// to which it does not have a direct connection.
virtual void learned_new_node_directly(const node_id& nid,
bool was_known_indirectly) = 0;
/// Called whenever BASP learns the ID of a remote node
/// to which it does not have a direct connection.
virtual void learned_new_node_indirectly(const node_id& nid) = 0;
/// Called whenever BASP learns the ID of a remote node.
virtual void learned_new_node(const node_id& nid) = 0;
/// Called if a heartbeat was received from `nid`
virtual void handle_heartbeat(const node_id& nid) = 0;
......@@ -107,9 +101,19 @@ public:
return namespace_.system().config();
}
/// Send messages that were buffered while connectivity establishment
/// was pending using `hdl`.
virtual void send_buffered_messages(execution_unit* ctx, node_id nid,
connection_handle hdl) = 0;
/// Returns a reference to the sent buffer.
virtual buffer_type& get_buffer(connection_handle hdl) = 0;
/// Returns a reference to a buffer to be sent to node with `nid`.
/// If communication with the node is esstablished, it picks the first
/// available handle, otherwise a buffer for a pending message is returned.
virtual buffer_type& get_buffer(node_id nid) = 0;
/// Flushes the underlying write buffer of `hdl`.
virtual void flush(connection_handle hdl) = 0;
......@@ -135,15 +139,15 @@ public:
void handle_heartbeat(execution_unit* ctx);
/// Returns a route to `target` or `none` on error.
optional<routing_table::route> lookup(const node_id& target);
routing_table::lookup_result lookup(const node_id& target);
/// Flushes the underlying buffer of `path`.
void flush(const routing_table::route& path);
void flush(connection_handle hdl);
/// Sends a BASP message and implicitly flushes the output buffer of `r`.
/// This function will update `hdr.payload_len` if a payload was written.
void write(execution_unit* ctx, const routing_table::route& r,
header& hdr, payload_writer* writer = nullptr);
void write(execution_unit* ctx, connection_handle hdl, header& hdr,
payload_writer* writer = nullptr);
/// Adds a new actor to the map of published actors.
void add_published_actor(uint16_t port,
......
......@@ -43,64 +43,69 @@ public:
virtual ~routing_table();
/// Describes a routing path to a node.
struct route {
const node_id& next_hop;
connection_handle hdl;
/// Result for a lookup of a node.
struct lookup_result {
/// Tracks whether the node is already known.
bool known;
/// Servant handle to communicate with the node -- if already created.
optional<connection_handle> hdl;
};
/// Describes a function object for erase operations that
/// is called for each indirectly lost connection.
using erase_callback = callback<const node_id&>;
/// Returns a route to `target` or `none` on error.
optional<route> lookup(const node_id& target);
/// Returns the ID of the peer connected via `hdl` or
/// Returns the ID of the peer reachable via `hdl` or
/// `none` if `hdl` is unknown.
node_id lookup_direct(const connection_handle& hdl) const;
/// Returns the handle offering a direct connection to `nid` or
/// `invalid_connection_handle` if no direct connection to `nid` exists.
optional<connection_handle> lookup_direct(const node_id& nid) const;
node_id lookup(const connection_handle& hdl) const;
/// Returns the next hop that would be chosen for `nid`
/// or `none` if there's no indirect route to `nid`.
node_id lookup_indirect(const node_id& nid) const;
/// Returns the state for communication with `nid` along with a handle
/// if communication is established or `none` if `nid` is unknown.
lookup_result lookup(const node_id& nid) const;
/// Adds a new direct route to the table.
/// Adds a new endpoint to the table.
/// @pre `hdl != invalid_connection_handle && nid != none`
void add_direct(const connection_handle& hdl, const node_id& nid);
void add(const node_id& nid, const connection_handle& hdl);
/// Adds a new indirect route to the table.
bool add_indirect(const node_id& hop, const node_id& dest);
/// Add a new endpoint to the table.
/// @pre `origin != none && nid != none`
void add(const node_id& nid, const node_id& origin);
/// Blacklist the route to `dest` via `hop`.
void blacklist(const node_id& hop, const node_id& dest);
/// Adds a new endpoint to the table that has no attached information.
/// that propagated information about the node.
/// @pre `nid != none`
void add(const node_id& nid);
/// Removes a direct connection and calls `cb` for any node
/// that became unreachable as a result of this operation,
/// including the node that is assigned as direct path for `hdl`.
void erase_direct(const connection_handle& hdl, erase_callback& cb);
void erase(const connection_handle& hdl, erase_callback& cb);
/// Removes any entry for indirect connection to `dest` and returns
/// `true` if `dest` had an indirect route, otherwise `false`.
bool erase_indirect(const node_id& dest);
/// Queries whether `dest` is reachable.
/// Queries whether `dest` is reachable directly.
bool reachable(const node_id& dest);
/// Removes all direct and indirect routes to `dest` and calls
/// `cb` for any node that became unreachable as a result of this
/// operation, including `dest`.
/// @returns the number of removed routes (direct and indirect)
size_t erase(const node_id& dest, erase_callback& cb);
/// Returns the parent broker.
inline abstract_broker* parent() {
return parent_;
}
/// Set the forwarding node that first mentioned `hdl`.
bool origin(const node_id& nid, const node_id& origin);
/// Get the forwarding node that first mentioned `hdl`
/// or `none` if the node is unknown.
optional<node_id> origin(const node_id& nid);
/// Set the handle for communication with `nid`.
bool handle(const node_id& nid, const connection_handle& hdl);
/// Get the handle for communication with `nid`
/// or `none` if the node is unknown.
optional<connection_handle> handle(const node_id& nid);
/// Get the addresses to reach `nid` or `none` if the node is unknown.
optional<const endpoint&> address(const node_id& nid);
/// Returns the local autoconnect endpoint.
const endpoint& autoconnect_endpoint();
......@@ -108,6 +113,14 @@ public:
void autoconnect_endpoint(uint16_t, network::address_listing);
public:
/// Entry to bundle information for a remote endpoint.
struct node_info {
/// Handle for the node if communication is established.
optional<connection_handle> hdl;
/// The endpoint who told us about the node.
optional<node_id> origin;
};
template <class Map, class Fallback>
typename Map::mapped_type
get_opt(const Map& m, const typename Map::key_type& k, Fallback&& x) const {
......@@ -117,16 +130,9 @@ public:
return std::forward<Fallback>(x);
}
using node_id_set = std::unordered_set<node_id>;
using indirect_entries = std::unordered_map<node_id, // dest
node_id_set>; // hop
abstract_broker* parent_;
std::unordered_map<connection_handle, node_id> direct_by_hdl_;
std::unordered_map<node_id, connection_handle> direct_by_nid_;
indirect_entries indirect_;
indirect_entries blacklist_;
std::unordered_map<connection_handle, node_id> nid_by_hdl_;
std::unordered_map<node_id, node_info> node_information_base_;
endpoint autoconnect_endpoint_;
};
......
......@@ -77,17 +77,17 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
std::vector<strong_actor_ptr>& stages, message& msg);
// performs bookkeeping such as managing `spawn_servers`
void learned_new_node(const node_id& nid);
void learned_new_node(const node_id& nid) override;
// inherited from basp::instance::callee
void learned_new_node_directly(const node_id& nid,
bool was_indirectly_before) override;
void send_buffered_messages(execution_unit* ctx, node_id nid,
connection_handle hdl) override;
// inherited from basp::instance::callee
void learned_new_node_indirectly(const node_id& nid) override;
buffer_type& get_buffer(connection_handle hdl) override;
// inherited from basp::instance::callee
buffer_type& get_buffer(connection_handle hdl) override;
buffer_type& get_buffer(node_id nid) override;
// inherited from basp::instance::callee
void flush(connection_handle hdl) override;
......@@ -102,6 +102,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
/// Cleans up any state for `hdl`.
void cleanup(connection_handle hdl);
/// Try to establish a connection to node with `nid`.
void connect(const node_id& nid);
// pointer to ourselves
broker* self;
......@@ -146,6 +149,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// actor
void handle_down_msg(down_msg&);
// buffer messages for nodes while connectivity is established
std::unordered_map<node_id, std::vector<buffer_type>> pending_connectivity;
static const char* name;
};
......
......@@ -46,13 +46,8 @@ struct connection_helper_state {
static const char* name;
};
behavior datagram_connection_broker(broker* self,
uint16_t port,
network::address_listing addresses,
actor system_broker);
behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b);
actor system_broker);
} // namespace io
} // namespace caf
This diff is collapsed.
......@@ -34,97 +34,45 @@ auto autoconnect_timeout = std::chrono::minutes(10);
const char* connection_helper_state::name = "connection_helper";
behavior datagram_connection_broker(broker* self, uint16_t port,
network::address_listing addresses,
actor system_broker) {
auto& mx = self->system().middleman().backend();
auto& this_node = self->system().node();
auto app_id = get_or(self->config(), "middleman.app-identifier",
defaults::middleman::app_identifier);
for (auto& kvp : addresses) {
for (auto& addr : kvp.second) {
auto eptr = mx.new_remote_udp_endpoint(addr, port);
if (eptr) {
auto hdl = (*eptr)->hdl();
self->add_datagram_servant(std::move(*eptr));
// TODO: UDP does not work at the moment.
// basp::instance::write_client_handshake(self->context(),
// self->wr_buf(hdl), this_node,
// app_id);
static_cast<void>(this_node);
static_cast<void>(hdl);
}
}
}
return {
[=](new_datagram_msg& msg) {
auto hdl = msg.handle;
self->send(system_broker, std::move(msg), self->take(hdl), port);
self->quit();
},
after(autoconnect_timeout) >> [=]() {
CAF_LOG_TRACE(CAF_ARG(""));
// nothing heard in about 10 minutes... just a call it a day, then
CAF_LOG_INFO("aborted direct connection attempt after 10min");
self->quit(exit_reason::user_shutdown);
}
};
}
behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b) {
CAF_LOG_TRACE(CAF_ARG(b));
self->monitor(b);
actor system_broker) {
CAF_LOG_TRACE(CAF_ARG(system_broker));
self->monitor(system_broker);
self->set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
self->quit(std::move(dm.reason));
});
return {
// this config is send from the remote `ConfigServ`
// this config is send from the remote `PeerServ`
[=](const std::string& item, message& msg) {
CAF_LOG_TRACE(CAF_ARG(item) << CAF_ARG(msg));
CAF_LOG_DEBUG("received requested config:" << CAF_ARG(msg));
CAF_LOG_DEBUG("received requested address:" << CAF_ARG(msg));
// whatever happens, we are done afterwards
self->quit();
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;
}
[&](basp::routing_table::endpoint& ep) {
auto port = ep.first;
auto& addrs = ep.second;
auto& mx = self->system().middleman().backend();
for (auto& kvp : addrs) {
for (auto& addr : kvp.second) {
auto hdl = mx.new_tcp_scribe(addr, port);
if (hdl) {
// Gotcha! Send scribe to our broker to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
self->send(system_broker, connect_atom::value, *hdl, port);
return;
}
}
CAF_LOG_INFO("could not connect to node directly");
} else if (item == "basp.default-connectivity-udp") {
auto& sys = self->system();
// create new broker to try addresses for communication via UDP
if (get_or(sys.config(), "middleman.attach-utility-actors", false))
self->system().middleman().spawn_broker<hidden>(
datagram_connection_broker, port, std::move(addresses), b
);
else
self->system().middleman().spawn_broker<detached + hidden>(
datagram_connection_broker, port, std::move(addresses), b
);
} else {
CAF_LOG_INFO("aborted direct connection attempt, unknown item: "
<< CAF_ARG(item));
}
CAF_LOG_ERROR("could not connect to node ");
}
});
},
after(autoconnect_timeout) >> [=] {
CAF_LOG_TRACE(CAF_ARG(""));
// nothing heard in about 10 minutes... just a call it a day, then
CAF_LOG_INFO("aborted direct connection attempt after 10min");
CAF_LOG_INFO("aborted connection attempt after 10min");
self->quit(exit_reason::user_shutdown);
}
};
......
......@@ -55,7 +55,7 @@ connection_state instance::handle(execution_unit* ctx,
callee_.purge_state(nid);
return none;
});
tbl_.erase_direct(dm.handle, cb);
tbl_.erase(dm.handle, cb);
return close_connection;
};
std::vector<char>* payload = nullptr;
......@@ -86,27 +86,27 @@ connection_state instance::handle(execution_unit* ctx,
void instance::handle_heartbeat(execution_unit* ctx) {
CAF_LOG_TRACE("");
for (auto& kvp: tbl_.direct_by_hdl_) {
for (auto& kvp: tbl_.nid_by_hdl_) {
CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second));
write_heartbeat(ctx, callee_.get_buffer(kvp.first));
callee_.flush(kvp.first);
}
}
optional<routing_table::route> instance::lookup(const node_id& target) {
routing_table::lookup_result instance::lookup(const node_id& target) {
return tbl_.lookup(target);
}
void instance::flush(const routing_table::route& path) {
callee_.flush(path.hdl);
void instance::flush(connection_handle hdl) {
callee_.flush(hdl);
}
void instance::write(execution_unit* ctx, const routing_table::route& r,
void instance::write(execution_unit* ctx, connection_handle hdl,
header& hdr, payload_writer* writer) {
CAF_LOG_TRACE(CAF_ARG(hdr));
CAF_ASSERT(hdr.payload_len == 0 || writer != nullptr);
write(ctx, callee_.get_buffer(r.hdl), hdr, writer);
flush(r);
write(ctx, callee_.get_buffer(hdl), hdr, writer);
flush(hdl);
}
void instance::add_published_actor(uint16_t port,
......@@ -169,8 +169,8 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(receiver)
<< CAF_ARG(mid) << CAF_ARG(msg));
CAF_ASSERT(receiver && system().node() != receiver->node());
auto path = lookup(receiver->node());
if (!path) {
auto lr = lookup(receiver->node());
if (!lr.known) {
notify<hook::message_sending_failed>(sender, receiver, mid, msg);
return false;
}
......@@ -180,9 +180,14 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
});
header hdr{message_type::dispatch_message, 0, 0, mid.integer_value(),
sender ? sender->id() : invalid_actor_id, receiver->id()};
write(ctx, callee_.get_buffer(path->hdl), hdr, &writer);
flush(*path);
notify<hook::message_sent>(sender, path->next_hop, receiver, mid, msg);
if (lr.hdl) {
auto hdl = std::move(*lr.hdl);
write(ctx, callee_.get_buffer(hdl), hdr, &writer);
flush(hdl);
} else {
write(ctx, callee_.get_buffer(receiver->node()), hdr, &writer);
}
notify<hook::message_sent>(sender, receiver->node(), receiver, mid, msg);
return true;
}
......@@ -296,16 +301,16 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
switch (hdr.operation) {
case message_type::server_handshake: {
node_id source_node;
basp::routing_table::endpoint autoconn_addr;
actor_id aid = invalid_actor_id;
std::set<std::string> sigs;
basp::routing_table::endpoint autoconn_addr;
if (!payload_valid()) {
CAF_LOG_ERROR("fail to receive the app identifier");
return false;
} else {
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
if (bd(source_node, remote_appid))
if (bd(source_node, remote_appid, autoconn_addr))
return false;
auto appid = get_or(callee_.config(), "middleman.app-identifier",
defaults::middleman::app_identifier);
......@@ -313,7 +318,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
CAF_LOG_ERROR("app identifier mismatch");
return false;
}
if (bd(aid, sigs, autoconn_addr))
if (bd(aid, sigs))
return false;
}
// Close self connection immediately after handshake.
......@@ -322,29 +327,28 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
callee_.finalize_handshake(source_node, aid, sigs);
return false;
}
auto lr = tbl_.lookup(source_node);
// Close redundant connections.
if (tbl_.lookup_direct(source_node)) {
if (lr.hdl) {
CAF_LOG_DEBUG("close redundant connection:" << CAF_ARG(source_node));
callee_.finalize_handshake(source_node, aid, sigs);
return false;
}
// Add new route to this node.
CAF_LOG_DEBUG("new connection:" << CAF_ARG(source_node));
tbl_.add_direct(hdl, source_node);
if (lr.known)
tbl_.handle(source_node, hdl);
else
tbl_.add(source_node, hdl);
// Store autoconnect address.
auto peer_server = system().registry().get(atom("PeerServ"));
anon_send(actor_cast<actor>(peer_server), put_atom::value,
to_string(source_node), make_message(std::move(autoconn_addr)));
// write handshake as client in response
auto path = tbl_.lookup(source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
return false;
}
write_client_handshake(ctx, callee_.get_buffer(path->hdl));
callee_.learned_new_node_directly(source_node, false);
write_client_handshake(ctx, callee_.get_buffer(hdl));
flush(hdl);
callee_.learned_new_node(source_node);
callee_.finalize_handshake(source_node, aid, sigs);
flush(*path);
callee_.send_buffered_messages(ctx, source_node, hdl);
break;
}
case message_type::client_handshake: {
......@@ -356,6 +360,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
node_id source_node;
std::string remote_appid;
basp::routing_table::endpoint autoconn_addr;
// TODO: Read addrs separately.
if (bd(source_node, remote_appid, autoconn_addr))
return false;
auto appid = get_if<std::string>(&callee_.config(),
......@@ -365,15 +370,21 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
CAF_LOG_ERROR("app identifier mismatch");
return false;
}
if (tbl_.lookup_direct(source_node)) {
auto lr = tbl_.lookup(source_node);
if (lr.hdl) {
CAF_LOG_DEBUG("received second client handshake:"
<< CAF_ARG(source_node));
break;
}
// Add route to this node.
// Add this node to our contacts.
CAF_LOG_DEBUG("new connection:" << CAF_ARG(source_node));
tbl_.add_direct(hdl, source_node);
callee_.learned_new_node_directly(source_node, false);
// Either add a new node or add the handle to a known one.
if (lr.known)
tbl_.handle(source_node, hdl);
else
tbl_.add(source_node, hdl);
callee_.learned_new_node(source_node);
callee_.send_buffered_messages(ctx, source_node, hdl);
// Store autoconnect address.
auto peer_server = system().registry().get(atom("PeerServ"));
anon_send(actor_cast<actor>(peer_server), put_atom::value,
......@@ -384,7 +395,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
// Sanity checks.
if (!payload_valid())
return false;
auto source_node = tbl_.lookup_direct(hdl);
auto source_node = tbl_.lookup(hdl);
if (source_node == none) {
CAF_LOG_ERROR("received dispatch_message before handshake");
return false;
......@@ -413,7 +424,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
break;
}
case message_type::announce_proxy: {
auto source_node = tbl_.lookup_direct(hdl);
auto source_node = tbl_.lookup(hdl);
if (source_node == none) {
CAF_LOG_ERROR("received announce_proxy before handshake");
return false;
......@@ -425,7 +436,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
// Sanity checks.
if (!payload_valid())
return false;
auto source_node = tbl_.lookup_direct(hdl);
auto source_node = tbl_.lookup(hdl);
if (source_node == none) {
CAF_LOG_ERROR("received announce_proxy before handshake");
return false;
......@@ -441,7 +452,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
break;
}
case message_type::heartbeat: {
auto source_node = tbl_.lookup_direct(hdl);
auto source_node = tbl_.lookup(hdl);
if (source_node == none) {
CAF_LOG_ERROR("received announce_proxy before handshake");
return false;
......
......@@ -389,7 +389,8 @@ void middleman::init(actor_system_config& cfg) {
.add_message_type<accept_handle>("@accept_handle")
.add_message_type<connection_handle>("@connection_handle")
.add_message_type<connection_passivated_msg>("@connection_passivated_msg")
.add_message_type<acceptor_passivated_msg>("@acceptor_passivated_msg");
.add_message_type<acceptor_passivated_msg>("@acceptor_passivated_msg")
.add_message_type<basp::routing_table::endpoint>("@autoconnect_endpoint");
// compute and set ID for this network node
node_id this_node{node_id::data::create_singleton()};
system().node_.swap(this_node);
......
......@@ -33,123 +33,87 @@ routing_table::~routing_table() {
// nop
}
optional<routing_table::route> routing_table::lookup(const node_id& target) {
auto hdl = lookup_direct(target);
if (hdl)
return route{target, *hdl};
// pick first available indirect route
auto i = indirect_.find(target);
if (i != indirect_.end()) {
auto& hops = i->second;
while (!hops.empty()) {
auto& hop = *hops.begin();
hdl = lookup_direct(hop);
if (hdl)
return route{hop, *hdl};
hops.erase(hops.begin());
}
}
return none;
node_id routing_table::lookup(const connection_handle& hdl) const {
return get_opt(nid_by_hdl_, hdl, none);
}
node_id routing_table::lookup_direct(const connection_handle& hdl) const {
return get_opt(direct_by_hdl_, hdl, none);
routing_table::lookup_result routing_table::lookup(const node_id& nid) const {
auto i = node_information_base_.find(nid);
if (i != node_information_base_.end())
return {true, i->second.hdl};
return {false, none};
}
optional<connection_handle>
routing_table::lookup_direct(const node_id& nid) const {
auto i = direct_by_nid_.find(nid);
if (i != direct_by_nid_.end())
return i->second;
return none;
void routing_table::erase(const connection_handle& hdl, erase_callback& cb) {
auto i = nid_by_hdl_.find(hdl);
if (i == nid_by_hdl_.end())
return;
cb(i->second);
parent_->parent().notify<hook::connection_lost>(i->second);
node_information_base_.erase(i->second);
nid_by_hdl_.erase(i->first);
// TODO: Look through other nodes and remove this one as an origin?
}
node_id routing_table::lookup_indirect(const node_id& nid) const {
auto i = indirect_.find(nid);
if (i == indirect_.end())
return none;
if (i->second.empty())
return none;
return *i->second.begin();
void routing_table::add(const node_id& nid, const connection_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};
parent_->parent().notify<hook::new_connection_established>(nid);
}
void routing_table::blacklist(const node_id& hop, const node_id& dest) {
blacklist_[dest].emplace(hop);
auto i = indirect_.find(dest);
if (i == indirect_.end())
return;
i->second.erase(hop);
if (i->second.empty())
indirect_.erase(i);
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};
// TODO: Some new related hook?
//parent_->parent().notify<hook::new_connection_established>(nid);
}
void routing_table::erase_direct(const connection_handle& hdl,
erase_callback& cb) {
auto i = direct_by_hdl_.find(hdl);
if (i == direct_by_hdl_.end())
return;
cb(i->second);
parent_->parent().notify<hook::connection_lost>(i->second);
direct_by_nid_.erase(i->second);
direct_by_hdl_.erase(i->first);
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};
// TODO: Some new related hook?
//parent_->parent().notify<hook::new_connection_established>(nid);
}
bool routing_table::erase_indirect(const node_id& dest) {
auto i = indirect_.find(dest);
if (i == indirect_.end())
return false;
if (parent_->parent().has_hook())
for (auto& nid : i->second)
parent_->parent().notify<hook::route_lost>(nid, dest);
indirect_.erase(i);
return true;
bool routing_table::reachable(const node_id& dest) {
auto i = node_information_base_.find(dest);
if (i != node_information_base_.end())
return i->second.hdl != none;
return false;
}
void routing_table::add_direct(const connection_handle& hdl,
const node_id& nid) {
CAF_ASSERT(direct_by_hdl_.count(hdl) == 0);
CAF_ASSERT(direct_by_nid_.count(nid) == 0);
direct_by_hdl_.emplace(hdl, nid);
direct_by_nid_.emplace(nid, hdl);
parent_->parent().notify<hook::new_connection_established>(nid);
bool routing_table::origin(const node_id& nid, const node_id& origin) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
i->second.origin = origin;
return true;
}
bool routing_table::add_indirect(const node_id& hop, const node_id& dest) {
auto i = blacklist_.find(dest);
if (i == blacklist_.end() || i->second.count(hop) == 0) {
auto& hops = indirect_[dest];
auto added_first = hops.empty();
hops.emplace(hop);
parent_->parent().notify<hook::new_route_added>(hop, dest);
return added_first;
}
return false; // blacklisted
optional<node_id> routing_table::origin(const node_id& nid) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return none;
return i->second.origin;
}
bool routing_table::reachable(const node_id& dest) {
return direct_by_nid_.count(dest) > 0 || indirect_.count(dest) > 0;
bool routing_table::handle(const node_id& nid, const connection_handle& hdl) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
i->second.hdl = hdl;
nid_by_hdl_.emplace(hdl, nid);
return true;
}
size_t routing_table::erase(const node_id& dest, erase_callback& cb) {
cb(dest);
size_t res = 0;
auto i = indirect_.find(dest);
if (i != indirect_.end()) {
res = i->second.size();
for (auto& nid : i->second) {
cb(nid);
parent_->parent().notify<hook::route_lost>(nid, dest);
}
indirect_.erase(i);
}
auto hdl = lookup_direct(dest);
if (hdl) {
direct_by_hdl_.erase(*hdl);
direct_by_nid_.erase(dest);
parent_->parent().notify<hook::connection_lost>(dest);
++res;
}
return res;
optional<connection_handle> routing_table::handle(const node_id& nid) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return none;
return i->second.hdl;
}
const routing_table::endpoint& routing_table::autoconnect_endpoint() {
......
This diff is collapsed.
......@@ -75,7 +75,6 @@ constexpr uint64_t default_operation_data = make_message_id().integer_value();
constexpr auto basp_atom = caf::atom("BASP");
constexpr auto spawn_serv_atom = caf::atom("SpawnServ");
constexpr auto config_serv_atom = caf::atom("ConfigServ");
} // namespace <anonymous>
......@@ -290,9 +289,9 @@ public:
make_message(sys_atom::value, get_atom::value, "info"));
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
auto path = unbox(tbl().lookup(n.id));
CAF_CHECK_EQUAL(path.hdl, n.connection);
CAF_CHECK_EQUAL(path.next_hop, n.id);
auto lr = tbl().lookup(n.id);
CAF_REQUIRE(lr.hdl);
CAF_CHECK_EQUAL(*lr.hdl, n.connection);
}
std::pair<basp::header, buffer> read_from_out_buf(connection_handle hdl) {
......@@ -568,8 +567,8 @@ CAF_TEST(remote_actor_and_send) {
mock(jupiter().connection,
{basp::message_type::server_handshake, 0, 0, basp::version,
jupiter().dummy_actor->id(), invalid_actor_id},
jupiter().id, std::string{}, jupiter().dummy_actor->id(), uint32_t{0},
basp::routing_table::endpoint{})
jupiter().id, std::string{}, basp::routing_table::endpoint{},
jupiter().dummy_actor->id(), uint32_t{0})
.receive(jupiter().connection, basp::message_type::client_handshake,
no_flags, any_vals, no_operation_data, invalid_actor_id,
invalid_actor_id, this_node(), std::string{},
......@@ -660,106 +659,10 @@ CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(basp_tests_with_autoconn, autoconn_enabled_fixture)
CAF_TEST_DISABLED(automatic_connection) {
// this tells our BASP broker to enable the automatic connection feature
//anon_send(aut(), ok_atom::value,
// "middleman.enable-automatic-connections", make_message(true));
//mpx()->exec_runnable(); // process publish message in basp_broker
// jupiter [remote hdl 0] -> mars [remote hdl 1] -> earth [this_node]
// (this node receives a message from jupiter via mars and responds via mars,
// but then also establishes a connection to jupiter directly)
auto check_node_in_tbl = [&](node& n) {
auto hdl = tbl().lookup_direct(n.id);
CAF_REQUIRE(hdl);
};
mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
CAF_MESSAGE("self: " << to_string(self()->address()));
auto ax = accept_handle::from_int(4242);
mpx()->provide_acceptor(4242, ax);
publish(self(), 4242);
mpx()->flush_runnables(); // process publish message in basp_broker
CAF_MESSAGE("connect to mars");
connect_node(mars(), ax, self()->id());
//CAF_CHECK_EQUAL(tbl().lookup_direct(mars().id).id(), mars().connection.id());
check_node_in_tbl(mars());
// TODO: this use case is no longer possible. Nodes are required to open a
// connection first, before sending messages (forwarding has been
// removed).
CAF_MESSAGE("simulate that an actor from jupiter "
"sends a message to us via mars");
mock(mars().connection,
{basp::message_type::dispatch_message, 0, 0, 0,
jupiter().dummy_actor->id(), self()->id()},
std::vector<actor_id>{}, make_message("hello from jupiter!"))
.receive(mars().connection, basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data, any_vals, invalid_actor_id,
spawn_serv_atom, std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info"))
.receive(mars().connection, basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data,
any_vals, // actor ID of an actor spawned by the BASP broker
invalid_actor_id, config_serv_atom, std::vector<actor_id>{},
make_message(get_atom::value, "basp.default-connectivity-tcp"))
.receive(mars().connection, basp::message_type::announce_proxy, no_flags,
no_payload, no_operation_data, invalid_actor_id,
jupiter().dummy_actor->id());
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), mars().id);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
auto connection_helper_actor = sys.latest_actor_id();
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
// create a dummy config server and respond to the name lookup
CAF_MESSAGE("receive ConfigServ of jupiter");
network::address_listing res;
res[network::protocol::ipv4].emplace_back("jupiter");
mock(mars().connection,
{basp::message_type::dispatch_message, 0, 0, 0, invalid_actor_id,
connection_helper_actor},
std::vector<actor_id>{},
make_message("basp.default-connectivity-tcp",
make_message(uint16_t{8080}, std::move(res))));
// our connection helper should now connect to jupiter and
// send the scribe handle over to the BASP broker
while (mpx()->has_pending_scribe("jupiter", 8080)) {
sched.run();
mpx()->flush_runnables();
}
CAF_REQUIRE(mpx()->output_buffer(mars().connection).empty());
// send handshake from jupiter
mock(jupiter().connection,
{basp::message_type::server_handshake, 0, 0, basp::version,
jupiter().dummy_actor->id(), invalid_actor_id},
std::string{}, jupiter().dummy_actor->id(), uint32_t{0})
.receive(jupiter().connection, basp::message_type::client_handshake,
no_flags, 1u, no_operation_data, invalid_actor_id,
invalid_actor_id, std::string{});
CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), none);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
check_node_in_tbl(jupiter());
check_node_in_tbl(mars());
CAF_MESSAGE("receive message from jupiter");
self()->receive(
[](const std::string& str) -> std::string {
CAF_CHECK_EQUAL(str, "hello from jupiter!");
return "hello from earth!";
}
);
mpx()->exec_runnable(); // process forwarded message in basp_broker
CAF_MESSAGE("response message must take direct route now");
mock().receive(jupiter().connection, basp::message_type::dispatch_message,
no_flags, any_vals, default_operation_data, self()->id(),
jupiter().dummy_actor->id(), std::vector<actor_id>{},
make_message("hello from earth!"));
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
}
CAF_TEST(read_address_after_handshake) {
auto check_node_in_tbl = [&](node& n) {
auto hdl = tbl().lookup_direct(n.id);
CAF_REQUIRE(hdl);
auto lr = tbl().lookup(n.id);
CAF_REQUIRE(lr.hdl);
};
mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
......
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