Commit f6b3eb61 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.

(cherry picked from commit 801a458f)
parent 9b5e6cdd
......@@ -113,7 +113,8 @@ namespace caf::defaults::middleman {
constexpr auto app_identifier = string_view{"generic-caf-app"};
constexpr auto network_backend = string_view{"default"};
constexpr auto max_consecutive_reads = size_t{50};
constexpr auto heartbeat_interval = size_t{0};
constexpr auto heartbeat_interval = timespan{0};
constexpr auto connection_timeout = timespan{0};
constexpr auto cached_udp_buffers = size_t{10};
constexpr auto max_pending_msgs = size_t{10};
......
......@@ -108,9 +108,17 @@ behavior basp_broker::make_behavior() {
}
auto heartbeat_interval = get_or(config(), "caf.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_v, 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);
// Note: we the scheduled time as integer representation to avoid having to
// assign a type ID to the time_point type.
scheduled_send(this, first_tick, tick_atom_v,
first_tick.time_since_epoch().count(), heartbeat_interval,
connection_timeout);
}
return behavior{
// received from underlying broker implementation
......@@ -361,10 +369,36 @@ behavior basp_broker::make_behavior() {
}
return {x, std::move(addr), port};
},
[=](tick_atom, size_t interval) {
[=](tick_atom, actor_clock::time_point::rep scheduled_rep,
timespan heartbeat_interval, timespan connection_timeout) {
auto scheduled_tse = actor_clock::time_point::duration{scheduled_rep};
auto scheduled = actor_clock::time_point{scheduled_tse};
auto now = clock().now();
if (now < scheduled) {
CAF_LOG_WARNING("received tick before its time, reschedule");
scheduled_send(this, scheduled, tick_atom_v,
scheduled.time_since_epoch().count(), 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_v,
interval);
// Schedule next tick.
scheduled_send(this, next_tick, tick_atom_v,
next_tick.time_since_epoch().count(), heartbeat_interval,
connection_timeout);
}};
}
......
......@@ -186,7 +186,7 @@ void middleman::add_module_options(actor_system_config& cfg) {
.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",
......
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