Commit 5f5e4932 authored by ufownl's avatar ufownl

Add heartbeats to BASP; streamline basp_broker

Streamline handling of `connection_closed_msg'.
parent 90a9c015
......@@ -122,6 +122,7 @@ public:
atom_value middleman_network_backend;
bool middleman_enable_automatic_connections;
size_t middleman_max_consecutive_reads;
size_t middleman_basp_heartbeat_interval;
// System parameters that are set while initializing modules.
node_id network_id;
......
......@@ -139,6 +139,9 @@ using spawn_atom = atom_constant<atom("spawn")>;
/// Used for migrating actors to other nodes.
using migrate_atom = atom_constant<atom("migrate")>;
/// Used for triggering periodic operations.
using tick_atom = atom_constant<atom("tick")>;
} // namespace caf
namespace std {
......
......@@ -142,6 +142,7 @@ actor_system_config::actor_system_config() {
middleman_network_backend = atom("default");
middleman_enable_automatic_connections = false;
middleman_max_consecutive_reads = 50;
middleman_basp_heartbeat_interval = 0;
nexus_port = 0;
}
......@@ -172,6 +173,8 @@ actor_system_config::actor_system_config(int argc, char** argv)
middleman_enable_automatic_connections)
.bind("middleman.max_consecutive_reads",
middleman_max_consecutive_reads)
.bind("middleman.basp-heartbeat-interval",
middleman_basp_heartbeat_interval)
.bind("probe.nexus-host", nexus_host)
.bind("probe.nexus-port", nexus_port);
detail::parse_ini(ini, consumer, std::cerr);
......@@ -206,6 +209,9 @@ actor_system_config::actor_system_config(int argc, char** argv)
{"caf#middleman.max_consecutive_reads",
"sets the maximum number of allowed I/O reads before scheduling others",
middleman_max_consecutive_reads},
{"caf#middleman.basp-heartbeat-interval",
"sets the interval (ms) of basp-heartbeat, 0 (default) means disabling it",
middleman_basp_heartbeat_interval},
{"caf#probe.nexus-host",
"sets the hostname or IP address for connecting to the Nexus",
nexus_host},
......@@ -252,7 +258,9 @@ actor_system_config::actor_system_config(int argc, char** argv)
<< "network-backend="
<< deep_to_string(middleman_network_backend) << endl
<< "enable-automatic-connections="
<< deep_to_string(middleman_enable_automatic_connections) << endl;
<< deep_to_string(middleman_enable_automatic_connections) << endl
<< "basp-heartbeat-interval="
<< middleman_basp_heartbeat_interval << endl;
}
args_remainder = std::move(res.remainder);
}
......
......@@ -204,7 +204,12 @@ enum class message_type : uint32_t {
/// that has been terminated.
///
/// ![](kill_proxy_instance.png)
kill_proxy_instance = 0x04
kill_proxy_instance = 0x04,
/// Send to remote node which has direct connection.
///
/// ![](heartbeat.png)
heartbeat = 0x05,
};
/// @relates message_type
......@@ -414,6 +419,9 @@ public:
/// 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`
virtual void handle_heartbeat(const node_id& nid) = 0;
/// Returns the actor namespace associated to this BASP protocol instance.
inline proxy_registry& proxies() {
return namespace_;
......@@ -441,8 +449,8 @@ public:
connection_state handle(execution_unit* ctx,
new_data_msg& dm, header& hdr, bool is_payload);
/// Handles connection shutdowns.
void handle(const connection_closed_msg& msg);
/// Sends heartbeat messages to all valid nodes those are directly connected.
void handle_heartbeat(execution_unit* ctx);
/// Handles failure or shutdown of a single node. This function purges
/// all routes to `affected_node` from the routing table.
......@@ -545,6 +553,10 @@ public:
actor_id aid,
exit_reason rsn);
/// Writes a `heartbeat` to `buf`.
void write_heartbeat(execution_unit* ctx,
buffer_type& buf, const node_id& remote_side);
inline const node_id& this_node() const {
return this_node_;
}
......@@ -572,6 +584,11 @@ inline bool is_handshake(const header& hdr) {
|| hdr.operation == message_type::client_handshake;
}
/// Checks wheter given header contains a heartbeat.
inline bool is_heartbeat(const header& hdr) {
return hdr.operation == message_type::heartbeat;
}
/// @}
} // namespace basp
......
......@@ -80,6 +80,10 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::listener
void learned_new_node_indirectly(const node_id& nid) override;
void handle_heartbeat(const node_id&) override {
// nop
}
// stores meta information for open connections
struct connection_context {
// denotes what message we expect from the remote node next
......@@ -98,8 +102,6 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
void set_context(connection_handle hdl);
bool erase_context(connection_handle hdl);
// pointer to ourselves
broker* self;
......
......@@ -315,6 +315,8 @@ private:
hook_uptr hooks_;
// actor offering asyncronous IO by managing this singleton instance
middleman_actor manager_;
// configure parameters
size_t basp_heartbeat_interval_;
};
} // namespace io
......
......@@ -46,6 +46,8 @@ std::string to_string(message_type x) {
return "announce_proxy_instance";
case message_type::kill_proxy_instance:
return "kill_proxy_instance";
case message_type::heartbeat:
return "heartbeat";
default:
return "???";
}
......@@ -129,6 +131,16 @@ bool kill_proxy_instance_valid(const header& hdr) {
&& ! zero(hdr.operation_data);
}
bool heartbeat_valid(const header& hdr) {
return valid(hdr.source_node)
&& valid(hdr.dest_node)
&& hdr.source_node != hdr.dest_node
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor)
&& zero(hdr.payload_len)
&& zero(hdr.operation_data);
}
} // namespace <anonymous>
bool valid(const header& hdr) {
......@@ -145,6 +157,8 @@ bool valid(const header& hdr) {
return announce_proxy_instance_valid(hdr);
case message_type::kill_proxy_instance:
return kill_proxy_instance_valid(hdr);
case message_type::heartbeat:
return heartbeat_valid(hdr);
}
}
......@@ -337,7 +351,9 @@ connection_state instance::handle(execution_unit* ctx,
return err();
}
// needs forwarding?
if (! is_handshake(hdr) && hdr.dest_node != this_node_) {
if (! is_handshake(hdr)
&& ! is_heartbeat(hdr)
&& hdr.dest_node != this_node_) {
auto path = lookup(hdr.dest_node);
if (path) {
binary_serializer bs{ctx, std::back_inserter(path->wr_buf)};
......@@ -452,6 +468,11 @@ connection_state instance::handle(execution_unit* ctx,
callee_.kill_proxy(hdr.source_node, hdr.source_actor,
static_cast<exit_reason>(hdr.operation_data));
break;
case message_type::heartbeat: {
CAF_LOG_TRACE("received heartbeat: " << CAF_ARG(hdr.source_node));
callee_.handle_heartbeat(hdr.source_node);
break;
}
default:
CAF_LOG_ERROR("invalid operation");
return err();
......@@ -459,11 +480,12 @@ connection_state instance::handle(execution_unit* ctx,
return await_header;
}
void instance::handle(const connection_closed_msg& msg) {
auto cb = make_callback([&](const node_id& nid){
callee_.purge_state(nid);
});
tbl_.erase_direct(msg.handle, cb);
void instance::handle_heartbeat(execution_unit* ctx) {
for (auto& kvp: tbl_.direct_by_hdl_) {
CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second));
write_heartbeat(ctx, tbl_.parent_->wr_buf(kvp.first), kvp.second);
tbl_.parent_->flush(kvp.first);
}
}
void instance::handle_node_shutdown(const node_id& affected_node) {
......@@ -672,6 +694,13 @@ void instance::write_kill_proxy_instance(execution_unit* ctx,
write(ctx, buf, hdr);
}
void instance::write_heartbeat(execution_unit* ctx,
buffer_type& buf,
const node_id& remote_side) {
write(ctx, buf, message_type::heartbeat, nullptr, 0,
this_node_, remote_side, invalid_actor_id, invalid_actor_id);
}
} // namespace basp
} // namespace io
} // namespace caf
......@@ -20,6 +20,7 @@
#include "caf/io/basp_broker.hpp"
#include <limits>
#include <chrono>
#include "caf/sec.hpp"
#include "caf/send.hpp"
......@@ -153,8 +154,16 @@ void basp_broker_state::purge_state(const node_id& nid) {
auto hdl = instance.tbl().lookup_direct(nid);
if (hdl == invalid_connection_handle)
return;
auto i = ctx.find(hdl);
if (i != ctx.end()) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
ctx.erase(i);
}
proxies().erase(nid);
ctx.erase(hdl);
known_remotes.erase(nid);
}
......@@ -424,20 +433,6 @@ void basp_broker_state::set_context(connection_handle hdl) {
this_context = &i->second;
}
bool basp_broker_state::erase_context(connection_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
auto i = ctx.find(hdl);
if (i == ctx.end())
return false;
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
ctx.erase(i);
return true;
}
/******************************************************************************
* basp_broker *
******************************************************************************/
......@@ -538,19 +533,13 @@ behavior basp_broker::make_behavior() {
// received from underlying broker implementation
[=](const connection_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
if (! state.erase_context(msg.handle))
return;
// TODO: currently we assume a node has gone offline once we lose
// a connection, we also could try to reach this node via other
// hops to be resilient to (rare) network failures or if a
// node is reachable via several interfaces and only one fails
auto nid = state.instance.tbl().lookup_direct(msg.handle);
if (nid != invalid_node_id) {
// tell BASP instance we've lost connection
state.instance.handle_node_shutdown(nid);
// remove all proxies
state.proxies().erase(nid);
}
// tell BASP instance we've lost connection
state.instance.handle_node_shutdown(nid);
CAF_ASSERT(nid == invalid_node_id
|| ! state.instance.tbl().reachable(nid));
},
......@@ -669,6 +658,11 @@ behavior basp_broker::make_behavior() {
}
return std::make_tuple(x, std::move(addr), port);
},
[=](tick_atom, size_t interval) {
state.instance.handle_heartbeat(context());
delayed_send(this, std::chrono::milliseconds{interval},
tick_atom::value, interval);
},
// catch-all error handler
others >> [=] {
CAF_LOG_ERROR("unexpected message:" << CAF_ARG(current_message()));
......
......@@ -286,6 +286,10 @@ void middleman::start() {
backend().thread_id(thread_.get_id());
}
auto basp = named_broker<basp_broker>(atom("BASP"));
if (basp_heartbeat_interval_ > 0) {
CAF_LOG_INFO("enable basp-heartbeat: " << CAF_ARG(basp_heartbeat_interval_));
anon_send(basp, tick_atom::value, basp_heartbeat_interval_);
}
manager_ = make_middleman_actor(system(), basp);
}
......@@ -343,6 +347,7 @@ void middleman::init(actor_system_config& cfg) {
// set scheduling parameters for multiplexer
backend().max_throughput(cfg.scheduler_max_throughput);
backend().max_consecutive_reads(cfg.middleman_max_consecutive_reads);
basp_heartbeat_interval_ = cfg.middleman_basp_heartbeat_interval;
}
actor_system::module::id_t middleman::id() const {
......
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