Unverified Commit 9e801235 authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #1096

Disconnect BASP nodes when reaching connection timeouts
parents 422ae141 e241e486
......@@ -38,12 +38,19 @@ relaxed-sleep-duration=10ms
; configures whether MMs try to span a full mesh
enable-automatic-connections=false
; application identifier of this node, prevents connection to other CAF
; instances with different identifier
app-identifier=""
; instances with incompatible identifiers (at least one entry must match)
app-identifiers=["generic-caf-app"]
; maximum number of consecutive I/O reads per broker
max-consecutive-reads=50
; heartbeat message interval in ms (0 disables heartbeating)
heartbeat-interval=0ms
; heartbeat message interval for periodic traffic
; (0 disables heartbeating)
heartbeat-interval=0s
; force disconnects of CAF nodes after receiving no traffic for this amount of
; time (0 disables this feature, but we highly recommend using heartbeats and
; connection timeouts), be careful when deploying CAF applications with
; different heartbeat intervals and connection timeouts: this timeout should
; line up with the *longest* heartbeat interval (ideally it's a multiple)
connection-timeout=0s
; configures whether the MM attaches its internal utility actors to the
; scheduler instead of dedicating individual threads (needed only for
; deterministic testing)
......@@ -52,10 +59,6 @@ attach-utility-actors=false
; setting this to true allows fully deterministic execution in unit test and
; requires the user to trigger I/O manually
manual-multiplexing=false
; disables communication via TCP
disable-tcp=false
; enable communication via UDP
enable-udp=false
; configures how many background workers are spawned for deserialization,
; by default CAF uses 1-4 workers depending on the number of cores
workers=<min(3, number of cores / 4) + 1>
......
......@@ -78,7 +78,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;
......
......@@ -132,11 +132,13 @@ enum class sec : uint8_t {
malformed_basp_message,
/// The middleman closed a connection because it failed to serialize or
/// deserialize a payload.
serializing_basp_payload_failed,
serializing_basp_payload_failed = 50,
/// The middleman closed a connection to itself or an already connected node.
redundant_connection,
/// Resolving a path on a remote node failed.
remote_lookup_failed,
/// Disconnected from a BASP node after reaching the connection timeout.
connection_timeout,
};
/// @relates sec
......
......@@ -125,6 +125,9 @@ actor_system_config::actor_system_config()
.add<size_t>("max-consecutive-reads",
"max. number of consecutive reads per broker")
.add<timespan>("heartbeat-interval", "interval of heartbeat messages")
.add<timespan>("connection-timeout",
"max. time between messages before declaring a node dead "
"(disabled if 0, ignored if heartbeats are disabled)")
.add<bool>("attach-utility-actors",
"schedule utility actors instead of dedicating threads")
.add<bool>("manual-multiplexing",
......
......@@ -109,7 +109,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;
......
......@@ -117,6 +117,8 @@ std::string to_string(sec x) {
return "redundant_connection";
case sec::remote_lookup_failed:
return "remote_lookup_failed";
case sec::connection_timeout:
return "connection_timeout";
};
}
......
......@@ -20,14 +20,14 @@
#include <unordered_map>
#include "caf/variant.hpp"
#include "caf/response_promise.hpp"
#include "caf/io/datagram_handle.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/actor_clock.hpp"
#include "caf/io/basp/connection_state.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/datagram_handle.hpp"
#include "caf/response_promise.hpp"
#include "caf/timestamp.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace io {
......@@ -48,6 +48,8 @@ struct endpoint_context {
uint16_t local_port;
// pending operations to be performed after handshake completed
optional<response_promise> callback;
// keeps track of when we've last received a message from this endpoint
actor_clock::time_point last_seen;
};
} // namespace basp
......
......@@ -122,8 +122,8 @@ public:
connection_state handle(execution_unit* ctx,
new_data_msg& dm, header& hdr, bool is_payload);
/// Sends heartbeat messages to all valid nodes those are directly connected.
void handle_heartbeat(execution_unit* ctx);
/// Sends heartbeat messages to all connected nodes.
void send_heartbeats(execution_unit* ctx);
/// Returns a route to `target` or `none` on error.
optional<routing_table::route> lookup(const node_id& target);
......
......@@ -18,24 +18,23 @@
#pragma once
#include <future>
#include <map>
#include <set>
#include <stack>
#include <string>
#include <future>
#include <vector>
#include <unordered_map>
#include <unordered_set>
#include <vector>
#include "caf/stateful_actor.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/io/basp/all.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/typed_broker.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/stateful_actor.hpp"
namespace caf {
namespace io {
......@@ -108,6 +107,7 @@ public:
void learned_new_node(const node_id& nid);
/// Sets `this_context` by either creating or accessing state for `hdl`.
/// Automatically sets `endpoint_context::last_seen` to `clock().now()`.
void set_context(connection_handle hdl);
/// Cleans up any state for `hdl`.
......
......@@ -83,7 +83,7 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
return handle(ctx, dm.handle, hdr, payload);
}
void instance::handle_heartbeat(execution_unit* ctx) {
void instance::send_heartbeats(execution_unit* ctx) {
CAF_LOG_TRACE("");
for (auto& kvp : tbl_.direct_by_hdl_) {
CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second));
......
......@@ -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,48 @@ behavior basp_broker::make_behavior() {
}
return std::make_tuple(x, std::move(addr), port);
},
[=](tick_atom, size_t interval) {
instance.handle_heartbeat(context());
delayed_send(this, std::chrono::milliseconds{interval}, tick_atom::value,
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.send_heartbeats(context());
// Check whether any node reached the disconnect timeout.
if (connection_timeout.count() > 0) {
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.
scheduled_send(this, next_tick, tick_atom::value, next_tick,
heartbeat_interval, connection_timeout);
}};
}
......@@ -548,6 +591,7 @@ void basp_broker::learned_new_node_indirectly(const node_id& nid) {
void basp_broker::set_context(connection_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
auto now = clock().now();
auto i = ctx.find(hdl);
if (i == ctx.end()) {
CAF_LOG_DEBUG("create new BASP context:" << CAF_ARG(hdl));
......@@ -559,8 +603,10 @@ void basp_broker::set_context(connection_handle hdl) {
invalid_actor_id};
i = ctx
.emplace(hdl, basp::endpoint_context{basp::await_header, hdr, hdl,
node_id{}, 0, 0, none})
node_id{}, 0, 0, none, now})
.first;
} else {
i->second.last_seen = now;
}
this_context = &i->second;
t_last_hop = &i->second.id;
......
......@@ -693,6 +693,7 @@ public:
cfg.set("logger.inline-output", true);
cfg.set("middleman.network-backend", caf::atom("testing"));
cfg.set("middleman.manual-multiplexing", true);
cfg.set("middleman.heartbeat-interval", caf::timespan{0});
cfg.set("middleman.workers", size_t{0});
return cfg;
}
......
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