Unverified Commit b63f959c authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #664

Remove indirect routing
parents 1ba9006e 97272b36
...@@ -45,11 +45,12 @@ namespace basp { ...@@ -45,11 +45,12 @@ namespace basp {
/// Describes a protocol instance managing multiple connections. /// Describes a protocol instance managing multiple connections.
class instance { class instance {
public: public:
using endpoint_handle = routing_table::endpoint_handle;
/// Provides a callback-based interface for certain BASP events. /// Provides a callback-based interface for certain BASP events.
class callee { class callee {
protected: protected:
using buffer_type = std::vector<char>; using buffer_type = std::vector<char>;
using endpoint_handle = variant<connection_handle, datagram_handle>; using endpoint_handle = instance::endpoint_handle;
public: public:
explicit callee(actor_system& sys, proxy_registry::backend& backend); explicit callee(actor_system& sys, proxy_registry::backend& backend);
...@@ -83,14 +84,8 @@ public: ...@@ -83,14 +84,8 @@ public:
std::vector<strong_actor_ptr>& forwarding_stack, std::vector<strong_actor_ptr>& forwarding_stack,
message& msg) = 0; message& msg) = 0;
/// Called whenever BASP learns the ID of a remote node /// Called whenever BASP learns the ID of a remote node.
/// to which it does not have a direct connection. virtual void learned_new_node(const node_id& nid) = 0;
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 if a heartbeat was received from `nid` /// Called if a heartbeat was received from `nid`
virtual void handle_heartbeat(const node_id& nid) = 0; virtual void handle_heartbeat(const node_id& nid) = 0;
...@@ -176,15 +171,15 @@ public: ...@@ -176,15 +171,15 @@ public:
/// Sends heartbeat messages to all valid nodes those are directly connected. /// Sends heartbeat messages to all valid nodes those are directly connected.
void handle_heartbeat(execution_unit* ctx); void handle_heartbeat(execution_unit* ctx);
/// Returns a route to `target` or `none` on error. /// Returns the handle for communication with `target` or `none` on error.
optional<routing_table::route> lookup(const node_id& target); optional<endpoint_handle> lookup(const node_id& target);
/// Flushes the underlying buffer of `path`. /// Flushes the underlying buffer of `hdl`.
void flush(const routing_table::route& path); void flush(endpoint_handle hdl);
/// Sends a BASP message and implicitly flushes the output buffer of `r`. /// Sends a BASP message and implicitly flushes the output buffer of `hdl`.
/// This function will update `hdr.payload_len` if a payload was written. /// This function will update `hdr.payload_len` if a payload was written.
void write(execution_unit* ctx, const routing_table::route& r, void write(execution_unit* ctx, endpoint_handle hdl,
header& hdr, payload_writer* writer = nullptr); header& hdr, payload_writer* writer = nullptr);
/// Adds a new actor to the map of published actors. /// Adds a new actor to the map of published actors.
...@@ -294,11 +289,11 @@ public: ...@@ -294,11 +289,11 @@ public:
bool handle(execution_unit* ctx, const Handle& hdl, header& hdr, bool handle(execution_unit* ctx, const Handle& hdl, header& hdr,
std::vector<char>* payload, bool tcp_based, std::vector<char>* payload, bool tcp_based,
optional<endpoint_context&> ep, optional<uint16_t> port) { optional<endpoint_context&> ep, optional<uint16_t> port) {
// function object for checking payload validity // Function object for checking payload validity.
auto payload_valid = [&]() -> bool { auto payload_valid = [&]() -> bool {
return payload != nullptr && payload->size() == hdr.payload_len; return payload != nullptr && payload->size() == hdr.payload_len;
}; };
// handle message to ourselves // Handle message to ourselves.
switch (hdr.operation) { switch (hdr.operation) {
case message_type::server_handshake: { case message_type::server_handshake: {
actor_id aid = invalid_actor_id; actor_id aid = invalid_actor_id;
...@@ -320,37 +315,31 @@ public: ...@@ -320,37 +315,31 @@ public:
if (e) if (e)
return false; return false;
} }
// close self connection after handshake is done // Close self connection after handshake is done.
if (hdr.source_node == this_node_) { if (hdr.source_node == this_node_) {
CAF_LOG_INFO("close connection to self immediately"); CAF_LOG_INFO("close connection to self immediately");
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
return false; return false;
} }
// close this connection if we already have a direct connection // Close this connection if we already established communication.
if (tbl_.lookup_direct(hdr.source_node)) { // TODO: Should we allow multiple "connections" over different
// transport protocols?
if (tbl_.lookup(hdr.source_node)) {
CAF_LOG_INFO("close connection since we already have a " CAF_LOG_INFO("close connection since we already have a "
"direct connection: " << CAF_ARG(hdr.source_node)); "connection: " << CAF_ARG(hdr.source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
return false; return false;
} }
// add direct route to this node and remove any indirect entry // Add this node to our contacts.
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node)); CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node));
tbl_.add_direct(hdl, hdr.source_node); tbl_.add(hdl, hdr.source_node);
auto was_indirect = tbl_.erase_indirect(hdr.source_node); // TODO: Add status, addresses, ... ?
// write handshake as client in response // Write handshake as client in response.
auto path = tbl_.lookup(hdr.source_node); if (tcp_based)
if (!path) { write_client_handshake(ctx, callee_.get_buffer(hdl), hdr.source_node);
CAF_LOG_ERROR("no route to host after server handshake"); callee_.learned_new_node(hdr.source_node);
return false;
}
if (tcp_based) {
auto ch = get<connection_handle>(path->hdl);
write_client_handshake(ctx, callee_.get_buffer(ch),
hdr.source_node);
}
callee_.learned_new_node_directly(hdr.source_node, was_indirect);
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path); flush(hdl);
break; break;
} }
case message_type::client_handshake: { case message_type::client_handshake: {
...@@ -368,49 +357,35 @@ public: ...@@ -368,49 +357,35 @@ public:
return false; return false;
} }
} }
if (tcp_based) { auto new_node = (this_node() != hdr.source_node
if (tbl_.lookup_direct(hdr.source_node)) { && !tbl_.lookup(hdr.source_node));
if (!new_node) {
if (tcp_based) {
CAF_LOG_INFO("received second client handshake:" CAF_LOG_INFO("received second client handshake:"
<< CAF_ARG(hdr.source_node)); << CAF_ARG(hdr.source_node));
break; break;
} }
// add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
tbl_.add_direct(hdl, hdr.source_node);
auto was_indirect = tbl_.erase_indirect(hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node, was_indirect);
} else { } else {
auto new_node = (this_node() != hdr.source_node // Add this node to our contacts.
&& !tbl_.lookup_direct(hdr.source_node)); CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node));
if (new_node) { tbl_.add(hdl, hdr.source_node);
// add direct route to this node and remove any indirect entry // TODO: Add status, addresses, ... ?
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node)); }
tbl_.add_direct(hdl, hdr.source_node); // Since udp is unreliable we answer, maybe our message was lost.
} if (!tcp_based) {
uint16_t seq = (ep && ep->requires_ordering) ? ep->seq_outgoing++ : 0; uint16_t seq = (ep && ep->requires_ordering) ? ep->seq_outgoing++ : 0;
write_server_handshake(ctx, write_server_handshake(ctx, callee_.get_buffer(hdl), port, seq);
callee_.get_buffer(hdl),
port, seq);
callee_.flush(hdl); callee_.flush(hdl);
if (new_node) {
auto was_indirect = tbl_.erase_indirect(hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node, was_indirect);
}
} }
// We have to call this after `write_server_handshake` because
// `learned_new_node` expects there to be an entry in the routing table.
if (new_node)
callee_.learned_new_node(hdr.source_node);
break; break;
} }
case message_type::dispatch_message: { case message_type::dispatch_message: {
if (!payload_valid()) if (!payload_valid())
return false; return false;
// in case the sender of this message was received via a third node,
// we assume that that node to offers a route to the original source
auto last_hop = tbl_.lookup_direct(hdl);
if (hdr.source_node != none
&& hdr.source_node != this_node_
&& last_hop != hdr.source_node
&& !tbl_.lookup_direct(hdr.source_node)
&& tbl_.add_indirect(last_hop, hdr.source_node))
callee_.learned_new_node_indirectly(hdr.source_node);
binary_deserializer bd{ctx, *payload}; binary_deserializer bd{ctx, *payload};
auto receiver_name = static_cast<atom_value>(0); auto receiver_name = static_cast<atom_value>(0);
std::vector<strong_actor_ptr> forwarding_stack; std::vector<strong_actor_ptr> forwarding_stack;
......
...@@ -30,6 +30,8 @@ ...@@ -30,6 +30,8 @@
#include "caf/io/basp/buffer_type.hpp" #include "caf/io/basp/buffer_type.hpp"
#include "caf/io/network/interfaces.hpp"
namespace std { namespace std {
template<> template<>
...@@ -58,59 +60,30 @@ public: ...@@ -58,59 +60,30 @@ public:
virtual ~routing_table(); virtual ~routing_table();
/// Describes a routing path to a node.
struct route {
const node_id& next_hop;
endpoint_handle hdl;
};
/// Describes a function object for erase operations that /// Describes a function object for erase operations that
/// is called for each indirectly lost connection. /// is called for each indirectly lost connection.
using erase_callback = callback<const node_id&>; using erase_callback = callback<const node_id&>;
/// Returns a route to `target` or `none` on error. /// Returns the ID of the peer reachable via `hdl` or
optional<route> lookup(const node_id& target);
/// Returns the ID of the peer connected via `hdl` or
/// `none` if `hdl` is unknown. /// `none` if `hdl` is unknown.
node_id lookup_direct(const endpoint_handle& hdl) const; node_id lookup(const endpoint_handle& hdl) const;
/// Returns the handle offering a direct connection to `nid` or /// Returns the handle for communication with `nid` or
/// `invalid_connection_handle` if no direct connection to `nid` exists. /// `none` if `nid` is unknown.
optional<endpoint_handle> lookup_direct(const node_id& nid) const; optional<endpoint_handle> lookup(const node_id& nid) 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;
/// Adds a new direct route to the table. /// Adds a new direct route to the table.
/// @pre `hdl != invalid_connection_handle && nid != none` /// @pre `hdl != invalid_connection_handle && nid != none`
void add_direct(const endpoint_handle& hdl, const node_id& nid); void add(const endpoint_handle& hdl, const node_id& nid);
/// Adds a new indirect route to the table.
bool add_indirect(const node_id& hop, const node_id& dest);
/// Blacklist the route to `dest` via `hop`.
void blacklist(const node_id& hop, const node_id& dest);
/// Removes a direct connection and calls `cb` for any node /// Removes a direct connection and calls `cb` for any node
/// that became unreachable as a result of this operation, /// that became unreachable as a result of this operation,
/// including the node that is assigned as direct path for `hdl`. /// including the node that is assigned as direct path for `hdl`.
void erase_direct(const endpoint_handle& hdl, erase_callback& cb); void erase(const endpoint_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.
bool reachable(const node_id& dest); 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. /// Returns the parent broker.
inline abstract_broker* parent() { inline abstract_broker* parent() {
return parent_; return parent_;
...@@ -126,16 +99,11 @@ public: ...@@ -126,16 +99,11 @@ public:
return std::forward<Fallback>(x); 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_; abstract_broker* parent_;
std::unordered_map<endpoint_handle, node_id> direct_by_hdl_; std::unordered_map<endpoint_handle, node_id> direct_by_hdl_;
// TODO: Do we need a list as a second argument as there could be
// multiple handles for different technologies?
std::unordered_map<node_id, endpoint_handle> direct_by_nid_; std::unordered_map<node_id, endpoint_handle> direct_by_nid_;
indirect_entries indirect_;
indirect_entries blacklist_;
}; };
/// @} /// @}
......
...@@ -81,14 +81,7 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -81,14 +81,7 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
std::vector<strong_actor_ptr>& stages, message& msg); std::vector<strong_actor_ptr>& stages, message& msg);
// performs bookkeeping such as managing `spawn_servers` // 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;
// inherited from basp::instance::callee
void learned_new_node_indirectly(const node_id& nid) override;
// inherited from basp::instance::callee // inherited from basp::instance::callee
uint16_t next_sequence_number(connection_handle hdl) override; uint16_t next_sequence_number(connection_handle hdl) override;
......
...@@ -43,7 +43,7 @@ namespace io { ...@@ -43,7 +43,7 @@ namespace io {
namespace { namespace {
// visitors to access handle variant of the context // Visitors to access handle variant of the context.
struct seq_num_visitor { struct seq_num_visitor {
using result_type = basp::sequence_type; using result_type = basp::sequence_type;
seq_num_visitor(basp_broker_state* ptr) : state(ptr) { } seq_num_visitor(basp_broker_state* ptr) : state(ptr) { }
...@@ -83,7 +83,7 @@ basp_broker_state::basp_broker_state(broker* selfptr) ...@@ -83,7 +83,7 @@ basp_broker_state::basp_broker_state(broker* selfptr)
} }
basp_broker_state::~basp_broker_state() { basp_broker_state::~basp_broker_state() {
// make sure all spawn servers are down // Make sure all spawn servers are down.
for (auto& kvp : spawn_servers) for (auto& kvp : spawn_servers)
anon_send_exit(kvp.second, exit_reason::kill); anon_send_exit(kvp.second, exit_reason::kill);
} }
...@@ -93,25 +93,27 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) { ...@@ -93,25 +93,27 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
CAF_ASSERT(nid != this_node()); CAF_ASSERT(nid != this_node());
if (nid == none || aid == invalid_actor_id) if (nid == none || aid == invalid_actor_id)
return nullptr; return nullptr;
// this member function is being called whenever we deserialize a // This member function is being called whenever we deserialize a
// payload received from a remote node; if a remote node A sends // payload received from a remote node; if a remote node A sends
// us a handle to a third node B, then we assume that A offers a route to B // us a handle to a third node B, then we assume that A offers a route to B.
if (nid != this_context->id if (nid != this_context->id && !instance.tbl().lookup(nid))
&& !instance.tbl().lookup_direct(nid) // TODO: Try to establish communication with the new node.
&& instance.tbl().add_indirect(this_context->id, nid)) CAF_CRITICAL("Not implemented.");
learned_new_node_indirectly(nid); // TODO: Everything below has to happen once we establish communication?
// we need to tell remote side we are watching this actor now; // I'm not sure yet, the functors can be attached earlier and we could
// use a direct route if possible, i.e., when talking to a third node // send trigger an error message if we cannot contact the remote node.
auto path = instance.tbl().lookup(nid); // We need to tell remote side we are watching this actor now;
if (!path) { // use a direct route if possible, i.e., when talking to a third node.
// this happens if and only if we don't have a path to `nid` auto ehdl = instance.tbl().lookup(nid);
// and current_context_->hdl has been blacklisted if (!ehdl) {
// This happens if and only if we don't have a path to `nid`
// and current_context_->hdl has been blacklisted.
CAF_LOG_INFO("cannot create a proxy instance for an actor " CAF_LOG_INFO("cannot create a proxy instance for an actor "
"running on a node we don't have a route to"); "running on a node we don't have a route to");
return nullptr; return nullptr;
} }
// create proxy and add functor that will be called if we // Create proxy and add functor that will be called if we
// receive a kill_proxy_instance message // receive a kill_proxy_instance message.
auto mm = &system().middleman(); auto mm = &system().middleman();
actor_config cfg; actor_config cfg;
auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>( auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>(
...@@ -119,9 +121,9 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) { ...@@ -119,9 +121,9 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
strong_actor_ptr selfptr{self->ctrl()}; strong_actor_ptr selfptr{self->ctrl()};
res->get()->attach_functor([=](const error& rsn) { res->get()->attach_functor([=](const error& rsn) {
mm->backend().post([=] { mm->backend().post([=] {
// using res->id() instead of aid keeps this actor instance alive // Using res->id() instead of aid keeps this actor instance alive
// until the original instance terminates, thus preventing subtle // until the original instance terminates, thus preventing subtle
// bugs with attachables // bugs with attachables.
auto bptr = static_cast<basp_broker*>(selfptr->get()); auto bptr = static_cast<basp_broker*>(selfptr->get());
if (!bptr->getf(abstract_actor::is_terminated_flag)) if (!bptr->getf(abstract_actor::is_terminated_flag))
bptr->state.proxies().erase(nid, res->id(), rsn); bptr->state.proxies().erase(nid, res->id(), rsn);
...@@ -131,12 +133,12 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) { ...@@ -131,12 +133,12 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
"write announce_proxy_instance:" "write announce_proxy_instance:"
<< CAF_ARG(nid) << CAF_ARG(aid)); << CAF_ARG(nid) << CAF_ARG(aid));
auto& ctx = *this_context; auto& ctx = *this_context;
// tell remote side we are monitoring this actor now // Tell remote side we are monitoring this actor now.
instance.write_announce_proxy(self->context(), instance.write_announce_proxy(self->context(),
get_buffer(this_context->hdl), get_buffer(this_context->hdl),
nid, aid, nid, aid,
ctx.requires_ordering ? ctx.seq_outgoing++ : 0); ctx.requires_ordering ? ctx.seq_outgoing++ : 0);
instance.flush(*path); instance.flush(*ehdl);
mm->notify<hook::new_remote_actor>(res); mm->notify<hook::new_remote_actor>(res);
return res; return res;
} }
...@@ -154,10 +156,10 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid, ...@@ -154,10 +156,10 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
if (cb == none) if (cb == none)
return; return;
strong_actor_ptr ptr; strong_actor_ptr ptr;
// aid can be invalid when connecting to the default port of a node // The aid can be invalid when connecting to the default port of a node.
if (aid != invalid_actor_id) { if (aid != invalid_actor_id) {
if (nid == this_node()) { if (nid == this_node()) {
// connected to self // Connected to self.
ptr = actor_cast<strong_actor_ptr>(system().registry().get(aid)); ptr = actor_cast<strong_actor_ptr>(system().registry().get(aid));
CAF_LOG_INFO_IF(!ptr, "actor not found:" << CAF_ARG(aid)); CAF_LOG_INFO_IF(!ptr, "actor not found:" << CAF_ARG(aid));
} else { } else {
...@@ -183,26 +185,27 @@ void basp_broker_state::send_kill_proxy_instance(const node_id& nid, ...@@ -183,26 +185,27 @@ void basp_broker_state::send_kill_proxy_instance(const node_id& nid,
CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(rsn)); CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(rsn));
if (rsn == none) if (rsn == none)
rsn = exit_reason::unknown; rsn = exit_reason::unknown;
auto path = instance.tbl().lookup(nid); auto ehdl = instance.tbl().lookup(nid);
if (!path) { if (!ehdl) {
CAF_LOG_INFO("cannot send exit message for proxy, no route to host:" CAF_LOG_INFO("cannot send exit message for proxy, no route to host:"
<< CAF_ARG(nid)); << CAF_ARG(nid));
return; return;
} }
auto hdl = std::move(*ehdl);
instance.write_kill_proxy(self->context(), instance.write_kill_proxy(self->context(),
get_buffer(path->hdl), get_buffer(hdl),
nid, aid, rsn, nid, aid, rsn,
visit(seq_num_visitor{this}, path->hdl)); visit(seq_num_visitor{this}, hdl));
instance.flush(*path); instance.flush(hdl);
} }
void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) { void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid)); CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid));
// source node has created a proxy for one of our actors // Source node has created a proxy for one of our actors.
auto ptr = system().registry().get(aid); auto ptr = system().registry().get(aid);
if (ptr == nullptr) { if (ptr == nullptr) {
CAF_LOG_DEBUG("kill proxy immediately"); CAF_LOG_DEBUG("kill proxy immediately");
// kill immediately if actor has already terminated // Kill immediately if actor has already terminated.
send_kill_proxy_instance(nid, aid, exit_reason::unknown); send_kill_proxy_instance(nid, aid, exit_reason::unknown);
} else { } else {
auto entry = ptr->address(); auto entry = ptr->address();
...@@ -324,12 +327,12 @@ void basp_broker_state::learned_new_node(const node_id& nid) { ...@@ -324,12 +327,12 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
} }
auto tmp = system().spawn<hidden>([=](event_based_actor* tself) -> behavior { auto tmp = system().spawn<hidden>([=](event_based_actor* tself) -> behavior {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// terminate when receiving a down message // Terminate when receiving a down message.
tself->set_down_handler([=](down_msg& dm) { tself->set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm)); CAF_LOG_TRACE(CAF_ARG(dm));
tself->quit(std::move(dm.reason)); tself->quit(std::move(dm.reason));
}); });
// skip messages until we receive the initial ok_atom // Skip messages until we receive the initial ok_atom.
tself->set_default_handler(skip); tself->set_default_handler(skip);
return { return {
[=](ok_atom, const std::string& /* key == "info" */, [=](ok_atom, const std::string& /* key == "info" */,
...@@ -366,74 +369,22 @@ void basp_broker_state::learned_new_node(const node_id& nid) { ...@@ -366,74 +369,22 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
auto msg = make_message(sys_atom::value, get_atom::value, "info"); auto msg = make_message(sys_atom::value, get_atom::value, "info");
return sink(name_atm, stages, msg); return sink(name_atm, stages, msg);
}); });
auto path = instance.tbl().lookup(nid); auto ehdl = instance.tbl().lookup(nid);
if (!path) { if (!ehdl) {
CAF_LOG_ERROR("learned_new_node called, but no route to nid"); CAF_LOG_ERROR("learned_new_node called, but no route to nid");
return; return;
} }
auto hdl = std::move(*ehdl);
// send message to SpawnServ of remote node // send message to SpawnServ of remote node
basp::header hdr{basp::message_type::dispatch_message, basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag, basp::header::named_receiver_flag,
0, 0, this_node(), nid, tmp.id(), invalid_actor_id, 0, 0, this_node(), nid, tmp.id(), invalid_actor_id,
visit(seq_num_visitor{this}, path->hdl)}; visit(seq_num_visitor{this}, hdl)};
// writing std::numeric_limits<actor_id>::max() is a hack to get // writing std::numeric_limits<actor_id>::max() is a hack to get
// this send-to-named-actor feature working with older CAF releases // this send-to-named-actor feature working with older CAF releases
instance.write(self->context(), get_buffer(path->hdl), instance.write(self->context(), get_buffer(hdl),
hdr, &writer); hdr, &writer);
instance.flush(*path); instance.flush(hdl);
}
void basp_broker_state::learned_new_node_directly(const node_id& nid,
bool was_indirectly_before) {
CAF_ASSERT(this_context != nullptr);
CAF_LOG_TRACE(CAF_ARG(nid));
if (!was_indirectly_before)
learned_new_node(nid);
}
void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
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
auto path = instance.tbl().lookup(nid);
if (!path) {
CAF_LOG_ERROR("learned_new_node_indirectly called, but no route to nid");
return;
}
if (path->next_hop == nid) {
CAF_LOG_ERROR("learned_new_node_indirectly called with direct connection");
return;
}
using namespace detail;
auto try_connect = [&](std::string item) {
auto tmp = system().config().middleman_detach_utility_actors
? system().spawn<detached + hidden>(connection_helper, self)
: system().spawn<hidden>(connection_helper, self);
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}, path->hdl)};
instance.write(self->context(), get_buffer(path->hdl),
hdr, &writer);
instance.flush(*path);
};
if (enable_tcp)
try_connect("basp.default-connectivity-tcp");
if (enable_udp)
try_connect("basp.default-connectivity-udp");
} }
void basp_broker_state::set_context(connection_handle hdl) { void basp_broker_state::set_context(connection_handle hdl) {
...@@ -484,7 +435,7 @@ void basp_broker_state::cleanup(connection_handle hdl) { ...@@ -484,7 +435,7 @@ void basp_broker_state::cleanup(connection_handle hdl) {
purge_state(nid); purge_state(nid);
return none; return none;
}); });
instance.tbl().erase_direct(hdl, cb); instance.tbl().erase(hdl, cb);
// Remove the context for `hdl`, making sure clients receive an error in case // Remove the context for `hdl`, making sure clients receive an error in case
// this connection was closed during handshake. // this connection was closed during handshake.
auto i = ctx_tcp.find(hdl); auto i = ctx_tcp.find(hdl);
...@@ -507,7 +458,7 @@ void basp_broker_state::cleanup(datagram_handle hdl) { ...@@ -507,7 +458,7 @@ void basp_broker_state::cleanup(datagram_handle hdl) {
purge_state(nid); purge_state(nid);
return none; return none;
}); });
instance.tbl().erase_direct(hdl, cb); instance.tbl().erase(hdl, cb);
// Remove the context for `hdl`, making sure clients receive an error in case // Remove the context for `hdl`, making sure clients receive an error in case
// this connection was closed during handshake. // this connection was closed during handshake.
auto i = ctx_udp.find(hdl); auto i = ctx_udp.find(hdl);
...@@ -644,8 +595,8 @@ behavior basp_broker::make_behavior() { ...@@ -644,8 +595,8 @@ behavior basp_broker::make_behavior() {
state.enable_udp = system().config().middleman_enable_udp; state.enable_udp = system().config().middleman_enable_udp;
if (system().config().middleman_enable_automatic_connections) { if (system().config().middleman_enable_automatic_connections) {
CAF_LOG_INFO("enable automatic connections"); CAF_LOG_INFO("enable automatic connections");
// open a random port and store a record for our peers how to // Open a random port and store a record for our peers how to
// connect to this broker directly in the configuration server // connect to this broker directly in the configuration server.
if (state.enable_tcp) { if (state.enable_tcp) {
auto res = add_tcp_doorman(uint16_t{0}); auto res = add_tcp_doorman(uint16_t{0});
if (res) { if (res) {
...@@ -676,7 +627,7 @@ behavior basp_broker::make_behavior() { ...@@ -676,7 +627,7 @@ behavior basp_broker::make_behavior() {
send(this, tick_atom::value, heartbeat_interval); send(this, tick_atom::value, heartbeat_interval);
} }
return { return {
// received from underlying broker implementation // Received from underlying broker implementation.
[=](new_data_msg& msg) { [=](new_data_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
state.set_context(msg.handle); state.set_context(msg.handle);
...@@ -696,7 +647,7 @@ behavior basp_broker::make_behavior() { ...@@ -696,7 +647,7 @@ behavior basp_broker::make_behavior() {
ctx.cstate = next; ctx.cstate = next;
} }
}, },
// received from auto connect broker for UDP communication // Received from auto connect broker for UDP communication.
[=](new_datagram_msg& msg, datagram_servant_ptr ptr, uint16_t port) { [=](new_datagram_msg& msg, datagram_servant_ptr ptr, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
auto hdl = ptr->hdl(); auto hdl = ptr->hdl();
...@@ -707,11 +658,11 @@ behavior basp_broker::make_behavior() { ...@@ -707,11 +658,11 @@ behavior basp_broker::make_behavior() {
ctx.local_port = local_port(hdl); ctx.local_port = local_port(hdl);
ctx.requires_ordering = true; ctx.requires_ordering = true;
ctx.seq_incoming = 0; ctx.seq_incoming = 0;
ctx.seq_outgoing = 1; // already sent the client handshake ctx.seq_outgoing = 1; // Already sent the client handshake.
// Let's not implement this twice // Let's not implement this twice.
send(this, std::move(msg)); send(this, std::move(msg));
}, },
// received from underlying broker implementation // Received from underlying broker implementation.
[=](new_datagram_msg& msg) { [=](new_datagram_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
state.set_context(msg.handle); state.set_context(msg.handle);
...@@ -728,12 +679,12 @@ behavior basp_broker::make_behavior() { ...@@ -728,12 +679,12 @@ behavior basp_broker::make_behavior() {
close(msg.handle); close(msg.handle);
} }
}, },
// received from the underlying broker implementation // Received from the underlying broker implementation.
[=](datagram_sent_msg& msg) { [=](datagram_sent_msg& msg) {
if (state.cached_buffers.size() < state.max_buffers) if (state.cached_buffers.size() < state.max_buffers)
state.cached_buffers.emplace(std::move(msg.buf)); state.cached_buffers.emplace(std::move(msg.buf));
}, },
// received from proxy instances // Received from proxy instances.
[=](forward_atom, strong_actor_ptr& src, [=](forward_atom, strong_actor_ptr& src,
const std::vector<strong_actor_ptr>& fwd_stack, const std::vector<strong_actor_ptr>& fwd_stack,
strong_actor_ptr& dest, message_id mid, const message& msg) { strong_actor_ptr& dest, message_id mid, const message& msg) {
...@@ -753,7 +704,7 @@ behavior basp_broker::make_behavior() { ...@@ -753,7 +704,7 @@ behavior basp_broker::make_behavior() {
srb(src, mid); srb(src, mid);
} }
}, },
// received from some system calls like whereis // Received from some system calls like whereis.
[=](forward_atom, const node_id& dest_node, atom_value dest_name, [=](forward_atom, const node_id& dest_node, atom_value dest_name,
const message& msg) -> result<message> { const message& msg) -> result<message> {
auto cme = current_mailbox_element(); auto cme = current_mailbox_element();
...@@ -766,11 +717,12 @@ behavior basp_broker::make_behavior() { ...@@ -766,11 +717,12 @@ behavior basp_broker::make_behavior() {
<< ", " << CAF_ARG(msg)); << ", " << CAF_ARG(msg));
if (!src) if (!src)
return sec::invalid_argument; return sec::invalid_argument;
auto path = this->state.instance.tbl().lookup(dest_node); auto ehdl = this->state.instance.tbl().lookup(dest_node);
if (!path) { if (!ehdl) {
CAF_LOG_ERROR("no route to receiving node"); CAF_LOG_ERROR("receiving node unknown");
return sec::no_route_to_receiving_node; return sec::no_route_to_receiving_node;
} }
auto hdl = std::move(*ehdl);
if (system().node() == src->node()) if (system().node() == src->node())
system().registry().put(src->id(), src); system().registry().put(src->id(), src);
auto writer = make_callback([&](serializer& sink) -> error { auto writer = make_callback([&](serializer& sink) -> error {
...@@ -780,13 +732,13 @@ behavior basp_broker::make_behavior() { ...@@ -780,13 +732,13 @@ behavior basp_broker::make_behavior() {
basp::header::named_receiver_flag, basp::header::named_receiver_flag,
0, cme->mid.integer_value(), state.this_node(), 0, cme->mid.integer_value(), state.this_node(),
dest_node, src->id(), invalid_actor_id, dest_node, src->id(), invalid_actor_id,
visit(seq_num_visitor{&state}, path->hdl)}; visit(seq_num_visitor{&state}, hdl)};
state.instance.write(context(), state.get_buffer(path->hdl), state.instance.write(context(), state.get_buffer(hdl),
hdr, &writer); hdr, &writer);
state.instance.flush(*path); state.instance.flush(hdl);
return delegated<message>(); return delegated<message>();
}, },
// received from underlying broker implementation // Received from underlying broker implementation.
[=](const new_connection_msg& msg) { [=](const new_connection_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
auto& bi = state.instance; auto& bi = state.instance;
...@@ -795,18 +747,18 @@ behavior basp_broker::make_behavior() { ...@@ -795,18 +747,18 @@ behavior basp_broker::make_behavior() {
state.flush(msg.handle); state.flush(msg.handle);
configure_read(msg.handle, receive_policy::exactly(basp::header_size)); configure_read(msg.handle, receive_policy::exactly(basp::header_size));
}, },
// received from underlying broker implementation // Received from underlying broker implementation.
[=](const connection_closed_msg& msg) { [=](const connection_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
state.cleanup(msg.handle); state.cleanup(msg.handle);
}, },
// received from underlying broker implementation // Received from underlying broker implementation.
[=](const acceptor_closed_msg& msg) { [=](const acceptor_closed_msg& msg) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto port = local_port(msg.handle); auto port = local_port(msg.handle);
state.instance.remove_published_actor(port); state.instance.remove_published_actor(port);
}, },
// received from middleman actor // Received from middleman actor.
[=](publish_atom, doorman_ptr& ptr, uint16_t port, [=](publish_atom, doorman_ptr& ptr, uint16_t port,
const strong_actor_ptr& whom, std::set<std::string>& sigs) { const strong_actor_ptr& whom, std::set<std::string>& sigs) {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(port) CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(port)
...@@ -817,7 +769,7 @@ behavior basp_broker::make_behavior() { ...@@ -817,7 +769,7 @@ behavior basp_broker::make_behavior() {
system().registry().put(whom->id(), whom); system().registry().put(whom->id(), whom);
state.instance.add_published_actor(port, whom, std::move(sigs)); state.instance.add_published_actor(port, whom, std::move(sigs));
}, },
// received from middleman actor (delegated) // Received from middleman actor (delegated).
[=](connect_atom, scribe_ptr& ptr, uint16_t port) { [=](connect_atom, scribe_ptr& ptr, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(port));
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
...@@ -830,7 +782,7 @@ behavior basp_broker::make_behavior() { ...@@ -830,7 +782,7 @@ behavior basp_broker::make_behavior() {
ctx.cstate = basp::await_header; ctx.cstate = basp::await_header;
ctx.callback = rp; ctx.callback = rp;
ctx.requires_ordering = false; ctx.requires_ordering = false;
// await server handshake // Await server handshake.
configure_read(hdl, receive_policy::exactly(basp::header_size)); configure_read(hdl, receive_policy::exactly(basp::header_size));
}, },
[=](publish_udp_atom, datagram_servant_ptr& ptr, uint16_t port, [=](publish_udp_atom, datagram_servant_ptr& ptr, uint16_t port,
...@@ -843,7 +795,7 @@ behavior basp_broker::make_behavior() { ...@@ -843,7 +795,7 @@ behavior basp_broker::make_behavior() {
system().registry().put(whom->id(), whom); system().registry().put(whom->id(), whom);
state.instance.add_published_actor(port, whom, std::move(sigs)); state.instance.add_published_actor(port, whom, std::move(sigs));
}, },
// received from middleman actor (delegated) // Received from middleman actor (delegated).
[=](contact_atom, datagram_servant_ptr& ptr, uint16_t port) { [=](contact_atom, datagram_servant_ptr& ptr, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(port));
auto rp = make_response_promise(); auto rp = make_response_promise();
...@@ -862,11 +814,11 @@ behavior basp_broker::make_behavior() { ...@@ -862,11 +814,11 @@ behavior basp_broker::make_behavior() {
none, ctx.seq_outgoing++); none, ctx.seq_outgoing++);
state.flush(hdl); state.flush(hdl);
}, },
// received from underlying broker implementation // Received from underlying broker implementation.
[=](const datagram_servant_closed_msg& msg) { [=](const datagram_servant_closed_msg& msg) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// since all handles share a port, we can take any of them to query for // Since all handles share a port, we can take any of them to query for
// port information // port information.
CAF_ASSERT(msg.handles.size() > 0); CAF_ASSERT(msg.handles.size() > 0);
auto port = local_port(msg.handles.front()); auto port = local_port(msg.handles.front());
state.instance.remove_published_actor(port); state.instance.remove_published_actor(port);
...@@ -902,8 +854,8 @@ behavior basp_broker::make_behavior() { ...@@ -902,8 +854,8 @@ behavior basp_broker::make_behavior() {
[=](close_atom, uint16_t port) -> result<void> { [=](close_atom, uint16_t port) -> result<void> {
if (port == 0) if (port == 0)
return sec::cannot_close_invalid_port; return sec::cannot_close_invalid_port;
// it is well-defined behavior to not have an actor published here, // It is well-defined behavior to not have an actor published here,
// hence the result can be ignored safely // hence the result can be ignored safely.
state.instance.remove_published_actor(port, nullptr); state.instance.remove_published_actor(port, nullptr);
auto res = close(hdl_by_port(port)); auto res = close(hdl_by_port(port));
if (res) if (res)
...@@ -914,10 +866,10 @@ behavior basp_broker::make_behavior() { ...@@ -914,10 +866,10 @@ behavior basp_broker::make_behavior() {
-> std::tuple<node_id, std::string, uint16_t> { -> std::tuple<node_id, std::string, uint16_t> {
std::string addr; std::string addr;
uint16_t port = 0; uint16_t port = 0;
auto hdl = state.instance.tbl().lookup_direct(x); auto ehdl = state.instance.tbl().lookup(x);
if (hdl) { if (ehdl) {
addr = visit(addr_visitor{this}, *hdl); addr = visit(addr_visitor{this}, *ehdl);
port = visit(port_visitor{this}, *hdl); port = visit(port_visitor{this}, *ehdl);
} }
return std::make_tuple(x, std::move(addr), port); return std::make_tuple(x, std::move(addr), port);
}, },
......
...@@ -63,13 +63,13 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -63,13 +63,13 @@ connection_state instance::handle(execution_unit* ctx,
new_data_msg& dm, header& hdr, new_data_msg& dm, header& hdr,
bool is_payload) { bool is_payload) {
CAF_LOG_TRACE(CAF_ARG(dm) << CAF_ARG(is_payload)); CAF_LOG_TRACE(CAF_ARG(dm) << CAF_ARG(is_payload));
// function object providing cleanup code on errors // Function object providing cleanup code on errors.
auto err = [&]() -> connection_state { auto err = [&]() -> connection_state {
auto cb = make_callback([&](const node_id& nid) -> error { auto cb = make_callback([&](const node_id& nid) -> error {
callee_.purge_state(nid); callee_.purge_state(nid);
return none; return none;
}); });
tbl_.erase_direct(dm.handle, cb); tbl_.erase(dm.handle, cb);
return close_connection; return close_connection;
}; };
std::vector<char>* payload = nullptr; std::vector<char>* payload = nullptr;
...@@ -93,23 +93,23 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -93,23 +93,23 @@ connection_state instance::handle(execution_unit* ctx,
} }
} }
CAF_LOG_DEBUG(CAF_ARG(hdr)); CAF_LOG_DEBUG(CAF_ARG(hdr));
// needs forwarding? // Needs forwarding?
if (!is_handshake(hdr) && !is_heartbeat(hdr) && hdr.dest_node != this_node_) { if (!is_handshake(hdr) && !is_heartbeat(hdr) && hdr.dest_node != this_node_) {
CAF_LOG_DEBUG("forward message"); CAF_LOG_DEBUG("forward message");
auto path = lookup(hdr.dest_node); auto ehdl = lookup(hdr.dest_node);
if (path) { if (ehdl) {
binary_serializer bs{ctx, callee_.get_buffer(path->hdl)}; binary_serializer bs{ctx, callee_.get_buffer(*ehdl)};
auto e = bs(hdr); auto e = bs(hdr);
if (e) if (e)
return err(); return err();
if (payload != nullptr) if (payload != nullptr)
bs.apply_raw(payload->size(), payload->data()); bs.apply_raw(payload->size(), payload->data());
flush(*path); flush(*ehdl);
notify<hook::message_forwarded>(hdr, payload); notify<hook::message_forwarded>(hdr, payload);
} else { } else {
CAF_LOG_INFO("cannot forward message, no route to destination"); CAF_LOG_INFO("cannot forward message, no route to destination");
if (hdr.source_node != this_node_) { if (hdr.source_node != this_node_) {
// TODO: signalize error back to sending node // TODO: Signalize error back to sending node.
auto reverse_path = lookup(hdr.source_node); auto reverse_path = lookup(hdr.source_node);
if (!reverse_path) { if (!reverse_path) {
CAF_LOG_WARNING("cannot send error message: no route to source"); CAF_LOG_WARNING("cannot send error message: no route to source");
...@@ -131,22 +131,22 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -131,22 +131,22 @@ connection_state instance::handle(execution_unit* ctx,
bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
endpoint_context& ep) { endpoint_context& ep) {
using itr_t = network::receive_buffer::iterator; using itr_t = network::receive_buffer::iterator;
// function object providing cleanup code on errors // Function object providing cleanup code on errors.
auto err = [&]() -> bool { auto err = [&]() -> bool {
auto cb = make_callback([&](const node_id& nid) -> error { auto cb = make_callback([&](const node_id& nid) -> error {
callee_.purge_state(nid); callee_.purge_state(nid);
return none; return none;
}); });
tbl_.erase_direct(dm.handle, cb); tbl_.erase(dm.handle, cb);
return false; return false;
}; };
// extract payload // Extract payload.
std::vector<char> pl_buf{std::move_iterator<itr_t>(std::begin(dm.buf) + std::vector<char> pl_buf{std::move_iterator<itr_t>(std::begin(dm.buf) +
basp::header_size), basp::header_size),
std::move_iterator<itr_t>(std::end(dm.buf))}; std::move_iterator<itr_t>(std::end(dm.buf))};
// resize header // Resize header.
dm.buf.resize(basp::header_size); dm.buf.resize(basp::header_size);
// extract header // Extract header.
binary_deserializer bd{ctx, dm.buf}; binary_deserializer bd{ctx, dm.buf};
auto e = bd(ep.hdr); auto e = bd(ep.hdr);
if (e || !valid(ep.hdr)) { if (e || !valid(ep.hdr)) {
...@@ -175,24 +175,24 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, ...@@ -175,24 +175,24 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
} }
// This is the expected message. // This is the expected message.
ep.seq_incoming += 1; ep.seq_incoming += 1;
// TODO: add optional reliability here // TODO: Add optional reliability here.
if (!is_handshake(ep.hdr) && !is_heartbeat(ep.hdr) if (!is_handshake(ep.hdr) && !is_heartbeat(ep.hdr)
&& ep.hdr.dest_node != this_node_) { && ep.hdr.dest_node != this_node_) {
CAF_LOG_DEBUG("forward message"); CAF_LOG_DEBUG("forward message");
auto path = lookup(ep.hdr.dest_node); auto ehdl = lookup(ep.hdr.dest_node);
if (path) { if (ehdl) {
binary_serializer bs{ctx, callee_.get_buffer(path->hdl)}; binary_serializer bs{ctx, callee_.get_buffer(*ehdl)};
auto ex = bs(ep.hdr); auto ex = bs(ep.hdr);
if (ex) if (ex)
return err(); return err();
if (payload != nullptr) if (payload != nullptr)
bs.apply_raw(payload->size(), payload->data()); bs.apply_raw(payload->size(), payload->data());
flush(*path); flush(*ehdl);
notify<hook::message_forwarded>(ep.hdr, payload); notify<hook::message_forwarded>(ep.hdr, payload);
} else { } else {
CAF_LOG_INFO("cannot forward message, no route to destination"); CAF_LOG_INFO("cannot forward message, no route to destination");
if (ep.hdr.source_node != this_node_) { if (ep.hdr.source_node != this_node_) {
// TODO: signalize error back to sending node // TODO: Signalize error back to sending node.
auto reverse_path = lookup(ep.hdr.source_node); auto reverse_path = lookup(ep.hdr.source_node);
if (!reverse_path) { if (!reverse_path) {
CAF_LOG_WARNING("cannot send error message: no route to source"); CAF_LOG_WARNING("cannot send error message: no route to source");
...@@ -224,20 +224,20 @@ void instance::handle_heartbeat(execution_unit* ctx) { ...@@ -224,20 +224,20 @@ void instance::handle_heartbeat(execution_unit* ctx) {
} }
} }
optional<routing_table::route> instance::lookup(const node_id& target) { optional<instance::endpoint_handle> instance::lookup(const node_id& target) {
return tbl_.lookup(target); return tbl_.lookup(target);
} }
void instance::flush(const routing_table::route& path) { void instance::flush(endpoint_handle hdl) {
callee_.flush(path.hdl); callee_.flush(hdl);
} }
void instance::write(execution_unit* ctx, const routing_table::route& r, void instance::write(execution_unit* ctx, instance::endpoint_handle hdl,
header& hdr, payload_writer* writer) { header& hdr, payload_writer* writer) {
CAF_LOG_TRACE(CAF_ARG(hdr)); CAF_LOG_TRACE(CAF_ARG(hdr));
CAF_ASSERT(hdr.payload_len == 0 || writer != nullptr); CAF_ASSERT(hdr.payload_len == 0 || writer != nullptr);
write(ctx, callee_.get_buffer(r.hdl), hdr, writer); write(ctx, callee_.get_buffer(hdl), hdr, writer);
flush(r); flush(hdl);
} }
void instance::add_published_actor(uint16_t port, void instance::add_published_actor(uint16_t port,
...@@ -295,7 +295,7 @@ size_t instance::remove_published_actor(const actor_addr& whom, ...@@ -295,7 +295,7 @@ size_t instance::remove_published_actor(const actor_addr& whom,
bool instance::is_greater(sequence_type lhs, sequence_type rhs, bool instance::is_greater(sequence_type lhs, sequence_type rhs,
sequence_type max_distance) { sequence_type max_distance) {
// distance between lhs and rhs is smaller than max_distance. // Distance between lhs and rhs is smaller than max_distance.
return ((lhs > rhs) && (lhs - rhs <= max_distance)) || return ((lhs > rhs) && (lhs - rhs <= max_distance)) ||
((lhs < rhs) && (rhs - lhs > max_distance)); ((lhs < rhs) && (rhs - lhs > max_distance));
} }
...@@ -307,8 +307,8 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender, ...@@ -307,8 +307,8 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(receiver) CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(receiver)
<< CAF_ARG(mid) << CAF_ARG(msg)); << CAF_ARG(mid) << CAF_ARG(msg));
CAF_ASSERT(receiver && system().node() != receiver->node()); CAF_ASSERT(receiver && system().node() != receiver->node());
auto path = lookup(receiver->node()); auto ehdl = lookup(receiver->node());
if (!path) { if (!ehdl) {
notify<hook::message_sending_failed>(sender, receiver, mid, msg); notify<hook::message_sending_failed>(sender, receiver, mid, msg);
return false; return false;
} }
...@@ -319,10 +319,10 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender, ...@@ -319,10 +319,10 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
header hdr{message_type::dispatch_message, 0, 0, mid.integer_value(), header hdr{message_type::dispatch_message, 0, 0, mid.integer_value(),
sender ? sender->node() : this_node(), receiver->node(), sender ? sender->node() : this_node(), receiver->node(),
sender ? sender->id() : invalid_actor_id, receiver->id(), sender ? sender->id() : invalid_actor_id, receiver->id(),
visit(seq_num_visitor{callee_}, path->hdl)}; visit(seq_num_visitor{callee_}, *ehdl)};
write(ctx, callee_.get_buffer(path->hdl), hdr, &writer); write(ctx, callee_.get_buffer(*ehdl), hdr, &writer);
flush(*path); flush(*ehdl);
notify<hook::message_sent>(sender, path->next_hop, receiver, mid, msg); notify<hook::message_sent>(sender, receiver->node(), receiver, mid, msg);
return true; return true;
} }
...@@ -332,7 +332,7 @@ void instance::write(execution_unit* ctx, buffer_type& buf, ...@@ -332,7 +332,7 @@ void instance::write(execution_unit* ctx, buffer_type& buf,
error err; error err;
if (pw != nullptr) { if (pw != nullptr) {
auto pos = buf.size(); auto pos = buf.size();
// write payload first (skip first 72 bytes and write header later) // Write payload first (skip first 72 bytes and write header later).
char placeholder[basp::header_size]; char placeholder[basp::header_size];
buf.insert(buf.end(), std::begin(placeholder), std::end(placeholder)); buf.insert(buf.end(), std::begin(placeholder), std::end(placeholder));
binary_serializer bs{ctx, buf}; binary_serializer bs{ctx, buf};
......
...@@ -33,58 +33,19 @@ routing_table::~routing_table() { ...@@ -33,58 +33,19 @@ routing_table::~routing_table() {
// nop // nop
} }
optional<routing_table::route> routing_table::lookup(const node_id& target) { node_id routing_table::lookup(const endpoint_handle& hdl) const {
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_direct(const endpoint_handle& hdl) const {
return get_opt(direct_by_hdl_, hdl, none); return get_opt(direct_by_hdl_, hdl, none);
} }
optional<routing_table::endpoint_handle> optional<routing_table::endpoint_handle>
routing_table::lookup_direct(const node_id& nid) const { routing_table::lookup(const node_id& nid) const {
auto i = direct_by_nid_.find(nid); auto i = direct_by_nid_.find(nid);
if (i != direct_by_nid_.end()) if (i != direct_by_nid_.end())
return i->second; return i->second;
return none; return none;
} }
node_id routing_table::lookup_indirect(const node_id& nid) const { void routing_table::erase(const endpoint_handle& hdl, erase_callback& cb) {
auto i = indirect_.find(nid);
if (i == indirect_.end())
return none;
if (i->second.empty())
return none;
return *i->second.begin();
}
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::erase_direct(const endpoint_handle& hdl,
erase_callback& cb) {
auto i = direct_by_hdl_.find(hdl); auto i = direct_by_hdl_.find(hdl);
if (i == direct_by_hdl_.end()) if (i == direct_by_hdl_.end())
return; return;
...@@ -94,19 +55,7 @@ void routing_table::erase_direct(const endpoint_handle& hdl, ...@@ -94,19 +55,7 @@ void routing_table::erase_direct(const endpoint_handle& hdl,
direct_by_hdl_.erase(i->first); direct_by_hdl_.erase(i->first);
} }
bool routing_table::erase_indirect(const node_id& dest) { void routing_table::add(const endpoint_handle& hdl, const node_id& nid) {
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;
}
void routing_table::add_direct(const endpoint_handle& hdl,
const node_id& nid) {
CAF_ASSERT(direct_by_hdl_.count(hdl) == 0); CAF_ASSERT(direct_by_hdl_.count(hdl) == 0);
CAF_ASSERT(direct_by_nid_.count(nid) == 0); CAF_ASSERT(direct_by_nid_.count(nid) == 0);
direct_by_hdl_.emplace(hdl, nid); direct_by_hdl_.emplace(hdl, nid);
...@@ -114,42 +63,8 @@ void routing_table::add_direct(const endpoint_handle& hdl, ...@@ -114,42 +63,8 @@ void routing_table::add_direct(const endpoint_handle& hdl,
parent_->parent().notify<hook::new_connection_established>(nid); parent_->parent().notify<hook::new_connection_established>(nid);
} }
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
}
bool routing_table::reachable(const node_id& dest) { bool routing_table::reachable(const node_id& dest) {
return direct_by_nid_.count(dest) > 0 || indirect_.count(dest) > 0; return direct_by_nid_.count(dest) > 0;
}
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;
} }
} // namespace basp } // namespace basp
......
...@@ -52,7 +52,7 @@ constexpr uint64_t no_operation_data = 0; ...@@ -52,7 +52,7 @@ constexpr uint64_t no_operation_data = 0;
constexpr auto basp_atom = caf::atom("BASP"); constexpr auto basp_atom = caf::atom("BASP");
constexpr auto spawn_serv_atom = caf::atom("SpawnServ"); constexpr auto spawn_serv_atom = caf::atom("SpawnServ");
constexpr auto config_serv_atom = caf::atom("ConfigServ"); // constexpr auto config_serv_atom = caf::atom("ConfigServ");
} // namespace <anonymous> } // namespace <anonymous>
...@@ -310,10 +310,9 @@ public: ...@@ -310,10 +310,9 @@ public:
make_message(sys_atom::value, get_atom::value, "info")); make_message(sys_atom::value, get_atom::value, "info"));
// test whether basp instance correctly updates the // test whether basp instance correctly updates the
// routing table upon receiving client handshakes // routing table upon receiving client handshakes
auto path = tbl().lookup(n.id); auto ehdl = tbl().lookup(n.id);
CAF_REQUIRE(path); CAF_REQUIRE(ehdl);
CAF_CHECK_EQUAL(path->hdl, n.connection); CAF_CHECK_EQUAL(*ehdl, n.connection);
CAF_CHECK_EQUAL(path->next_hop, n.id);
} }
std::pair<basp::header, buffer> read_from_out_buf(connection_handle hdl) { std::pair<basp::header, buffer> read_from_out_buf(connection_handle hdl) {
...@@ -723,6 +722,7 @@ CAF_TEST(actor_serialize_and_deserialize) { ...@@ -723,6 +722,7 @@ CAF_TEST(actor_serialize_and_deserialize) {
std::vector<actor_id>{}, msg); std::vector<actor_id>{}, msg);
} }
/*
CAF_TEST(indirect_connections) { CAF_TEST(indirect_connections) {
// this node receives a message from jupiter via mars and responds via mars // this node receives a message from jupiter via mars and responds via mars
// and any ad-hoc automatic connection requests are ignored // and any ad-hoc automatic connection requests are ignored
...@@ -772,11 +772,13 @@ CAF_TEST(indirect_connections) { ...@@ -772,11 +772,13 @@ CAF_TEST(indirect_connections) {
std::vector<actor_id>{}, std::vector<actor_id>{},
make_message("hello from earth!")); make_message("hello from earth!"));
} }
*/
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(basp_tests_with_autoconn, autoconn_enabled_fixture) CAF_TEST_FIXTURE_SCOPE(basp_tests_with_autoconn, autoconn_enabled_fixture)
/*
CAF_TEST(automatic_connection) { CAF_TEST(automatic_connection) {
// this tells our BASP broker to enable the automatic connection feature // this tells our BASP broker to enable the automatic connection feature
//anon_send(aut(), ok_atom::value, //anon_send(aut(), ok_atom::value,
...@@ -787,9 +789,9 @@ CAF_TEST(automatic_connection) { ...@@ -787,9 +789,9 @@ CAF_TEST(automatic_connection) {
// but then also establishes a connection to jupiter directly) // but then also establishes a connection to jupiter directly)
auto check_node_in_tbl = [&](node& n) { auto check_node_in_tbl = [&](node& n) {
io::id_visitor id_vis; io::id_visitor id_vis;
auto hdl = tbl().lookup_direct(n.id); auto ehdl = tbl().lookup(n.id);
CAF_REQUIRE(hdl); CAF_REQUIRE(ehdl);
CAF_CHECK_EQUAL(visit(id_vis, *hdl), n.connection.id()); CAF_CHECK_EQUAL(visit(id_vis, *ehdl), n.connection.id());
}; };
mpx()->provide_scribe("jupiter", 8080, jupiter().connection); mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080)); CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
...@@ -831,8 +833,8 @@ CAF_TEST(automatic_connection) { ...@@ -831,8 +833,8 @@ CAF_TEST(automatic_connection) {
no_operation_data, this_node(), jupiter().id, no_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id()); invalid_actor_id, jupiter().dummy_actor->id());
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u); 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(jupiter().id), mars().id);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none); CAF_CHECK_EQUAL(tbl().lookup(mars().id), none);
auto connection_helper_actor = sys.latest_actor_id(); auto connection_helper_actor = sys.latest_actor_id();
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u); CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
// create a dummy config server and respond to the name lookup // create a dummy config server and respond to the name lookup
...@@ -887,5 +889,6 @@ CAF_TEST(automatic_connection) { ...@@ -887,5 +889,6 @@ CAF_TEST(automatic_connection) {
make_message("hello from earth!")); make_message("hello from earth!"));
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u); CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
} }
*/
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -56,7 +56,7 @@ constexpr uint64_t no_operation_data = 0; ...@@ -56,7 +56,7 @@ constexpr uint64_t no_operation_data = 0;
constexpr auto basp_atom = caf::atom("BASP"); constexpr auto basp_atom = caf::atom("BASP");
constexpr auto spawn_serv_atom = caf::atom("SpawnServ"); constexpr auto spawn_serv_atom = caf::atom("SpawnServ");
constexpr auto config_serv_atom = caf::atom("ConfigServ"); // constexpr auto config_serv_atom = caf::atom("ConfigServ");
} // namespace <anonymous> } // namespace <anonymous>
...@@ -309,10 +309,9 @@ public: ...@@ -309,10 +309,9 @@ public:
make_message(sys_atom::value, get_atom::value, "info")); make_message(sys_atom::value, get_atom::value, "info"));
// test whether basp instance correctly updates the // test whether basp instance correctly updates the
// routing table upon receiving client handshakes // routing table upon receiving client handshakes
auto path = tbl().lookup(n.id); auto ehdl= tbl().lookup(n.id);
CAF_REQUIRE(path); CAF_REQUIRE(ehdl);
CAF_CHECK_EQUAL(path->hdl, n.endpoint); CAF_CHECK_EQUAL(*ehdl, n.endpoint);
CAF_CHECK_EQUAL(path->next_hop, n.id);
} }
std::pair<basp::header, buffer> read_from_out_buf(datagram_handle hdl) { std::pair<basp::header, buffer> read_from_out_buf(datagram_handle hdl) {
...@@ -829,6 +828,7 @@ CAF_TEST(actor_serialize_and_deserialize_udp) { ...@@ -829,6 +828,7 @@ CAF_TEST(actor_serialize_and_deserialize_udp) {
std::vector<actor_id>{}, msg); std::vector<actor_id>{}, msg);
} }
/*
CAF_TEST(indirect_connections_udp) { CAF_TEST(indirect_connections_udp) {
// this node receives a message from jupiter via mars and responds via mars // this node receives a message from jupiter via mars and responds via mars
// and any ad-hoc automatic connection requests are ignored // and any ad-hoc automatic connection requests are ignored
...@@ -878,6 +878,7 @@ CAF_TEST(indirect_connections_udp) { ...@@ -878,6 +878,7 @@ CAF_TEST(indirect_connections_udp) {
std::vector<actor_id>{}, std::vector<actor_id>{},
make_message("hello from earth!")); make_message("hello from earth!"));
} }
*/
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -1023,6 +1024,7 @@ CAF_TEST_FIXTURE_SCOPE_END() ...@@ -1023,6 +1024,7 @@ CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(basp_udp_tests_with_autoconn, autoconn_enabled_fixture) CAF_TEST_FIXTURE_SCOPE(basp_udp_tests_with_autoconn, autoconn_enabled_fixture)
/*
CAF_TEST(automatic_connection_udp) { CAF_TEST(automatic_connection_udp) {
// this tells our BASP broker to enable the automatic connection feature // this tells our BASP broker to enable the automatic connection feature
//anon_send(aut(), ok_atom::value, //anon_send(aut(), ok_atom::value,
...@@ -1033,9 +1035,9 @@ CAF_TEST(automatic_connection_udp) { ...@@ -1033,9 +1035,9 @@ CAF_TEST(automatic_connection_udp) {
// but then also establishes a connection to jupiter directly) // but then also establishes a connection to jupiter directly)
auto check_node_in_tbl = [&](node& n) { auto check_node_in_tbl = [&](node& n) {
io::id_visitor id_vis; io::id_visitor id_vis;
auto hdl = tbl().lookup_direct(n.id); auto ehdl = tbl().lookup(n.id);
CAF_REQUIRE(hdl); CAF_REQUIRE(ehdl);
CAF_CHECK_EQUAL(visit(id_vis, *hdl), n.endpoint.id()); CAF_CHECK_EQUAL(visit(id_vis, *ehdl), n.endpoint.id());
}; };
mpx()->provide_datagram_servant("jupiter", 8080, jupiter().endpoint); mpx()->provide_datagram_servant("jupiter", 8080, jupiter().endpoint);
CAF_CHECK(mpx()->has_pending_remote_endpoint("jupiter", 8080)); CAF_CHECK(mpx()->has_pending_remote_endpoint("jupiter", 8080));
...@@ -1142,5 +1144,6 @@ CAF_TEST(automatic_connection_udp) { ...@@ -1142,5 +1144,6 @@ CAF_TEST(automatic_connection_udp) {
make_message("hello from earth!")); make_message("hello from earth!"));
CAF_CHECK_EQUAL(mpx()->output_queue(mars().endpoint).size(), 0u); CAF_CHECK_EQUAL(mpx()->output_queue(mars().endpoint).size(), 0u);
} }
*/
CAF_TEST_FIXTURE_SCOPE_END() 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