Commit 97272b36 authored by Joseph Noir's avatar Joseph Noir

Remove routing

parent 1ba9006e
......@@ -45,11 +45,12 @@ namespace basp {
/// Describes a protocol instance managing multiple connections.
class instance {
public:
using endpoint_handle = routing_table::endpoint_handle;
/// Provides a callback-based interface for certain BASP events.
class callee {
protected:
using buffer_type = std::vector<char>;
using endpoint_handle = variant<connection_handle, datagram_handle>;
using endpoint_handle = instance::endpoint_handle;
public:
explicit callee(actor_system& sys, proxy_registry::backend& backend);
......@@ -83,14 +84,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;
......@@ -176,15 +171,15 @@ public:
/// Sends heartbeat messages to all valid nodes those are directly connected.
void handle_heartbeat(execution_unit* ctx);
/// Returns a route to `target` or `none` on error.
optional<routing_table::route> lookup(const node_id& target);
/// Returns the handle for communication with `target` or `none` on error.
optional<endpoint_handle> lookup(const node_id& target);
/// Flushes the underlying buffer of `path`.
void flush(const routing_table::route& path);
/// Flushes the underlying buffer of `hdl`.
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.
void write(execution_unit* ctx, const routing_table::route& r,
void write(execution_unit* ctx, endpoint_handle hdl,
header& hdr, payload_writer* writer = nullptr);
/// Adds a new actor to the map of published actors.
......@@ -294,11 +289,11 @@ public:
bool handle(execution_unit* ctx, const Handle& hdl, header& hdr,
std::vector<char>* payload, bool tcp_based,
optional<endpoint_context&> ep, optional<uint16_t> port) {
// function object for checking payload validity
// Function object for checking payload validity.
auto payload_valid = [&]() -> bool {
return payload != nullptr && payload->size() == hdr.payload_len;
};
// handle message to ourselves
// Handle message to ourselves.
switch (hdr.operation) {
case message_type::server_handshake: {
actor_id aid = invalid_actor_id;
......@@ -320,37 +315,31 @@ public:
if (e)
return false;
}
// close self connection after handshake is done
// Close self connection after handshake is done.
if (hdr.source_node == this_node_) {
CAF_LOG_INFO("close connection to self immediately");
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return false;
}
// close this connection if we already have a direct connection
if (tbl_.lookup_direct(hdr.source_node)) {
// Close this connection if we already established communication.
// 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 "
"direct connection: " << CAF_ARG(hdr.source_node));
"connection: " << CAF_ARG(hdr.source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return false;
}
// 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);
// write handshake as client in response
auto path = tbl_.lookup(hdr.source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
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);
// Add this node to our contacts.
CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node));
tbl_.add(hdl, hdr.source_node);
// TODO: Add status, addresses, ... ?
// Write handshake as client in response.
if (tcp_based)
write_client_handshake(ctx, callee_.get_buffer(hdl), hdr.source_node);
callee_.learned_new_node(hdr.source_node);
callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path);
flush(hdl);
break;
}
case message_type::client_handshake: {
......@@ -368,49 +357,35 @@ public:
return false;
}
}
if (tcp_based) {
if (tbl_.lookup_direct(hdr.source_node)) {
auto new_node = (this_node() != hdr.source_node
&& !tbl_.lookup(hdr.source_node));
if (!new_node) {
if (tcp_based) {
CAF_LOG_INFO("received second client handshake:"
<< CAF_ARG(hdr.source_node));
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 {
auto new_node = (this_node() != hdr.source_node
&& !tbl_.lookup_direct(hdr.source_node));
if (new_node) {
// 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);
}
// Add this node to our contacts.
CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node));
tbl_.add(hdl, hdr.source_node);
// TODO: Add status, addresses, ... ?
}
// 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;
write_server_handshake(ctx,
callee_.get_buffer(hdl),
port, seq);
write_server_handshake(ctx, callee_.get_buffer(hdl), port, seq);
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;
}
case message_type::dispatch_message: {
if (!payload_valid())
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};
auto receiver_name = static_cast<atom_value>(0);
std::vector<strong_actor_ptr> forwarding_stack;
......
......@@ -30,6 +30,8 @@
#include "caf/io/basp/buffer_type.hpp"
#include "caf/io/network/interfaces.hpp"
namespace std {
template<>
......@@ -58,59 +60,30 @@ public:
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
/// 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 endpoint_handle& hdl) const;
node_id lookup(const endpoint_handle& hdl) const;
/// Returns the handle offering a direct connection to `nid` or
/// `invalid_connection_handle` if no direct connection to `nid` exists.
optional<endpoint_handle> lookup_direct(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;
/// Returns the handle for communication with `nid` or
/// `none` if `nid` is unknown.
optional<endpoint_handle> lookup(const node_id& nid) const;
/// Adds a new direct route to the table.
/// @pre `hdl != invalid_connection_handle && nid != none`
void add_direct(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);
void add(const endpoint_handle& hdl, 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 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);
void erase(const endpoint_handle& hdl, erase_callback& cb);
/// Queries whether `dest` is reachable.
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_;
......@@ -126,16 +99,11 @@ 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<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_;
indirect_entries indirect_;
indirect_entries blacklist_;
};
/// @}
......
......@@ -81,14 +81,7 @@ 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);
// 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;
void learned_new_node(const node_id& nid) override;
// inherited from basp::instance::callee
uint16_t next_sequence_number(connection_handle hdl) override;
......
......@@ -43,7 +43,7 @@ namespace io {
namespace {
// visitors to access handle variant of the context
// Visitors to access handle variant of the context.
struct seq_num_visitor {
using result_type = basp::sequence_type;
seq_num_visitor(basp_broker_state* ptr) : state(ptr) { }
......@@ -83,7 +83,7 @@ basp_broker_state::basp_broker_state(broker* selfptr)
}
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)
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) {
CAF_ASSERT(nid != this_node());
if (nid == none || aid == invalid_actor_id)
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
// us a handle to a third node B, then we assume that A offers a route to B
if (nid != this_context->id
&& !instance.tbl().lookup_direct(nid)
&& instance.tbl().add_indirect(this_context->id, nid))
learned_new_node_indirectly(nid);
// we need to tell remote side we are watching this actor now;
// use a direct route if possible, i.e., when talking to a third node
auto path = instance.tbl().lookup(nid);
if (!path) {
// this happens if and only if we don't have a path to `nid`
// and current_context_->hdl has been blacklisted
// us a handle to a third node B, then we assume that A offers a route to B.
if (nid != this_context->id && !instance.tbl().lookup(nid))
// TODO: Try to establish communication with the new node.
CAF_CRITICAL("Not implemented.");
// TODO: Everything below has to happen once we establish communication?
// I'm not sure yet, the functors can be attached earlier and we could
// send trigger an error message if we cannot contact the remote node.
// We need to tell remote side we are watching this actor now;
// use a direct route if possible, i.e., when talking to a third node.
auto ehdl = instance.tbl().lookup(nid);
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 "
"running on a node we don't have a route to");
return nullptr;
}
// create proxy and add functor that will be called if we
// receive a kill_proxy_instance message
// Create proxy and add functor that will be called if we
// receive a kill_proxy_instance message.
auto mm = &system().middleman();
actor_config cfg;
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) {
strong_actor_ptr selfptr{self->ctrl()};
res->get()->attach_functor([=](const error& rsn) {
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
// bugs with attachables
// bugs with attachables.
auto bptr = static_cast<basp_broker*>(selfptr->get());
if (!bptr->getf(abstract_actor::is_terminated_flag))
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) {
"write announce_proxy_instance:"
<< CAF_ARG(nid) << CAF_ARG(aid));
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(),
get_buffer(this_context->hdl),
nid, aid,
ctx.requires_ordering ? ctx.seq_outgoing++ : 0);
instance.flush(*path);
instance.flush(*ehdl);
mm->notify<hook::new_remote_actor>(res);
return res;
}
......@@ -154,10 +156,10 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
if (cb == none)
return;
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 (nid == this_node()) {
// connected to self
// Connected to self.
ptr = actor_cast<strong_actor_ptr>(system().registry().get(aid));
CAF_LOG_INFO_IF(!ptr, "actor not found:" << CAF_ARG(aid));
} else {
......@@ -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));
if (rsn == none)
rsn = exit_reason::unknown;
auto path = instance.tbl().lookup(nid);
if (!path) {
auto ehdl = instance.tbl().lookup(nid);
if (!ehdl) {
CAF_LOG_INFO("cannot send exit message for proxy, no route to host:"
<< CAF_ARG(nid));
return;
}
auto hdl = std::move(*ehdl);
instance.write_kill_proxy(self->context(),
get_buffer(path->hdl),
get_buffer(hdl),
nid, aid, rsn,
visit(seq_num_visitor{this}, path->hdl));
instance.flush(*path);
visit(seq_num_visitor{this}, hdl));
instance.flush(hdl);
}
void basp_broker_state::proxy_announced(const node_id& nid, actor_id 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);
if (ptr == nullptr) {
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);
} else {
auto entry = ptr->address();
......@@ -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 {
CAF_LOG_TRACE("");
// terminate when receiving a down message
// Terminate when receiving a down message.
tself->set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
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);
return {
[=](ok_atom, const std::string& /* key == "info" */,
......@@ -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");
return sink(name_atm, stages, msg);
});
auto path = instance.tbl().lookup(nid);
if (!path) {
auto ehdl = instance.tbl().lookup(nid);
if (!ehdl) {
CAF_LOG_ERROR("learned_new_node called, but no route to nid");
return;
}
auto hdl = std::move(*ehdl);
// send message to SpawnServ of remote node
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)};
visit(seq_num_visitor{this}, hdl)};
// writing std::numeric_limits<actor_id>::max() is a hack to get
// 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);
instance.flush(*path);
}
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");
instance.flush(hdl);
}
void basp_broker_state::set_context(connection_handle hdl) {
......@@ -484,7 +435,7 @@ void basp_broker_state::cleanup(connection_handle hdl) {
purge_state(nid);
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
// this connection was closed during handshake.
auto i = ctx_tcp.find(hdl);
......@@ -507,7 +458,7 @@ void basp_broker_state::cleanup(datagram_handle hdl) {
purge_state(nid);
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
// this connection was closed during handshake.
auto i = ctx_udp.find(hdl);
......@@ -644,8 +595,8 @@ behavior basp_broker::make_behavior() {
state.enable_udp = system().config().middleman_enable_udp;
if (system().config().middleman_enable_automatic_connections) {
CAF_LOG_INFO("enable automatic connections");
// open a random port and store a record for our peers how to
// connect to this broker directly in the configuration server
// Open a random port and store a record for our peers how to
// connect to this broker directly in the configuration server.
if (state.enable_tcp) {
auto res = add_tcp_doorman(uint16_t{0});
if (res) {
......@@ -676,7 +627,7 @@ behavior basp_broker::make_behavior() {
send(this, tick_atom::value, heartbeat_interval);
}
return {
// received from underlying broker implementation
// Received from underlying broker implementation.
[=](new_data_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
state.set_context(msg.handle);
......@@ -696,7 +647,7 @@ behavior basp_broker::make_behavior() {
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) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
auto hdl = ptr->hdl();
......@@ -707,11 +658,11 @@ behavior basp_broker::make_behavior() {
ctx.local_port = local_port(hdl);
ctx.requires_ordering = true;
ctx.seq_incoming = 0;
ctx.seq_outgoing = 1; // already sent the client handshake
// Let's not implement this twice
ctx.seq_outgoing = 1; // Already sent the client handshake.
// Let's not implement this twice.
send(this, std::move(msg));
},
// received from underlying broker implementation
// Received from underlying broker implementation.
[=](new_datagram_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
state.set_context(msg.handle);
......@@ -728,12 +679,12 @@ behavior basp_broker::make_behavior() {
close(msg.handle);
}
},
// received from the underlying broker implementation
// Received from the underlying broker implementation.
[=](datagram_sent_msg& msg) {
if (state.cached_buffers.size() < state.max_buffers)
state.cached_buffers.emplace(std::move(msg.buf));
},
// received from proxy instances
// Received from proxy instances.
[=](forward_atom, strong_actor_ptr& src,
const std::vector<strong_actor_ptr>& fwd_stack,
strong_actor_ptr& dest, message_id mid, const message& msg) {
......@@ -753,7 +704,7 @@ behavior basp_broker::make_behavior() {
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,
const message& msg) -> result<message> {
auto cme = current_mailbox_element();
......@@ -766,11 +717,12 @@ behavior basp_broker::make_behavior() {
<< ", " << CAF_ARG(msg));
if (!src)
return sec::invalid_argument;
auto path = this->state.instance.tbl().lookup(dest_node);
if (!path) {
CAF_LOG_ERROR("no route to receiving node");
auto ehdl = this->state.instance.tbl().lookup(dest_node);
if (!ehdl) {
CAF_LOG_ERROR("receiving node unknown");
return sec::no_route_to_receiving_node;
}
auto hdl = std::move(*ehdl);
if (system().node() == src->node())
system().registry().put(src->id(), src);
auto writer = make_callback([&](serializer& sink) -> error {
......@@ -780,13 +732,13 @@ behavior basp_broker::make_behavior() {
basp::header::named_receiver_flag,
0, cme->mid.integer_value(), state.this_node(),
dest_node, src->id(), invalid_actor_id,
visit(seq_num_visitor{&state}, path->hdl)};
state.instance.write(context(), state.get_buffer(path->hdl),
visit(seq_num_visitor{&state}, hdl)};
state.instance.write(context(), state.get_buffer(hdl),
hdr, &writer);
state.instance.flush(*path);
state.instance.flush(hdl);
return delegated<message>();
},
// received from underlying broker implementation
// Received from underlying broker implementation.
[=](const new_connection_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
auto& bi = state.instance;
......@@ -795,18 +747,18 @@ behavior basp_broker::make_behavior() {
state.flush(msg.handle);
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) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
state.cleanup(msg.handle);
},
// received from underlying broker implementation
// Received from underlying broker implementation.
[=](const acceptor_closed_msg& msg) {
CAF_LOG_TRACE("");
auto port = local_port(msg.handle);
state.instance.remove_published_actor(port);
},
// received from middleman actor
// Received from middleman actor.
[=](publish_atom, doorman_ptr& ptr, uint16_t port,
const strong_actor_ptr& whom, std::set<std::string>& sigs) {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(port)
......@@ -817,7 +769,7 @@ behavior basp_broker::make_behavior() {
system().registry().put(whom->id(), whom);
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) {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(port));
CAF_ASSERT(ptr != nullptr);
......@@ -830,7 +782,7 @@ behavior basp_broker::make_behavior() {
ctx.cstate = basp::await_header;
ctx.callback = rp;
ctx.requires_ordering = false;
// await server handshake
// Await server handshake.
configure_read(hdl, receive_policy::exactly(basp::header_size));
},
[=](publish_udp_atom, datagram_servant_ptr& ptr, uint16_t port,
......@@ -843,7 +795,7 @@ behavior basp_broker::make_behavior() {
system().registry().put(whom->id(), whom);
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) {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(port));
auto rp = make_response_promise();
......@@ -862,11 +814,11 @@ behavior basp_broker::make_behavior() {
none, ctx.seq_outgoing++);
state.flush(hdl);
},
// received from underlying broker implementation
// Received from underlying broker implementation.
[=](const datagram_servant_closed_msg& msg) {
CAF_LOG_TRACE("");
// since all handles share a port, we can take any of them to query for
// port information
// Since all handles share a port, we can take any of them to query for
// port information.
CAF_ASSERT(msg.handles.size() > 0);
auto port = local_port(msg.handles.front());
state.instance.remove_published_actor(port);
......@@ -902,8 +854,8 @@ behavior basp_broker::make_behavior() {
[=](close_atom, uint16_t port) -> result<void> {
if (port == 0)
return sec::cannot_close_invalid_port;
// it is well-defined behavior to not have an actor published here,
// hence the result can be ignored safely
// It is well-defined behavior to not have an actor published here,
// hence the result can be ignored safely.
state.instance.remove_published_actor(port, nullptr);
auto res = close(hdl_by_port(port));
if (res)
......@@ -914,10 +866,10 @@ behavior basp_broker::make_behavior() {
-> std::tuple<node_id, std::string, uint16_t> {
std::string addr;
uint16_t port = 0;
auto hdl = state.instance.tbl().lookup_direct(x);
if (hdl) {
addr = visit(addr_visitor{this}, *hdl);
port = visit(port_visitor{this}, *hdl);
auto ehdl = state.instance.tbl().lookup(x);
if (ehdl) {
addr = visit(addr_visitor{this}, *ehdl);
port = visit(port_visitor{this}, *ehdl);
}
return std::make_tuple(x, std::move(addr), port);
},
......
......@@ -63,13 +63,13 @@ connection_state instance::handle(execution_unit* ctx,
new_data_msg& dm, header& hdr,
bool 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 cb = make_callback([&](const node_id& nid) -> error {
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;
......@@ -93,23 +93,23 @@ connection_state instance::handle(execution_unit* ctx,
}
}
CAF_LOG_DEBUG(CAF_ARG(hdr));
// needs forwarding?
// Needs forwarding?
if (!is_handshake(hdr) && !is_heartbeat(hdr) && hdr.dest_node != this_node_) {
CAF_LOG_DEBUG("forward message");
auto path = lookup(hdr.dest_node);
if (path) {
binary_serializer bs{ctx, callee_.get_buffer(path->hdl)};
auto ehdl = lookup(hdr.dest_node);
if (ehdl) {
binary_serializer bs{ctx, callee_.get_buffer(*ehdl)};
auto e = bs(hdr);
if (e)
return err();
if (payload != nullptr)
bs.apply_raw(payload->size(), payload->data());
flush(*path);
flush(*ehdl);
notify<hook::message_forwarded>(hdr, payload);
} else {
CAF_LOG_INFO("cannot forward message, no route to destination");
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);
if (!reverse_path) {
CAF_LOG_WARNING("cannot send error message: no route to source");
......@@ -131,22 +131,22 @@ connection_state instance::handle(execution_unit* ctx,
bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
endpoint_context& ep) {
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 cb = make_callback([&](const node_id& nid) -> error {
callee_.purge_state(nid);
return none;
});
tbl_.erase_direct(dm.handle, cb);
tbl_.erase(dm.handle, cb);
return false;
};
// extract payload
// Extract payload.
std::vector<char> pl_buf{std::move_iterator<itr_t>(std::begin(dm.buf) +
basp::header_size),
std::move_iterator<itr_t>(std::end(dm.buf))};
// resize header
// Resize header.
dm.buf.resize(basp::header_size);
// extract header
// Extract header.
binary_deserializer bd{ctx, dm.buf};
auto e = bd(ep.hdr);
if (e || !valid(ep.hdr)) {
......@@ -175,24 +175,24 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
}
// This is the expected message.
ep.seq_incoming += 1;
// TODO: add optional reliability here
// TODO: Add optional reliability here.
if (!is_handshake(ep.hdr) && !is_heartbeat(ep.hdr)
&& ep.hdr.dest_node != this_node_) {
CAF_LOG_DEBUG("forward message");
auto path = lookup(ep.hdr.dest_node);
if (path) {
binary_serializer bs{ctx, callee_.get_buffer(path->hdl)};
auto ehdl = lookup(ep.hdr.dest_node);
if (ehdl) {
binary_serializer bs{ctx, callee_.get_buffer(*ehdl)};
auto ex = bs(ep.hdr);
if (ex)
return err();
if (payload != nullptr)
bs.apply_raw(payload->size(), payload->data());
flush(*path);
flush(*ehdl);
notify<hook::message_forwarded>(ep.hdr, payload);
} else {
CAF_LOG_INFO("cannot forward message, no route to destination");
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);
if (!reverse_path) {
CAF_LOG_WARNING("cannot send error message: no route to source");
......@@ -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);
}
void instance::flush(const routing_table::route& path) {
callee_.flush(path.hdl);
void instance::flush(endpoint_handle 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) {
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,
......@@ -295,7 +295,7 @@ size_t instance::remove_published_actor(const actor_addr& whom,
bool instance::is_greater(sequence_type lhs, sequence_type rhs,
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)) ||
((lhs < rhs) && (rhs - lhs > max_distance));
}
......@@ -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_ARG(mid) << CAF_ARG(msg));
CAF_ASSERT(receiver && system().node() != receiver->node());
auto path = lookup(receiver->node());
if (!path) {
auto ehdl = lookup(receiver->node());
if (!ehdl) {
notify<hook::message_sending_failed>(sender, receiver, mid, msg);
return false;
}
......@@ -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(),
sender ? sender->node() : this_node(), receiver->node(),
sender ? sender->id() : invalid_actor_id, receiver->id(),
visit(seq_num_visitor{callee_}, path->hdl)};
write(ctx, callee_.get_buffer(path->hdl), hdr, &writer);
flush(*path);
notify<hook::message_sent>(sender, path->next_hop, receiver, mid, msg);
visit(seq_num_visitor{callee_}, *ehdl)};
write(ctx, callee_.get_buffer(*ehdl), hdr, &writer);
flush(*ehdl);
notify<hook::message_sent>(sender, receiver->node(), receiver, mid, msg);
return true;
}
......@@ -332,7 +332,7 @@ void instance::write(execution_unit* ctx, buffer_type& buf,
error err;
if (pw != nullptr) {
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];
buf.insert(buf.end(), std::begin(placeholder), std::end(placeholder));
binary_serializer bs{ctx, buf};
......
......@@ -33,58 +33,19 @@ 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_direct(const endpoint_handle& hdl) const {
node_id routing_table::lookup(const endpoint_handle& hdl) const {
return get_opt(direct_by_hdl_, hdl, none);
}
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);
if (i != direct_by_nid_.end())
return i->second;
return none;
}
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::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) {
void routing_table::erase(const endpoint_handle& hdl, erase_callback& cb) {
auto i = direct_by_hdl_.find(hdl);
if (i == direct_by_hdl_.end())
return;
......@@ -94,19 +55,7 @@ void routing_table::erase_direct(const endpoint_handle& hdl,
direct_by_hdl_.erase(i->first);
}
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;
}
void routing_table::add_direct(const endpoint_handle& hdl,
const node_id& nid) {
void routing_table::add(const endpoint_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);
......@@ -114,42 +63,8 @@ void routing_table::add_direct(const endpoint_handle& hdl,
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) {
return direct_by_nid_.count(dest) > 0 || indirect_.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;
return direct_by_nid_.count(dest) > 0;
}
} // namespace basp
......
......@@ -52,7 +52,7 @@ constexpr uint64_t no_operation_data = 0;
constexpr auto basp_atom = caf::atom("BASP");
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>
......@@ -310,10 +310,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 = tbl().lookup(n.id);
CAF_REQUIRE(path);
CAF_CHECK_EQUAL(path->hdl, n.connection);
CAF_CHECK_EQUAL(path->next_hop, n.id);
auto ehdl = tbl().lookup(n.id);
CAF_REQUIRE(ehdl);
CAF_CHECK_EQUAL(*ehdl, n.connection);
}
std::pair<basp::header, buffer> read_from_out_buf(connection_handle hdl) {
......@@ -723,6 +722,7 @@ CAF_TEST(actor_serialize_and_deserialize) {
std::vector<actor_id>{}, msg);
}
/*
CAF_TEST(indirect_connections) {
// this node receives a message from jupiter via mars and responds via mars
// and any ad-hoc automatic connection requests are ignored
......@@ -772,11 +772,13 @@ CAF_TEST(indirect_connections) {
std::vector<actor_id>{},
make_message("hello from earth!"));
}
*/
CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(basp_tests_with_autoconn, autoconn_enabled_fixture)
/*
CAF_TEST(automatic_connection) {
// this tells our BASP broker to enable the automatic connection feature
//anon_send(aut(), ok_atom::value,
......@@ -787,9 +789,9 @@ CAF_TEST(automatic_connection) {
// but then also establishes a connection to jupiter directly)
auto check_node_in_tbl = [&](node& n) {
io::id_visitor id_vis;
auto hdl = tbl().lookup_direct(n.id);
CAF_REQUIRE(hdl);
CAF_CHECK_EQUAL(visit(id_vis, *hdl), n.connection.id());
auto ehdl = tbl().lookup(n.id);
CAF_REQUIRE(ehdl);
CAF_CHECK_EQUAL(visit(id_vis, *ehdl), n.connection.id());
};
mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
......@@ -831,8 +833,8 @@ CAF_TEST(automatic_connection) {
no_operation_data, this_node(), jupiter().id,
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);
CAF_CHECK_EQUAL(tbl().lookup(jupiter().id), mars().id);
CAF_CHECK_EQUAL(tbl().lookup(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
......@@ -887,5 +889,6 @@ CAF_TEST(automatic_connection) {
make_message("hello from earth!"));
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
}
*/
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -56,7 +56,7 @@ constexpr uint64_t no_operation_data = 0;
constexpr auto basp_atom = caf::atom("BASP");
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>
......@@ -309,10 +309,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 = tbl().lookup(n.id);
CAF_REQUIRE(path);
CAF_CHECK_EQUAL(path->hdl, n.endpoint);
CAF_CHECK_EQUAL(path->next_hop, n.id);
auto ehdl= tbl().lookup(n.id);
CAF_REQUIRE(ehdl);
CAF_CHECK_EQUAL(*ehdl, n.endpoint);
}
std::pair<basp::header, buffer> read_from_out_buf(datagram_handle hdl) {
......@@ -829,6 +828,7 @@ CAF_TEST(actor_serialize_and_deserialize_udp) {
std::vector<actor_id>{}, msg);
}
/*
CAF_TEST(indirect_connections_udp) {
// this node receives a message from jupiter via mars and responds via mars
// and any ad-hoc automatic connection requests are ignored
......@@ -878,6 +878,7 @@ CAF_TEST(indirect_connections_udp) {
std::vector<actor_id>{},
make_message("hello from earth!"));
}
*/
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(automatic_connection_udp) {
// this tells our BASP broker to enable the automatic connection feature
//anon_send(aut(), ok_atom::value,
......@@ -1033,9 +1035,9 @@ CAF_TEST(automatic_connection_udp) {
// but then also establishes a connection to jupiter directly)
auto check_node_in_tbl = [&](node& n) {
io::id_visitor id_vis;
auto hdl = tbl().lookup_direct(n.id);
CAF_REQUIRE(hdl);
CAF_CHECK_EQUAL(visit(id_vis, *hdl), n.endpoint.id());
auto ehdl = tbl().lookup(n.id);
CAF_REQUIRE(ehdl);
CAF_CHECK_EQUAL(visit(id_vis, *ehdl), n.endpoint.id());
};
mpx()->provide_datagram_servant("jupiter", 8080, jupiter().endpoint);
CAF_CHECK(mpx()->has_pending_remote_endpoint("jupiter", 8080));
......@@ -1142,5 +1144,6 @@ CAF_TEST(automatic_connection_udp) {
make_message("hello from earth!"));
CAF_CHECK_EQUAL(mpx()->output_queue(mars().endpoint).size(), 0u);
}
*/
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