Commit 24073a6d authored by Joseph Noir's avatar Joseph Noir

Track node connectivity, adjust lookup function

parent 83db9a07
......@@ -170,9 +170,6 @@ public:
/// Sends heartbeat messages to all valid nodes those are directly connected.
void handle_heartbeat(execution_unit* ctx);
/// Returns the handle for communication with `target` or `none` on error.
optional<endpoint_handle> lookup(const node_id& target);
/// Flushes the underlying buffer of `hdl`.
void flush(endpoint_handle hdl);
......
......@@ -51,6 +51,7 @@ namespace basp {
/// Stores routing information for a single broker participating as
/// BASP peer and provides both direct and indirect paths.
// TODO: Rename `routing_table`.
class routing_table {
public:
using endpoint_handle = variant<connection_handle, datagram_handle>;
......@@ -59,6 +60,24 @@ public:
virtual ~routing_table();
/// Describes the communication status for a remote endpoint.
enum connectivity {
/// There is currently an ongoing communication attempt.
pending,
/// Communication with the node is established.
established,
/// Communication failure detected.
failed
};
/// Result for lookups that includes the state and a potential handle.
struct contact {
/// Tracks the state to determine if we can sent messages or have to buffer.
connectivity conn;
/// Servant handle to communicate with the node -- if already created.
optional<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&>;
......@@ -67,14 +86,18 @@ public:
/// `none` if `hdl` is unknown.
node_id lookup(const endpoint_handle& hdl) const;
/// Returns the handle for communication with `nid` or
/// `none` if `nid` is unknown.
optional<endpoint_handle> lookup(const node_id& nid) const;
/// Returns the state for communication with `nid` along with a handle
/// if communication is established or `none` if `nid` is unknown.
optional<contact> lookup(const node_id& nid) const;
/// Adds a new direct route to the table.
/// Adds a new endpoint to the table.
/// @pre `hdl != invalid_connection_handle && nid != none`
void add(const endpoint_handle& hdl, const node_id& nid);
/// Adds a new node that is not reachable yet. It's state is set to `pending`.
/// @pre `nid != none`
void add(const node_id& nid);
/// Removes a direct connection and calls `cb` for any node
/// that became unreachable as a result of this operation,
/// including the node that is assigned as direct path for `hdl`.
......@@ -88,7 +111,35 @@ public:
return parent_;
}
/// Set the communication state of node with `nid`.
bool status(const node_id& nid, connectivity new_status);
/// Get the communication state of node with `nid`.
optional<connectivity> status(const node_id& nid);
/// Set the forwarding node that first mentioned `hdl`.
bool forwarder(const node_id& nid, endpoint_handle hdl);
/// Get the forwarding node that first mentioned `hdl`
/// or `none` if the node is unknown.
optional<endpoint_handle> forwarder(const node_id& nid);
/// Add `addrs` to the addresses to reach `nid`.
bool addresses(const node_id& nid, network::address_listing addrs);
/// Get the addresses to reach `nid` or `none` if the node is unknown.
optional<const network::address_listing&> addresses(const node_id& nid);
public:
/// Entry to bundle information for a remote endpoint.
struct node_info {
contact details;
/// Interfaces of the nodes for sharing with neighbors.
network::address_listing addrs;
/// The endpoint who told us about the node.
optional<endpoint_handle> origin;
};
template <class Map, class Fallback>
typename Map::mapped_type
get_opt(const Map& m, const typename Map::key_type& k, Fallback&& x) const {
......@@ -99,10 +150,12 @@ public:
}
abstract_broker* parent_;
std::unordered_map<endpoint_handle, node_id> direct_by_hdl_;
std::unordered_map<endpoint_handle, node_id> nid_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> hdl_by_nid_;
std::unordered_map<node_id, node_info> node_information_base_;
};
/// @}
......
......@@ -104,8 +104,9 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
// 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) {
auto ec = instance.tbl().lookup(nid);
// TODO: Should this communication already be established?
if (!ec) {
// 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 "
......@@ -138,7 +139,7 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
get_buffer(this_context->hdl),
nid, aid,
ctx.requires_ordering ? ctx.seq_outgoing++ : 0);
instance.flush(*ehdl);
instance.flush(*ec->hdl);
mm->notify<hook::new_remote_actor>(res);
return res;
}
......@@ -185,18 +186,25 @@ 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 ehdl = instance.tbl().lookup(nid);
if (!ehdl) {
CAF_LOG_INFO("cannot send exit message for proxy, no route to host:"
auto ec = instance.tbl().lookup(nid);
/// TODO: Let's assume that the handle is valid if the status is established.
if (!ec || ec->conn == basp::routing_table::connectivity::failed) {
CAF_LOG_INFO("cannot send exit message for proxy, host unreachable"
<< CAF_ARG(nid));
return;
}
auto hdl = std::move(*ehdl);
instance.write_kill_proxy(self->context(),
get_buffer(hdl),
nid, aid, rsn,
visit(seq_num_visitor{this}, hdl));
instance.flush(hdl);
auto c = std::move(*ec);
if (c.conn == basp::routing_table::connectivity::established) {
instance.write_kill_proxy(self->context(),
get_buffer(*c.hdl),
nid, aid, rsn,
visit(seq_num_visitor{this}, *c.hdl));
instance.flush(*c.hdl);
} else {
// TODO: Buffer message!
CAF_CRITICAL("ibasp_broker_state::send_kill_proxy_instance with buffering "
"not implemented!");
}
}
void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
......@@ -369,22 +377,28 @@ 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 ehdl = instance.tbl().lookup(nid);
if (!ehdl) {
auto ec = instance.tbl().lookup(nid);
if (!ec || ec->conn == basp::routing_table::connectivity::failed) {
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}, 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(hdl),
hdr, &writer);
instance.flush(hdl);
auto c = std::move(*ec);
if (c.conn == basp::routing_table::connectivity::established) {
// 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}, *c.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(*c.hdl),
hdr, &writer);
instance.flush(*c.hdl);
} else {
// TODO: Implement this, can it happen?
CAF_CRITICAL("basp_broker_state::learned_new_node with buffering not "
"implemented!");
}
}
void basp_broker_state::set_context(connection_handle hdl) {
......@@ -717,25 +731,31 @@ behavior basp_broker::make_behavior() {
<< ", " << CAF_ARG(msg));
if (!src)
return sec::invalid_argument;
auto ehdl = this->state.instance.tbl().lookup(dest_node);
if (!ehdl) {
CAF_LOG_ERROR("receiving node unknown");
auto ec = this->state.instance.tbl().lookup(dest_node);
if (!ec || ec->conn == basp::routing_table::connectivity::failed) {
CAF_LOG_ERROR("host unknown or unreachable");
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 {
return sink(dest_name, cme->stages, const_cast<message&>(msg));
});
basp::header hdr{basp::message_type::dispatch_message,
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}, hdl)};
state.instance.write(context(), state.get_buffer(hdl),
hdr, &writer);
state.instance.flush(hdl);
auto c = std::move(*ec);
if (c.conn == basp::routing_table::connectivity::established) {
if (system().node() == src->node())
system().registry().put(src->id(), src);
auto writer = make_callback([&](serializer& sink) -> error {
return sink(dest_name, cme->stages, const_cast<message&>(msg));
});
basp::header hdr{basp::message_type::dispatch_message,
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}, *c.hdl)};
state.instance.write(context(), state.get_buffer(*c.hdl),
hdr, &writer);
state.instance.flush(*c.hdl);
} else {
// TODO: Buffer the message in the basp broker.
CAF_CRITICAL("basp_broker forward_atom with buffering "
"not implemented!");
}
return delegated<message>();
},
// Received from underlying broker implementation.
......@@ -866,10 +886,10 @@ behavior basp_broker::make_behavior() {
-> std::tuple<node_id, std::string, uint16_t> {
std::string addr;
uint16_t port = 0;
auto ehdl = state.instance.tbl().lookup(x);
if (ehdl) {
addr = visit(addr_visitor{this}, *ehdl);
port = visit(port_visitor{this}, *ehdl);
auto ec = state.instance.tbl().lookup(x);
if (ec && ec->hdl) {
addr = visit(addr_visitor{this}, *ec->hdl);
port = visit(port_visitor{this}, *ec->hdl);
}
return std::make_tuple(x, std::move(addr), port);
},
......
......@@ -59,9 +59,8 @@ instance::instance(abstract_broker* parent, callee& lstnr)
CAF_ASSERT(this_node_ != none);
}
connection_state instance::handle(execution_unit* ctx,
new_data_msg& dm, header& hdr,
bool is_payload) {
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.
auto err = [&]() -> connection_state {
......@@ -95,33 +94,8 @@ connection_state instance::handle(execution_unit* ctx,
CAF_LOG_DEBUG(CAF_ARG(hdr));
// Needs forwarding?
if (!is_handshake(hdr) && !is_heartbeat(hdr) && hdr.dest_node != this_node_) {
CAF_LOG_DEBUG("forward message");
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(*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.
auto reverse_path = lookup(hdr.source_node);
if (!reverse_path) {
CAF_LOG_WARNING("cannot send error message: no route to source");
} else {
CAF_LOG_WARNING("not implemented yet: signalize forward failure");
}
} else {
CAF_LOG_WARNING("lost packet with probably spoofed source");
}
notify<hook::message_forwarding_failed>(hdr, payload);
}
return await_header;
// TODO: Forwarding should no longer happen.
return err();
}
if (!handle(ctx, dm.handle, hdr, payload, true, none, none))
return err();
......@@ -177,34 +151,9 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
ep.seq_incoming += 1;
// 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 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(*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.
auto reverse_path = lookup(ep.hdr.source_node);
if (!reverse_path) {
CAF_LOG_WARNING("cannot send error message: no route to source");
} else {
CAF_LOG_WARNING("not implemented yet: signalize forward failure");
}
} else {
CAF_LOG_WARNING("lost packet with probably spoofed source");
}
notify<hook::message_forwarding_failed>(ep.hdr, payload);
}
return true;
&& ep.hdr.dest_node != this_node_) {
// TODO: Forwarding should no longer happen.
return err();
}
if (!handle(ctx, dm.handle, ep.hdr, payload, false, ep, ep.local_port))
return err();
......@@ -216,7 +165,7 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
void instance::handle_heartbeat(execution_unit* ctx) {
CAF_LOG_TRACE("");
for (auto& kvp: tbl_.direct_by_hdl_) {
for (auto& kvp: tbl_.nid_by_hdl_) {
CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second));
write_heartbeat(ctx, callee_.get_buffer(kvp.first),
kvp.second, visit(seq_num_visitor{callee_}, kvp.first));
......@@ -224,10 +173,6 @@ void instance::handle_heartbeat(execution_unit* ctx) {
}
}
optional<instance::endpoint_handle> instance::lookup(const node_id& target) {
return tbl_.lookup(target);
}
void instance::flush(endpoint_handle hdl) {
callee_.flush(hdl);
}
......@@ -307,23 +252,32 @@ 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 ehdl = lookup(receiver->node());
if (!ehdl) {
auto ec = tbl_.lookup(receiver->node());
/// TODO: Let's assume that the handle is valid if the status is established.
if (!ec || ec->cs == routing_table::connectivity::failed) {
notify<hook::message_sending_failed>(sender, receiver, mid, msg);
return false;
}
auto writer = make_callback([&](serializer& sink) -> error {
return sink(const_cast<std::vector<strong_actor_ptr>&>(forwarding_stack),
const_cast<message&>(msg));
});
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_}, *ehdl)};
write(ctx, callee_.get_buffer(*ehdl), hdr, &writer);
flush(*ehdl);
notify<hook::message_sent>(sender, receiver->node(), receiver, mid, msg);
return true;
auto c = std::move(*ec);
if (c.cs == routing_table::connectivity::established) {
auto writer = make_callback([&](serializer& sink) -> error {
return sink(const_cast<std::vector<strong_actor_ptr>&>(forwarding_stack),
const_cast<message&>(msg));
});
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_}, *c.hdl)};
write(ctx, callee_.get_buffer(*c.hdl), hdr, &writer);
flush(*c.hdl);
notify<hook::message_sent>(sender, receiver->node(), receiver, mid, msg);
return true;
} else {
// lr.cs == routing_table::communication::pending
// TODO: Buffer the message in the basp broker.
CAF_CRITICAL("instance::disaptch with buffering not implemented!");
return false;
}
}
void instance::write(execution_unit* ctx, buffer_type& buf,
......
......@@ -34,37 +34,113 @@ routing_table::~routing_table() {
}
node_id routing_table::lookup(const endpoint_handle& hdl) const {
return get_opt(direct_by_hdl_, hdl, none);
return get_opt(nid_by_hdl_, hdl, none);
}
optional<routing_table::endpoint_handle>
optional<routing_table::contact>
routing_table::lookup(const node_id& nid) const {
auto i = direct_by_nid_.find(nid);
if (i != direct_by_nid_.end())
auto i = node_information_base_.find(nid);
if (i != node_information_base_.end())
return i->second.details;
/*
auto i = hdl_by_nid_.find(nid);
if (i != hdl_by_nid_.end())
return i->second;
*/
return none;
}
void routing_table::erase(const endpoint_handle& hdl, erase_callback& cb) {
auto i = direct_by_hdl_.find(hdl);
if (i == direct_by_hdl_.end())
auto i = nid_by_hdl_.find(hdl);
if (i == nid_by_hdl_.end())
return;
cb(i->second);
parent_->parent().notify<hook::connection_lost>(i->second);
direct_by_nid_.erase(i->second);
direct_by_hdl_.erase(i->first);
// TODO: Should we keep address information and set the state to 'none'?
node_information_base_.erase(i->second);
//hdl_by_nid_.erase(i->second);
nid_by_hdl_.erase(i->first);
}
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);
direct_by_nid_.emplace(nid, hdl);
CAF_ASSERT(nid_by_hdl_.count(hdl) == 0);
//CAF_ASSERT(hdl_by_nid_.count(nid) == 0);
CAF_ASSERT(node_information_base_.count(nid) == 0);
nid_by_hdl_.emplace(hdl, nid);
//hdl_by_nid_.emplace(nid, hdl);
node_information_base_[nid] = node_info{{connectivity::established,
hdl}, {}, none};
parent_->parent().notify<hook::new_connection_established>(nid);
}
void routing_table::add(const node_id& nid) {
//CAF_ASSERT(hdl_by_nid_.count(nid) == 0);
CAF_ASSERT(node_information_base_.count(nid) == 0);
node_information_base_[nid] = node_info{{connectivity::pending,
none}, {}, none};
// TODO: Some new related hook?
//parent_->parent().notify<hook::new_connection_established>(nid);
}
bool routing_table::reachable(const node_id& dest) {
return direct_by_nid_.count(dest) > 0;
auto i = node_information_base_.find(dest);
if (i != node_information_base_.end())
return i->second.details.cs == connectivity::established;
return false;
}
bool routing_table::status(const node_id& nid,
routing_table::connectivity new_status) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
i->second.details.cs = new_status;
return true;
}
optional<routing_table::connectivity>
routing_table::status(const node_id& nid) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return none;
return i->second.details.cs;
}
bool routing_table::forwarder(const node_id& nid,
routing_table::endpoint_handle origin) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
i->second.origin = origin;
return true;
}
optional<routing_table::endpoint_handle>
routing_table::forwarder(const node_id& nid) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return none;
return i->second.origin;
}
bool routing_table::addresses(const node_id& nid,
network::address_listing addrs) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
for (auto& p : addrs) {
auto& existing = i->second.addrs[p.first];
existing.insert(existing.end(), p.second.begin(), p.second.end());
}
return true;
}
optional<const network::address_listing&>
routing_table::addresses(const node_id& nid) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return none;
return i->second.addrs;
}
} // namespace basp
......
......@@ -310,9 +310,10 @@ public:
make_message(sys_atom::value, get_atom::value, "info"));
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
auto ehdl = tbl().lookup(n.id);
CAF_REQUIRE(ehdl);
CAF_CHECK_EQUAL(*ehdl, n.connection);
auto ec = tbl().lookup(n.id);
CAF_REQUIRE(ec);
CAF_REQUIRE(ec->hdl);
CAF_CHECK_EQUAL(*ec->hdl, n.connection);
}
std::pair<basp::header, buffer> read_from_out_buf(connection_handle hdl) {
......
......@@ -309,9 +309,10 @@ public:
make_message(sys_atom::value, get_atom::value, "info"));
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
auto ehdl= tbl().lookup(n.id);
CAF_REQUIRE(ehdl);
CAF_CHECK_EQUAL(*ehdl, n.endpoint);
auto ec= tbl().lookup(n.id);
CAF_REQUIRE(ec);
CAF_REQUIRE(ec->hdl);
CAF_CHECK_EQUAL(*ec->hdl, n.endpoint);
}
std::pair<basp::header, buffer> read_from_out_buf(datagram_handle hdl) {
......
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