Unverified Commit 1d2b18e4 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #1071

Allow actors to monitor CAF nodes
parents 383f3c28 78fc80af
...@@ -37,6 +37,9 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -37,6 +37,9 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- The class `exit_msg` finally got its missing `operator==` (#1039). - The class `exit_msg` finally got its missing `operator==` (#1039).
- The class `node_id` received an overload for `parse` to allow users to convert - The class `node_id` received an overload for `parse` to allow users to convert
the output of `to_string` back to the original ID (#1058). the output of `to_string` back to the original ID (#1058).
- Actors can now `monitor` and `demonitor` CAF nodes (#1042). Monitoring a CAF
node causes the actor system to send a `node_down_msg` to the observer when
losing connection to the monitored node.
### Changed ### Changed
...@@ -82,6 +85,10 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -82,6 +85,10 @@ is based on [Keep a Changelog](https://keepachangelog.com).
deadlocks when calling blocking functions in message handlers. This function deadlocks when calling blocking functions in message handlers. This function
now behaves as expected (#1016). now behaves as expected (#1016).
- Exceptions while handling requests now trigger error messages (#1055). - Exceptions while handling requests now trigger error messages (#1055).
- The member function `demonitor` falsely refused typed actor handles. Actors
could monitor typed actors but not demonitoring it again. This member function
is now a template that accepts any actor handle in the same way `monitor`
already did.
## [0.17.5] - Unreleased ## [0.17.5] - Unreleased
......
...@@ -167,6 +167,22 @@ public: ...@@ -167,6 +167,22 @@ public:
using module_array = std::array<module_ptr, module::num_ids>; using module_array = std::array<module_ptr, module::num_ids>;
/// An (optional) component of the actor system with networking capabilities.
class CAF_CORE_EXPORT networking_module : public module {
public:
~networking_module() override;
/// Causes the module to send a `node_down_msg` to `observer` if this system
/// loses connection to `node`.
virtual void monitor(const node_id& node, const actor_addr& observer) = 0;
/// Causes the module remove one entry for `observer` from the list of
/// actors that receive a `node_down_msg` if this system loses connection to
/// `node`. Each call to `monitor` requires one call to `demonitor` in order
/// to unsubscribe the `observer` completely.
virtual void demonitor(const node_id& node, const actor_addr& observer) = 0;
};
/// @warning The system stores a reference to `cfg`, which means the /// @warning The system stores a reference to `cfg`, which means the
/// config object must outlive the actor system. /// config object must outlive the actor system.
explicit actor_system(actor_system_config& cfg); explicit actor_system(actor_system_config& cfg);
...@@ -279,6 +295,17 @@ public: ...@@ -279,6 +295,17 @@ public:
/// Blocks this caller until all actors are done. /// Blocks this caller until all actors are done.
void await_all_actors_done() const; void await_all_actors_done() const;
/// Send a `node_down_msg` to `observer` if this system loses connection to
/// `node`.
/// @note Calling this function *n* times causes the system to send
/// `node_down_msg` *n* times to the observer. In order to not receive
/// the messages, the observer must call `demonitor` *n* times.
void monitor(const node_id& node, const actor_addr& observer);
/// Removes `observer` from the list of actors that receive a `node_down_msg`
/// if this system loses connection to `node`.
void demonitor(const node_id& node, const actor_addr& observer);
/// Called by `spawn` when used to create a class-based actor to /// Called by `spawn` when used to create a class-based actor to
/// apply automatic conversions to `xs` before spawning the actor. /// apply automatic conversions to `xs` before spawning the actor.
/// Should not be called by users of the library directly. /// Should not be called by users of the library directly.
......
...@@ -151,19 +151,20 @@ class stateful_actor; ...@@ -151,19 +151,20 @@ class stateful_actor;
// -- structs ------------------------------------------------------------------ // -- structs ------------------------------------------------------------------
struct unit_t;
struct exit_msg;
struct down_msg; struct down_msg;
struct timeout_msg;
struct stream_slots;
struct upstream_msg;
struct group_down_msg;
struct downstream_msg; struct downstream_msg;
struct open_stream_msg; struct exit_msg;
struct invalid_actor_t; struct group_down_msg;
struct invalid_actor_addr_t;
struct illegal_message_element; struct illegal_message_element;
struct invalid_actor_addr_t;
struct invalid_actor_t;
struct node_down_msg;
struct open_stream_msg;
struct prohibit_top_level_spawn_marker; struct prohibit_top_level_spawn_marker;
struct stream_slots;
struct timeout_msg;
struct unit_t;
struct upstream_msg;
// -- free template functions -------------------------------------------------- // -- free template functions --------------------------------------------------
......
...@@ -288,6 +288,10 @@ public: ...@@ -288,6 +288,10 @@ public:
current_element_ = ptr; current_element_ = ptr;
} }
/// Adds a unidirectional `monitor` to `node`.
/// @note Each call to `monitor` creates a new, independent monitor.
void monitor(const node_id& node);
/// Adds a unidirectional `monitor` to `whom`. /// Adds a unidirectional `monitor` to `whom`.
/// @note Each call to `monitor` creates a new, independent monitor. /// @note Each call to `monitor` creates a new, independent monitor.
template <message_priority P = message_priority::normal, class Handle = actor> template <message_priority P = message_priority::normal, class Handle = actor>
...@@ -298,8 +302,12 @@ public: ...@@ -298,8 +302,12 @@ public:
/// Removes a monitor from `whom`. /// Removes a monitor from `whom`.
void demonitor(const actor_addr& whom); void demonitor(const actor_addr& whom);
/// Removes a monitor from `node`.
void demonitor(const node_id& node);
/// Removes a monitor from `whom`. /// Removes a monitor from `whom`.
inline void demonitor(const actor& whom) { template <class Handle>
void demonitor(const Handle& whom) {
demonitor(whom.address()); demonitor(whom.address());
} }
......
...@@ -200,6 +200,9 @@ public: ...@@ -200,6 +200,9 @@ public:
/// Function object for handling down messages. /// Function object for handling down messages.
using down_handler = std::function<void(pointer, down_msg&)>; using down_handler = std::function<void(pointer, down_msg&)>;
/// Function object for handling node down messages.
using node_down_handler = std::function<void(pointer, node_down_msg&)>;
/// Function object for handling exit messages. /// Function object for handling exit messages.
using exit_handler = std::function<void(pointer, exit_msg&)>; using exit_handler = std::function<void(pointer, exit_msg&)>;
...@@ -241,6 +244,8 @@ public: ...@@ -241,6 +244,8 @@ public:
static void default_down_handler(pointer ptr, down_msg& x); static void default_down_handler(pointer ptr, down_msg& x);
static void default_node_down_handler(pointer ptr, node_down_msg& x);
static void default_exit_handler(pointer ptr, exit_msg& x); static void default_exit_handler(pointer ptr, exit_msg& x);
#ifndef CAF_NO_EXCEPTIONS #ifndef CAF_NO_EXCEPTIONS
...@@ -378,6 +383,22 @@ public: ...@@ -378,6 +383,22 @@ public:
set_down_handler([fun](scheduled_actor*, down_msg& x) { fun(x); }); set_down_handler([fun](scheduled_actor*, down_msg& x) { fun(x); });
} }
/// Sets a custom handler for node down messages.
void set_node_down_handler(node_down_handler fun) {
if (fun)
node_down_handler_ = std::move(fun);
else
node_down_handler_ = default_node_down_handler;
}
/// Sets a custom handler for down messages.
template <class T>
auto set_node_down_handler(T fun)
-> decltype(fun(std::declval<node_down_msg&>())) {
set_node_down_handler(
[fun](scheduled_actor*, node_down_msg& x) { fun(x); });
}
/// Sets a custom handler for error messages. /// Sets a custom handler for error messages.
inline void set_exit_handler(exit_handler fun) { inline void set_exit_handler(exit_handler fun) {
if (fun) if (fun)
...@@ -884,6 +905,9 @@ protected: ...@@ -884,6 +905,9 @@ protected:
/// Customization point for setting a default `down_msg` callback. /// Customization point for setting a default `down_msg` callback.
down_handler down_handler_; down_handler down_handler_;
/// Customization point for setting a default `down_msg` callback.
node_down_handler node_down_handler_;
/// Customization point for setting a default `exit_msg` callback. /// Customization point for setting a default `exit_msg` callback.
exit_handler exit_handler_; exit_handler exit_handler_;
......
...@@ -90,6 +90,34 @@ typename Inspector::result_type inspect(Inspector& f, group_down_msg& x) { ...@@ -90,6 +90,34 @@ typename Inspector::result_type inspect(Inspector& f, group_down_msg& x) {
return f(meta::type_name("group_down_msg"), x.source); return f(meta::type_name("group_down_msg"), x.source);
} }
/// Sent to all actors monitoring a node when CAF loses connection to it.
struct node_down_msg {
/// The disconnected node.
node_id node;
/// The cause for the disconnection. No error (a default-constructed error
/// object) indicates an ordinary shutdown.
error reason;
};
/// @relates node_down_msg
inline bool operator==(const node_down_msg& x,
const node_down_msg& y) noexcept {
return x.node == y.node && x.reason == y.reason;
}
/// @relates node_down_msg
inline bool operator!=(const node_down_msg& x,
const node_down_msg& y) noexcept {
return !(x == y);
}
/// @relates node_down_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, node_down_msg& x) {
return f(meta::type_name("node_down_msg"), x.node, x.reason);
}
/// Signalizes a timeout event. /// Signalizes a timeout event.
/// @note This message is handled implicitly by the runtime system. /// @note This message is handled implicitly by the runtime system.
struct timeout_msg { struct timeout_msg {
......
...@@ -211,6 +211,10 @@ const char* actor_system::module::name() const noexcept { ...@@ -211,6 +211,10 @@ const char* actor_system::module::name() const noexcept {
} }
} }
actor_system::networking_module::~networking_module() {
// nop
}
actor_system::actor_system(actor_system_config& cfg) actor_system::actor_system(actor_system_config& cfg)
: profiler_(cfg.profiler), : profiler_(cfg.profiler),
ids_(0), ids_(0),
...@@ -412,6 +416,24 @@ void actor_system::await_all_actors_done() const { ...@@ -412,6 +416,24 @@ void actor_system::await_all_actors_done() const {
registry_.await_running_count_equal(0); registry_.await_running_count_equal(0);
} }
void actor_system::monitor(const node_id& node, const actor_addr& observer) {
// TODO: Currently does not work with other modules, in particular caf_net.
auto mm = modules_[module::middleman].get();
if (mm == nullptr)
return;
auto mm_dptr = static_cast<networking_module*>(mm);
mm_dptr->monitor(node, observer);
}
void actor_system::demonitor(const node_id& node, const actor_addr& observer) {
// TODO: Currently does not work with other modules, in particular caf_net.
auto mm = modules_[module::middleman].get();
if (mm == nullptr)
return;
auto mm_dptr = static_cast<networking_module*>(mm);
mm_dptr->demonitor(node, observer);
}
actor_clock& actor_system::clock() noexcept { actor_clock& actor_system::clock() noexcept {
return scheduler().clock(); return scheduler().clock();
} }
......
...@@ -74,6 +74,10 @@ void local_actor::monitor(abstract_actor* ptr, message_priority priority) { ...@@ -74,6 +74,10 @@ void local_actor::monitor(abstract_actor* ptr, message_priority priority) {
default_attachable::make_monitor(ptr->address(), address(), priority)); default_attachable::make_monitor(ptr->address(), address(), priority));
} }
void local_actor::monitor(const node_id& node) {
system().monitor(node, address());
}
void local_actor::demonitor(const actor_addr& whom) { void local_actor::demonitor(const actor_addr& whom) {
CAF_LOG_TRACE(CAF_ARG(whom)); CAF_LOG_TRACE(CAF_ARG(whom));
auto ptr = actor_cast<strong_actor_ptr>(whom); auto ptr = actor_cast<strong_actor_ptr>(whom);
...@@ -84,6 +88,10 @@ void local_actor::demonitor(const actor_addr& whom) { ...@@ -84,6 +88,10 @@ void local_actor::demonitor(const actor_addr& whom) {
} }
} }
void local_actor::demonitor(const node_id& node) {
system().demonitor(node, address());
}
void local_actor::on_exit() { void local_actor::on_exit() {
// nop // nop
} }
......
...@@ -86,6 +86,12 @@ void scheduled_actor::default_down_handler(scheduled_actor* ptr, down_msg& x) { ...@@ -86,6 +86,12 @@ void scheduled_actor::default_down_handler(scheduled_actor* ptr, down_msg& x) {
<< ", name: " << ptr->name() << "]: " << to_string(x) << std::endl; << ", name: " << ptr->name() << "]: " << to_string(x) << std::endl;
} }
void scheduled_actor::default_node_down_handler(scheduled_actor* ptr,
node_down_msg& x) {
aout(ptr) << "*** unhandled node down message [id: " << ptr->id()
<< ", name: " << ptr->name() << "]: " << to_string(x) << std::endl;
}
void scheduled_actor::default_exit_handler(scheduled_actor* ptr, exit_msg& x) { void scheduled_actor::default_exit_handler(scheduled_actor* ptr, exit_msg& x) {
if (x.reason) if (x.reason)
default_error_handler(ptr, x.reason); default_error_handler(ptr, x.reason);
...@@ -120,6 +126,7 @@ scheduled_actor::scheduled_actor(actor_config& cfg) ...@@ -120,6 +126,7 @@ scheduled_actor::scheduled_actor(actor_config& cfg)
default_handler_(print_and_drop), default_handler_(print_and_drop),
error_handler_(default_error_handler), error_handler_(default_error_handler),
down_handler_(default_down_handler), down_handler_(default_down_handler),
node_down_handler_(default_node_down_handler),
exit_handler_(default_exit_handler), exit_handler_(default_exit_handler),
private_thread_(nullptr) private_thread_(nullptr)
#ifndef CAF_NO_EXCEPTIONS #ifndef CAF_NO_EXCEPTIONS
...@@ -609,6 +616,11 @@ scheduled_actor::categorize(mailbox_element& x) { ...@@ -609,6 +616,11 @@ scheduled_actor::categorize(mailbox_element& x) {
call_handler(down_handler_, this, dm); call_handler(down_handler_, this, dm);
return message_category::internal; return message_category::internal;
} }
if (content.match_elements<node_down_msg>()) {
auto dm = content.move_if_unshared<node_down_msg>(0);
call_handler(node_down_handler_, this, dm);
return message_category::internal;
}
if (content.match_elements<error>()) { if (content.match_elements<error>()) {
auto err = content.move_if_unshared<error>(0); auto err = content.move_if_unshared<error>(0);
call_handler(error_handler_, this, err); call_handler(error_handler_, this, err);
......
...@@ -60,6 +60,7 @@ set(CAF_IO_TEST_SOURCES ...@@ -60,6 +60,7 @@ set(CAF_IO_TEST_SOURCES
test/io/basp_broker.cpp test/io/basp_broker.cpp
test/io/broker.cpp test/io/broker.cpp
test/io/http_broker.cpp test/io/http_broker.cpp
test/io/monitor.cpp
test/io/network/default_multiplexer.cpp test/io/network/default_multiplexer.cpp
test/io/network/ip_endpoint.cpp test/io/network/ip_endpoint.cpp
test/io/receive_buffer.cpp test/io/receive_buffer.cpp
......
...@@ -104,6 +104,9 @@ public: ...@@ -104,6 +104,9 @@ public:
// -- utility functions ------------------------------------------------------ // -- utility functions ------------------------------------------------------
/// Sends `node_down_msg` to all registered observers.
void emit_node_down_msg(const node_id& node, const error& reason);
/// Performs bookkeeping such as managing `spawn_servers`. /// Performs bookkeeping such as managing `spawn_servers`.
void learned_new_node(const node_id& nid); void learned_new_node(const node_id& nid);
...@@ -147,6 +150,10 @@ public: ...@@ -147,6 +150,10 @@ public:
/// 'SpawnServ' instance on the remote side. /// 'SpawnServ' instance on the remote side.
std::unordered_map<node_id, actor> spawn_servers; std::unordered_map<node_id, actor> spawn_servers;
/// Keeps track of actors that wish to receive a `node_down_msg` if a
/// particular node fails.
std::unordered_map<node_id, std::vector<actor_addr>> node_observers;
/// Configures whether BASP automatically open new connections to optimize /// Configures whether BASP automatically open new connections to optimize
/// routing paths by forming a mesh between all nodes. /// routing paths by forming a mesh between all nodes.
bool automatic_connections = false; bool automatic_connections = false;
......
...@@ -40,7 +40,7 @@ ...@@ -40,7 +40,7 @@
namespace caf::io { namespace caf::io {
/// Manages brokers and network backends. /// Manages brokers and network backends.
class CAF_IO_EXPORT middleman : public actor_system::module { class CAF_IO_EXPORT middleman : public actor_system::networking_module {
public: public:
friend class ::caf::actor_system; friend class ::caf::actor_system;
...@@ -195,6 +195,10 @@ public: ...@@ -195,6 +195,10 @@ public:
void* subtype_ptr() override; void* subtype_ptr() override;
void monitor(const node_id& node, const actor_addr& observer) override;
void demonitor(const node_id& node, const actor_addr& observer) override;
/// Spawns a new functor-based broker. /// Spawns a new functor-based broker.
template <spawn_options Os = no_spawn_options, template <spawn_options Os = no_spawn_options,
class F = std::function<void(broker*)>, class... Ts> class F = std::function<void(broker*)>, class... Ts>
......
...@@ -77,6 +77,13 @@ void basp_broker::on_exit() { ...@@ -77,6 +77,13 @@ void basp_broker::on_exit() {
// that the middleman calls this in its stop() function. However, // that the middleman calls this in its stop() function. However,
// ultimately we should find a nonblocking solution here. // ultimately we should find a nonblocking solution here.
instance.hub().await_workers(); instance.hub().await_workers();
// All nodes are offline now. We use a default-constructed error code to
// signal ordinary shutdown.
for (const auto& [node, observer_list] : node_observers)
for (const auto& observer : observer_list)
if (auto hdl = actor_cast<actor>(observer))
anon_send(hdl, node_down_msg{node, error{}});
node_observers.clear();
// Release any obsolete state. // Release any obsolete state.
ctx.clear(); ctx.clear();
// Make sure all spawn servers are down before clearing the container. // Make sure all spawn servers are down before clearing the container.
...@@ -200,6 +207,47 @@ behavior basp_broker::make_behavior() { ...@@ -200,6 +207,47 @@ behavior basp_broker::make_behavior() {
proxy->id()); proxy->id());
flush(hdl); flush(hdl);
}, },
// received from the middleman whenever a node becomes observed by a local
// actor
[=](monitor_atom, const node_id& node, const actor_addr& observer) {
// Sanity checks.
if (!observer || !node)
return;
// Add to the list if a list for this node already exists.
auto i = node_observers.find(node);
if (i != node_observers.end()) {
i->second.emplace_back(observer);
return;
}
// Check whether the node is still connected at the moment and send the
// observer a message immediately otherwise.
if (instance.tbl().lookup(node) == none) {
if (auto hdl = actor_cast<actor>(observer)) {
// TODO: we might want to consider keeping the exit reason for nodes,
// at least for some time. Otherwise, we'll have to send a
// generic "don't know" exit reason. Probably an improvement we
// should consider in caf_net.
anon_send(hdl, node_down_msg{node, sec::no_context});
}
return;
}
std::vector<actor_addr> xs{observer};
node_observers.emplace(node, std::move(xs));
},
// received from the middleman whenever a node becomes observed by a local
// actor
[=](demonitor_atom, const node_id& node, const actor_addr& observer) {
auto i = node_observers.find(node);
if (i == node_observers.end())
return;
auto& observers = i->second;
auto j = std::find(observers.begin(), observers.end(), observer);
if (j != observers.end()) {
observers.erase(j);
if (observers.empty())
node_observers.erase(i);
}
},
// received from underlying broker implementation // received from underlying broker implementation
[=](const new_connection_msg& msg) { [=](const new_connection_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
...@@ -460,6 +508,16 @@ void basp_broker::handle_down_msg(down_msg& dm) { ...@@ -460,6 +508,16 @@ void basp_broker::handle_down_msg(down_msg& dm) {
monitored_actors.erase(i); monitored_actors.erase(i);
} }
void basp_broker::emit_node_down_msg(const node_id& node, const error& reason) {
auto i = node_observers.find(node);
if (i == node_observers.end())
return;
for (const auto& observer : i->second)
if (auto hdl = actor_cast<actor>(observer))
anon_send(hdl, node_down_msg{node, reason});
node_observers.erase(i);
}
void basp_broker::learned_new_node(const node_id& nid) { void basp_broker::learned_new_node(const node_id& nid) {
CAF_LOG_TRACE(CAF_ARG(nid)); CAF_LOG_TRACE(CAF_ARG(nid));
if (spawn_servers.count(nid) > 0) { if (spawn_servers.count(nid) > 0) {
...@@ -565,10 +623,12 @@ void basp_broker::set_context(connection_handle hdl) { ...@@ -565,10 +623,12 @@ void basp_broker::set_context(connection_handle hdl) {
void basp_broker::connection_cleanup(connection_handle hdl, sec code) { void basp_broker::connection_cleanup(connection_handle hdl, sec code) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(code)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(code));
// Remove handle from the routing table and clean up any node-specific state // Remove handle from the routing table, notify all observers, and clean up
// we might still have. // any node-specific state we might still have.
if (auto nid = instance.tbl().erase_direct(hdl)) if (auto nid = instance.tbl().erase_direct(hdl)) {
emit_node_down_msg(nid, code);
purge_state(nid); purge_state(nid);
}
// Remove the context for `hdl`, making sure clients receive an error in case // Remove the context for `hdl`, making sure clients receive an error in case
// this connection was closed during handshake. // this connection was closed during handshake.
auto i = ctx.find(hdl); auto i = ctx.find(hdl);
......
...@@ -399,6 +399,16 @@ void* middleman::subtype_ptr() { ...@@ -399,6 +399,16 @@ void* middleman::subtype_ptr() {
return this; return this;
} }
void middleman::monitor(const node_id& node, const actor_addr& observer) {
auto basp = named_broker<basp_broker>("BASP");
anon_send(basp, monitor_atom_v, node, observer);
}
void middleman::demonitor(const node_id& node, const actor_addr& observer) {
auto basp = named_broker<basp_broker>("BASP");
anon_send(basp, demonitor_atom_v, node, observer);
}
middleman::~middleman() { middleman::~middleman() {
// nop // nop
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE io.monitor
#include "caf/io/middleman.hpp"
#include "caf/test/io_dsl.hpp"
using namespace caf;
namespace {
struct fixture : point_to_point_fixture<> {
actor observer;
fixture() {
mars_id = mars.sys.node();
}
// Connects earth and mars, storing the connection handles in earth_conn and
// mars_conn.
void connect() {
auto acc = next_accept_handle();
std::tie(earth_conn, mars_conn)
= prepare_connection(earth, mars, "localhost", 8080, acc);
CAF_CHECK_EQUAL(earth.publish(actor{earth.self}, 8080), 8080);
CAF_CHECK(mars.remote_actor("localhost", 8080));
}
void disconnect() {
anon_send(earth.bb, io::connection_closed_msg{earth_conn});
earth.handle_io_event();
anon_send(mars.bb, io::connection_closed_msg{mars_conn});
mars.handle_io_event();
}
node_id mars_id;
io::connection_handle earth_conn;
io::connection_handle mars_conn;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(monitor_tests, fixture)
CAF_TEST(disconnects cause node_down_msg) {
connect();
earth.self->monitor(mars_id);
run();
disconnect();
expect_on(earth, (node_down_msg),
to(earth.self).with(node_down_msg{mars_id, error{}}));
CAF_CHECK(earth.self->mailbox().empty());
}
CAF_TEST(node_down_msg calls the special node_down_handler) {
connect();
bool node_down_handler_called = false;
auto observer = earth.sys.spawn([&](event_based_actor* self) -> behavior {
self->monitor(mars_id);
self->set_node_down_handler([&](node_down_msg& dm) {
CAF_CHECK_EQUAL(dm.node, mars_id);
node_down_handler_called = true;
});
return [] {};
});
run();
disconnect();
expect_on(earth, (node_down_msg),
to(observer).with(node_down_msg{mars_id, error{}}));
CAF_CHECK(node_down_handler_called);
}
CAF_TEST(calling monitor n times produces n node_down_msg) {
connect();
for (int i = 0; i < 5; ++i)
earth.self->monitor(mars_id);
run();
disconnect();
for (int i = 0; i < 5; ++i)
expect_on(earth, (node_down_msg),
to(earth.self).with(node_down_msg{mars_id, error{}}));
CAF_CHECK_EQUAL(earth.self->mailbox().size(), 0u);
}
CAF_TEST(each demonitor only cancels one node_down_msg) {
connect();
for (int i = 0; i < 5; ++i)
earth.self->monitor(mars_id);
run();
earth.self->demonitor(mars_id);
run();
disconnect();
for (int i = 0; i < 4; ++i)
expect_on(earth, (node_down_msg),
to(earth.self).with(node_down_msg{mars_id, error{}}));
CAF_CHECK_EQUAL(earth.self->mailbox().size(), 0u);
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -72,7 +72,7 @@ public: ...@@ -72,7 +72,7 @@ public:
: mm(this->sys.middleman()), : mm(this->sys.middleman()),
mpx(dynamic_cast<caf::io::network::test_multiplexer&>(mm.backend())), mpx(dynamic_cast<caf::io::network::test_multiplexer&>(mm.backend())),
run_all_nodes(std::move(fun)) { run_all_nodes(std::move(fun)) {
// nop bb = mm.named_broker<caf::io::basp_broker>("BASP");
} }
test_node_fixture() : test_node_fixture([=] { this->run(); }) { test_node_fixture() : test_node_fixture([=] { this->run(); }) {
...@@ -109,6 +109,9 @@ public: ...@@ -109,6 +109,9 @@ public:
/// Reference to the middleman's event multiplexer. /// Reference to the middleman's event multiplexer.
caf::io::network::test_multiplexer& mpx; caf::io::network::test_multiplexer& mpx;
/// Handle to the BASP broker.
caf::actor bb;
/// Callback for triggering all nodes when simulating a network of CAF nodes. /// Callback for triggering all nodes when simulating a network of CAF nodes.
run_all_nodes_fun run_all_nodes; run_all_nodes_fun run_all_nodes;
...@@ -284,9 +287,13 @@ public: ...@@ -284,9 +287,13 @@ public:
}; };
#define expect_on(where, types, fields) \ #define expect_on(where, types, fields) \
CAF_MESSAGE(#where << ": expect" << #types << "." << #fields); \ do { \
expect_clause< CAF_EXPAND(CAF_DSL_LIST types) >{where . sched} . fields CAF_MESSAGE(#where << ": expect" << #types << "." << #fields); \
expect_clause<CAF_EXPAND(CAF_DSL_LIST types)>{where.sched}.fields; \
} while (false)
#define disallow_on(where, types, fields) \ #define disallow_on(where, types, fields) \
CAF_MESSAGE(#where << ": disallow" << #types << "." << #fields); \ do { \
disallow_clause< CAF_EXPAND(CAF_DSL_LIST types) >{where . sched} . fields CAF_MESSAGE(#where << ": disallow" << #types << "." << #fields); \
disallow_clause<CAF_EXPAND(CAF_DSL_LIST types)>{where.sched}.fields; \
} while (false)
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