Commit 801a458f authored by Dominik Charousset's avatar Dominik Charousset

Re-implement heartbeat using scheduled_send

By using `scheduled_send` instead of `delayed_send`, we can keep track
of the timeliness of our heartbeats. If the BASP broker processes its
tick after the next interval is over, we can assume an overload
situation or too low heartbeat intervals.

Further, the type of the `heartbeat_interval` constant was changed to
`timespan` in order to fix a mismatch between the actor system config
and the BASP broker. Relates #1095.
parent b00983e6
......@@ -79,7 +79,8 @@ namespace middleman {
extern std::vector<std::string> app_identifiers;
extern const atom_value network_backend;
extern const size_t max_consecutive_reads;
extern const size_t heartbeat_interval;
extern const timespan heartbeat_interval;
extern const timespan connection_timeout;
extern const size_t cached_udp_buffers;
extern const size_t max_pending_msgs;
extern const size_t workers;
......
......@@ -127,7 +127,7 @@ actor_system_config::actor_system_config()
.add<timespan>("heartbeat-interval", "interval of heartbeat messages")
.add<timespan>("connection-timeout",
"max. time between messages before declaring a node dead "
"(ignored if heartbeats are disabled)")
"(disabled if 0, ignored if heartbeats are disabled)")
.add<bool>("attach-utility-actors",
"schedule utility actors instead of dedicating threads")
.add<bool>("manual-multiplexing",
......
......@@ -110,7 +110,8 @@ namespace middleman {
std::vector<std::string> app_identifiers{"generic-caf-app"};
const atom_value network_backend = atom("default");
const size_t max_consecutive_reads = 50;
const size_t heartbeat_interval = 0;
const timespan heartbeat_interval = timespan{0};
const timespan connection_timeout = timespan{0};
const size_t cached_udp_buffers = 10;
const size_t max_pending_msgs = 10;
const size_t workers = min(3u, std::thread::hardware_concurrency() / 4u) + 1;
......
......@@ -116,9 +116,14 @@ behavior basp_broker::make_behavior() {
}
auto heartbeat_interval = get_or(config(), "middleman.heartbeat-interval",
defaults::middleman::heartbeat_interval);
if (heartbeat_interval > 0) {
if (heartbeat_interval.count() > 0) {
CAF_LOG_DEBUG("enable heartbeat" << CAF_ARG(heartbeat_interval));
send(this, tick_atom::value, heartbeat_interval);
auto now = clock().now();
auto first_tick = now + heartbeat_interval;
auto connection_timeout = get_or(config(), "middleman.connection-timeout",
defaults::middleman::connection_timeout);
scheduled_send(this, first_tick, tick_atom::value, first_tick,
heartbeat_interval, connection_timeout);
}
return behavior{
// received from underlying broker implementation
......@@ -329,10 +334,32 @@ behavior basp_broker::make_behavior() {
}
return std::make_tuple(x, std::move(addr), port);
},
[=](tick_atom, size_t interval) {
[=](tick_atom, actor_clock::time_point scheduled,
timespan heartbeat_interval, timespan connection_timeout) {
auto now = clock().now();
if (now < scheduled) {
CAF_LOG_WARNING("received tick before its time, reschedule");
scheduled_send(this, scheduled, tick_atom::value, scheduled,
heartbeat_interval, connection_timeout);
return;
}
auto next_tick = scheduled + heartbeat_interval;
if (now >= next_tick) {
CAF_LOG_ERROR(
"Lagging a full heartbeat interval behind! Interval too low "
"or BASP actor overloaded! Other nodes may disconnect.");
while (now >= next_tick)
next_tick += heartbeat_interval;
} else if (now >= scheduled + (heartbeat_interval / 2)) {
CAF_LOG_WARNING("Lagging more than 50% of a heartbeat interval behind! "
"Interval too low or BASP actor overloaded!");
}
// Send out heartbeats.
instance.handle_heartbeat(context());
delayed_send(this, std::chrono::milliseconds{interval}, tick_atom::value,
interval);
// Schedule next tick.
scheduled_send(this, next_tick, tick_atom::value, next_tick,
heartbeat_interval, connection_timeout);
}};
}
......
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