Commit 0b739718 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/pr-1096-integration'

parents cde2db78 d9399afc
...@@ -32,6 +32,14 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -32,6 +32,14 @@ is based on [Keep a Changelog](https://keepachangelog.com).
and accepted in config files as well as on the command line (#1238). and accepted in config files as well as on the command line (#1238).
- Silence a deprecated-enum-conversion warning for `std::byte` (#1230). - Silence a deprecated-enum-conversion warning for `std::byte` (#1230).
## [0.18.1] - 2021-03-19
### Fixed
- Version 0.18.0 introduced a regression on the system parameter
`caf.middleman.heartbeat-interval` (#1235). We have addressed the issue by
porting the original fix for CAF 0.17.5 (#1095) to the 0.18 series.
## [0.18.0] - 2021-01-25 ## [0.18.0] - 2021-01-25
### Added ### Added
......
...@@ -22,7 +22,7 @@ ...@@ -22,7 +22,7 @@
/// Denotes version of CAF in the format {MAJOR}{MINOR}{PATCH}, /// Denotes version of CAF in the format {MAJOR}{MINOR}{PATCH},
/// whereas each number is a two-digit decimal number without /// whereas each number is a two-digit decimal number without
/// leading zeros (e.g. 900 is version 0.9.0). /// leading zeros (e.g. 900 is version 0.9.0).
#define CAF_VERSION 1800 #define CAF_VERSION 1801
/// Defined to the major version number of CAF. /// Defined to the major version number of CAF.
#define CAF_MAJOR_VERSION (CAF_VERSION / 10000) #define CAF_MAJOR_VERSION (CAF_VERSION / 10000)
......
...@@ -113,7 +113,8 @@ namespace caf::defaults::middleman { ...@@ -113,7 +113,8 @@ namespace caf::defaults::middleman {
constexpr auto app_identifier = string_view{"generic-caf-app"}; constexpr auto app_identifier = string_view{"generic-caf-app"};
constexpr auto network_backend = string_view{"default"}; constexpr auto network_backend = string_view{"default"};
constexpr auto max_consecutive_reads = size_t{50}; constexpr auto max_consecutive_reads = size_t{50};
constexpr auto heartbeat_interval = size_t{0}; constexpr auto heartbeat_interval = timespan{10'000'000'000};
constexpr auto connection_timeout = timespan{30'000'000'000};
constexpr auto cached_udp_buffers = size_t{10}; constexpr auto cached_udp_buffers = size_t{10};
constexpr auto max_pending_msgs = size_t{10}; constexpr auto max_pending_msgs = size_t{10};
......
...@@ -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;
}; };
......
...@@ -6,14 +6,14 @@ ...@@ -6,14 +6,14 @@
#include <unordered_map> #include <unordered_map>
#include "caf/response_promise.hpp" #include "caf/actor_clock.hpp"
#include "caf/variant.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/datagram_handle.hpp"
#include "caf/io/basp/connection_state.hpp" #include "caf/io/basp/connection_state.hpp"
#include "caf/io/basp/header.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::io::basp { namespace caf::io::basp {
...@@ -32,6 +32,8 @@ struct endpoint_context { ...@@ -32,6 +32,8 @@ struct endpoint_context {
uint16_t local_port; uint16_t local_port;
// pending operations to be performed after handshake completed // pending operations to be performed after handshake completed
optional<response_promise> callback; optional<response_promise> callback;
// keeps track of when we've last received a message from this endpoint
actor_clock::time_point last_seen;
}; };
} // namespace caf::io::basp } // namespace caf::io::basp
...@@ -97,6 +97,7 @@ public: ...@@ -97,6 +97,7 @@ public:
void learned_new_node(const node_id& nid); void learned_new_node(const node_id& nid);
/// Sets `this_context` by either creating or accessing state for `hdl`. /// 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); void set_context(connection_handle hdl);
/// Cleans up any state for `hdl`. /// Cleans up any state for `hdl`.
......
...@@ -108,9 +108,19 @@ behavior basp_broker::make_behavior() { ...@@ -108,9 +108,19 @@ behavior basp_broker::make_behavior() {
} }
auto heartbeat_interval = get_or(config(), "caf.middleman.heartbeat-interval", auto heartbeat_interval = get_or(config(), "caf.middleman.heartbeat-interval",
defaults::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)); auto now = clock().now();
send(this, tick_atom_v, heartbeat_interval); auto first_tick = now + heartbeat_interval;
auto connection_timeout = get_or(config(),
"caf.middleman.connection-timeout",
defaults::middleman::connection_timeout);
CAF_LOG_DEBUG("enable heartbeat" << CAF_ARG(heartbeat_interval)
<< CAF_ARG(connection_timeout));
// Note: we send 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{ return behavior{
// received from underlying broker implementation // received from underlying broker implementation
...@@ -361,10 +371,50 @@ behavior basp_broker::make_behavior() { ...@@ -361,10 +371,50 @@ behavior basp_broker::make_behavior() {
} }
return {x, std::move(addr), port}; 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()); instance.handle_heartbeat(context());
delayed_send(this, std::chrono::milliseconds{interval}, tick_atom_v, // Check whether any node reached the disconnect timeout.
interval); 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_v,
next_tick.time_since_epoch().count(), heartbeat_interval,
connection_timeout);
}}; }};
} }
...@@ -587,6 +637,7 @@ void basp_broker::learned_new_node_indirectly(const node_id& nid) { ...@@ -587,6 +637,7 @@ void basp_broker::learned_new_node_indirectly(const node_id& nid) {
void basp_broker::set_context(connection_handle hdl) { void basp_broker::set_context(connection_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl)); CAF_LOG_TRACE(CAF_ARG(hdl));
auto now = clock().now();
auto i = ctx.find(hdl); auto i = ctx.find(hdl);
if (i == ctx.end()) { if (i == ctx.end()) {
CAF_LOG_DEBUG("create new BASP context:" << CAF_ARG(hdl)); CAF_LOG_DEBUG("create new BASP context:" << CAF_ARG(hdl));
...@@ -598,8 +649,10 @@ void basp_broker::set_context(connection_handle hdl) { ...@@ -598,8 +649,10 @@ void basp_broker::set_context(connection_handle hdl) {
invalid_actor_id}; invalid_actor_id};
i = ctx i = ctx
.emplace(hdl, basp::endpoint_context{basp::await_header, hdr, hdl, .emplace(hdl, basp::endpoint_context{basp::await_header, hdr, hdl,
node_id{}, 0, 0, none}) node_id{}, 0, 0, none, now})
.first; .first;
} else {
i->second.last_seen = now;
} }
this_context = &i->second; this_context = &i->second;
t_last_hop = &i->second.id; t_last_hop = &i->second.id;
......
...@@ -184,6 +184,9 @@ void middleman::add_module_options(actor_system_config& cfg) { ...@@ -184,6 +184,9 @@ void middleman::add_module_options(actor_system_config& cfg) {
.add<size_t>("max-consecutive-reads", .add<size_t>("max-consecutive-reads",
"max. number of consecutive reads per broker") "max. number of consecutive reads per broker")
.add<timespan>("heartbeat-interval", "interval of heartbeat messages") .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", .add<bool>("attach-utility-actors",
"schedule utility actors instead of dedicating threads") "schedule utility actors instead of dedicating threads")
.add<bool>("manual-multiplexing", .add<bool>("manual-multiplexing",
......
...@@ -93,6 +93,8 @@ public: ...@@ -93,6 +93,8 @@ public:
fixture(bool autoconn = false) fixture(bool autoconn = false)
: sys(cfg.load<io::middleman, network::test_multiplexer>() : sys(cfg.load<io::middleman, network::test_multiplexer>()
.set("caf.middleman.enable-automatic-connections", autoconn) .set("caf.middleman.enable-automatic-connections", autoconn)
.set("caf.middleman.heartbeat-interval", timespan{0})
.set("caf.middleman.connection-timeout", timespan{0})
.set("caf.middleman.workers", size_t{0}) .set("caf.middleman.workers", size_t{0})
.set("caf.scheduler.policy", autoconn ? "testing" : "stealing") .set("caf.scheduler.policy", autoconn ? "testing" : "stealing")
.set("caf.logger.inline-output", true) .set("caf.logger.inline-output", true)
......
...@@ -668,6 +668,7 @@ public: ...@@ -668,6 +668,7 @@ public:
cfg.set("caf.middleman.network-backend", "testing"); cfg.set("caf.middleman.network-backend", "testing");
cfg.set("caf.middleman.manual-multiplexing", true); cfg.set("caf.middleman.manual-multiplexing", true);
cfg.set("caf.middleman.workers", size_t{0}); cfg.set("caf.middleman.workers", size_t{0});
cfg.set("caf.middleman.heartbeat-interval", caf::timespan{0});
} }
cfg.set("caf.stream.credit-policy", "token-based"); cfg.set("caf.stream.credit-policy", "token-based");
cfg.set("caf.stream.token-based-policy.batch-size", 50); cfg.set("caf.stream.token-based-policy.batch-size", 50);
......
...@@ -126,7 +126,7 @@ with the developer blog checked out one level above, i.e.: ...@@ -126,7 +126,7 @@ with the developer blog checked out one level above, i.e.:
" "
if [ $(git rev-parse --abbrev-ref HEAD) != "master" ]; then if [ $(git rev-parse --abbrev-ref HEAD) != "master" ]; then
raise_error "not in master branch" ask_permission "not in master branch, continue on branch [y] or abort [n]?"
fi fi
# assumed files # assumed files
......
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