Commit fc3f966b authored by Dominik Charousset's avatar Dominik Charousset

Merge pull request #433 from ufownl/topic/basp_heartbeat

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