Commit f2b1460c authored by Dominik Charousset's avatar Dominik Charousset

Disconnect nodes when reaching connection_timeout

parent 801a458f
...@@ -132,11 +132,13 @@ enum class sec : uint8_t { ...@@ -132,11 +132,13 @@ enum class sec : uint8_t {
malformed_basp_message, malformed_basp_message,
/// The middleman closed a connection because it failed to serialize or /// The middleman closed a connection because it failed to serialize or
/// deserialize a payload. /// deserialize a payload.
serializing_basp_payload_failed, serializing_basp_payload_failed = 50,
/// The middleman closed a connection to itself or an already connected node. /// The middleman closed a connection to itself or an already connected node.
redundant_connection, redundant_connection,
/// Resolving a path on a remote node failed. /// Resolving a path on a remote node failed.
remote_lookup_failed, remote_lookup_failed,
/// Disconnected from a BASP node after reaching the connection timeout.
connection_timeout,
}; };
/// @relates sec /// @relates sec
......
...@@ -117,6 +117,8 @@ std::string to_string(sec x) { ...@@ -117,6 +117,8 @@ std::string to_string(sec x) {
return "redundant_connection"; return "redundant_connection";
case sec::remote_lookup_failed: case sec::remote_lookup_failed:
return "remote_lookup_failed"; return "remote_lookup_failed";
case sec::connection_timeout:
return "connection_timeout";
}; };
} }
......
...@@ -345,9 +345,9 @@ behavior basp_broker::make_behavior() { ...@@ -345,9 +345,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;
...@@ -357,6 +357,20 @@ behavior basp_broker::make_behavior() { ...@@ -357,6 +357,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::value, next_tick, scheduled_send(this, next_tick, tick_atom::value, next_tick,
heartbeat_interval, connection_timeout); 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