Commit 87baa353 authored by Dominik Charousset's avatar Dominik Charousset

Integrate review feedback

parent 606b75da
...@@ -38,12 +38,19 @@ relaxed-sleep-duration=10ms ...@@ -38,12 +38,19 @@ relaxed-sleep-duration=10ms
; configures whether MMs try to span a full mesh ; configures whether MMs try to span a full mesh
enable-automatic-connections=false enable-automatic-connections=false
; application identifier of this node, prevents connection to other CAF ; application identifier of this node, prevents connection to other CAF
; instances with different identifier ; instances with incompatible identifiers (at least one entry must match)
app-identifier="" app-identifiers=["generic-caf-app"]
; maximum number of consecutive I/O reads per broker ; maximum number of consecutive I/O reads per broker
max-consecutive-reads=50 max-consecutive-reads=50
; heartbeat message interval in ms (0 disables heartbeating) ; heartbeat message interval for periodic traffic
heartbeat-interval=0ms ; (0 disables heartbeating - not recommended)
heartbeat-interval=10s
; force disconnects of CAF nodes after receiving no traffic for this amount of
; time (0 disables this feature - not recommended), be careful when deploying
; CAF applications with different heartbeat intervals and connection timeouts:
; this timeout should lign up with the *longest* heartbeat interval (ideally
; it's a multiple)
connection-timeout=30s
; configures whether the MM attaches its internal utility actors to the ; configures whether the MM attaches its internal utility actors to the
; scheduler instead of dedicating individual threads (needed only for ; scheduler instead of dedicating individual threads (needed only for
; deterministic testing) ; deterministic testing)
...@@ -52,10 +59,6 @@ attach-utility-actors=false ...@@ -52,10 +59,6 @@ attach-utility-actors=false
; setting this to true allows fully deterministic execution in unit test and ; setting this to true allows fully deterministic execution in unit test and
; requires the user to trigger I/O manually ; requires the user to trigger I/O manually
manual-multiplexing=false 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, ; configures how many background workers are spawned for deserialization,
; by default CAF uses 1-4 workers depending on the number of cores ; by default CAF uses 1-4 workers depending on the number of cores
workers=<min(3, number of cores / 4) + 1> workers=<min(3, number of cores / 4) + 1>
......
...@@ -115,8 +115,6 @@ actor_system_config::actor_system_config() ...@@ -115,8 +115,6 @@ actor_system_config::actor_system_config()
"excluded components for logging") "excluded components for logging")
.add<bool>("inline-output", "disable logger thread (for testing only!)"); .add<bool>("inline-output", "disable logger thread (for testing only!)");
opt_group{custom_options_, "middleman"} opt_group{custom_options_, "middleman"}
.add<atom_value>("network-backend",
"either 'default' or 'asio' (if available)")
.add<std::vector<string>>("app-identifiers", .add<std::vector<string>>("app-identifiers",
"valid application identifiers of this node") "valid application identifiers of this node")
.add<string>("app-identifier", "DEPRECATED: use app-identifiers instead") .add<string>("app-identifier", "DEPRECATED: use app-identifiers instead")
......
...@@ -122,8 +122,8 @@ public: ...@@ -122,8 +122,8 @@ public:
connection_state handle(execution_unit* ctx, connection_state handle(execution_unit* ctx,
new_data_msg& dm, header& hdr, bool is_payload); new_data_msg& dm, header& hdr, bool is_payload);
/// Sends heartbeat messages to all valid nodes those are directly connected. /// Sends heartbeat messages to all connected nodes.
void handle_heartbeat(execution_unit* ctx); void send_heartbeats(execution_unit* ctx);
/// Returns a route to `target` or `none` on error. /// Returns a route to `target` or `none` on error.
optional<routing_table::route> lookup(const node_id& target); optional<routing_table::route> lookup(const node_id& target);
......
...@@ -83,7 +83,7 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm, ...@@ -83,7 +83,7 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
return handle(ctx, dm.handle, hdr, payload); return handle(ctx, dm.handle, hdr, payload);
} }
void instance::handle_heartbeat(execution_unit* ctx) { void instance::send_heartbeats(execution_unit* ctx) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
for (auto& kvp : tbl_.direct_by_hdl_) { for (auto& kvp : tbl_.direct_by_hdl_) {
CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second)); CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second));
......
...@@ -356,11 +356,12 @@ behavior basp_broker::make_behavior() { ...@@ -356,11 +356,12 @@ behavior basp_broker::make_behavior() {
"Interval too low or BASP actor overloaded!"); "Interval too low or BASP actor overloaded!");
} }
// Send out heartbeats. // Send out heartbeats.
instance.handle_heartbeat(context()); instance.send_heartbeats(context());
// Check whether any node reached the disconnect timeout. // Check whether any node reached the disconnect timeout.
if (connection_timeout.count() > 0) {
for (auto i = ctx.begin(); i != ctx.end();) { for (auto i = ctx.begin(); i != ctx.end();) {
if (i->second.last_seen + connection_timeout < now) { if (i->second.last_seen + connection_timeout < now) {
CAF_LOG_WARNING("Disconnect BASP node: reached connection timeout!"); CAF_LOG_WARNING("Disconnect BASP node: reached connection timeout");
auto hdl = i->second.hdl; auto hdl = i->second.hdl;
// connection_cleanup below calls ctx.erase, so we need to increase // connection_cleanup below calls ctx.erase, so we need to increase
// the iterator now, before it gets invalidated. // the iterator now, before it gets invalidated.
...@@ -371,6 +372,7 @@ behavior basp_broker::make_behavior() { ...@@ -371,6 +372,7 @@ behavior basp_broker::make_behavior() {
++i; ++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