Commit 01c4e2e5 authored by Dominik Charousset's avatar Dominik Charousset

Disconnect nodes when reaching connection_timeout

(cherry picked from commit f2b1460c)
parent f6b3eb61
...@@ -161,6 +161,8 @@ enum class sec : uint8_t { ...@@ -161,6 +161,8 @@ enum class sec : uint8_t {
no_such_key = 65, no_such_key = 65,
/// An destroyed a response promise without calling deliver or delegate on it. /// An destroyed a response promise without calling deliver or delegate on it.
broken_promise, broken_promise,
/// Disconnected from a BASP node after reaching the connection timeout.
connection_timeout,
}; };
// --(rst-sec-end)-- // --(rst-sec-end)--
......
...@@ -150,6 +150,8 @@ std::string to_string(sec x) { ...@@ -150,6 +150,8 @@ std::string to_string(sec x) {
return "caf::sec::no_such_key"; return "caf::sec::no_such_key";
case sec::broken_promise: case sec::broken_promise:
return "caf::sec::broken_promise"; return "caf::sec::broken_promise";
case sec::connection_timeout:
return "caf::sec::connection_timeout";
}; };
} }
...@@ -355,6 +357,9 @@ bool from_string(string_view in, sec& out) { ...@@ -355,6 +357,9 @@ bool from_string(string_view in, sec& out) {
} else if (in == "caf::sec::broken_promise") { } else if (in == "caf::sec::broken_promise") {
out = sec::broken_promise; out = sec::broken_promise;
return true; return true;
} else if (in == "caf::sec::connection_timeout") {
out = sec::connection_timeout;
return true;
} else { } else {
return false; return false;
} }
...@@ -433,6 +438,7 @@ bool from_integer(std::underlying_type_t<sec> in, ...@@ -433,6 +438,7 @@ bool from_integer(std::underlying_type_t<sec> in,
case sec::unsupported_operation: case sec::unsupported_operation:
case sec::no_such_key: case sec::no_such_key:
case sec::broken_promise: case sec::broken_promise:
case sec::connection_timeout:
out = result; out = result;
return true; return true;
}; };
......
...@@ -383,9 +383,9 @@ behavior basp_broker::make_behavior() { ...@@ -383,9 +383,9 @@ behavior basp_broker::make_behavior() {
} }
auto next_tick = scheduled + heartbeat_interval; auto next_tick = scheduled + heartbeat_interval;
if (now >= next_tick) { if (now >= next_tick) {
CAF_LOG_ERROR( CAF_LOG_ERROR("Lagging a full heartbeat interval behind! "
"Lagging a full heartbeat interval behind! Interval too low " "Interval too low or BASP actor overloaded! "
"or BASP actor overloaded! Other nodes may disconnect."); "Other nodes may disconnect.");
while (now >= next_tick) while (now >= next_tick)
next_tick += heartbeat_interval; next_tick += heartbeat_interval;
...@@ -395,6 +395,20 @@ behavior basp_broker::make_behavior() { ...@@ -395,6 +395,20 @@ behavior basp_broker::make_behavior() {
} }
// Send out heartbeats. // Send out heartbeats.
instance.handle_heartbeat(context()); instance.handle_heartbeat(context());
// Check whether any node reached the disconnect timeout.
for (auto i = ctx.begin(); i != ctx.end();) {
if (i->second.last_seen + connection_timeout < now) {
CAF_LOG_WARNING("Disconnect BASP node: reached connection timeout!");
auto hdl = i->second.hdl;
// connection_cleanup below calls ctx.erase, so we need to increase
// the iterator now, before it gets invalidated.
++i;
connection_cleanup(hdl, sec::connection_timeout);
close(hdl);
} else {
++i;
}
}
// Schedule next tick. // Schedule next tick.
scheduled_send(this, next_tick, tick_atom_v, scheduled_send(this, next_tick, tick_atom_v,
next_tick.time_since_epoch().count(), heartbeat_interval, next_tick.time_since_epoch().count(), heartbeat_interval,
......
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