Commit ed4e4418 authored by Dominik Charousset's avatar Dominik Charousset

Organize hooks via actor system config

parent 975a8c1d
...@@ -43,6 +43,12 @@ namespace caf { ...@@ -43,6 +43,12 @@ namespace caf {
/// Configures an `actor_system` on startup. /// Configures an `actor_system` on startup.
class actor_system_config { class actor_system_config {
public: public:
// -- member types -----------------------------------------------------------
using hook_factory = std::function<io::hook* (actor_system&)>;
using hook_factory_vector = std::vector<hook_factory>;
using module_factory = std::function<actor_system::module* (actor_system&)>; using module_factory = std::function<actor_system::module* (actor_system&)>;
using module_factory_vector = std::vector<module_factory>; using module_factory_vector = std::vector<module_factory>;
...@@ -63,11 +69,13 @@ public: ...@@ -63,11 +69,13 @@ public:
using option_ptr = std::unique_ptr<config_option>; using option_ptr = std::unique_ptr<config_option>;
using options_vector = std::vector<option_ptr>; using option_vector = std::vector<option_ptr>;
// -- nested classes ---------------------------------------------------------
class opt_group { class opt_group {
public: public:
opt_group(options_vector& xs, const char* category); opt_group(option_vector& xs, const char* category);
template <class T> template <class T>
opt_group& add(T& storage, const char* name, const char* explanation) { opt_group& add(T& storage, const char* name, const char* explanation) {
...@@ -76,10 +84,12 @@ public: ...@@ -76,10 +84,12 @@ public:
} }
private: private:
options_vector& xs_; option_vector& xs_;
const char* cat_; const char* cat_;
}; };
// -- constructors, destructors, and assignment operators --------------------
virtual ~actor_system_config(); virtual ~actor_system_config();
actor_system_config(); actor_system_config();
...@@ -89,10 +99,20 @@ public: ...@@ -89,10 +99,20 @@ public:
actor_system_config(const actor_system_config&) = delete; actor_system_config(const actor_system_config&) = delete;
actor_system_config& operator=(const actor_system_config&) = delete; actor_system_config& operator=(const actor_system_config&) = delete;
// -- modifiers --------------------------------------------------------------
/// Parses `args` as tuple of strings containing CLI options
/// and `ini_stream` as INI formatted input stream.
actor_system_config& parse(message& args, std::istream& ini_stream); actor_system_config& parse(message& args, std::istream& ini_stream);
/// Parses the CLI options `{argc, argv}` and
/// `ini_stream` as INI formatted input stream.
actor_system_config& parse(int argc, char** argv, std::istream& ini_stream); actor_system_config& parse(int argc, char** argv, std::istream& ini_stream);
/// Parses the CLI options `{argc, argv}` and
/// tries to open `config_file_name` as INI formatted config file.
/// The parsers tries to open `caf-application.ini` if `config_file_name`
/// is `nullptr`.
actor_system_config& parse(int argc, char** argv, actor_system_config& parse(int argc, char** argv,
const char* config_file_name = nullptr); const char* config_file_name = nullptr);
...@@ -166,6 +186,21 @@ public: ...@@ -166,6 +186,21 @@ public:
return *this; return *this;
} }
/// Adds a factory for a new hook type to the middleman (if loaded).
template <class Factory>
actor_system_config& add_hook_factory(Factory f) {
hook_factories.push_back(f);
return *this;
}
/// Adds a hook type to the middleman (if loaded).
template <class Hook>
actor_system_config& add_hook_type() {
return add_hook_factory([](actor_system& sys) -> Hook* {
return new Hook(sys);
});
}
/// Stores whether the help text for this config object was /// Stores whether the help text for this config object was
/// printed. If set to `true`, the application should not use /// printed. If set to `true`, the application should not use
/// this config object to initialize an `actor_system` and /// this config object to initialize an `actor_system` and
...@@ -224,13 +259,14 @@ public: ...@@ -224,13 +259,14 @@ public:
actor_factory_map actor_factories; actor_factory_map actor_factories;
module_factory_vector module_factories; module_factory_vector module_factories;
error_renderer_map error_renderers; error_renderer_map error_renderers;
hook_factory_vector hook_factories;
int (*slave_mode_fun)(actor_system&, const actor_system_config&); int (*slave_mode_fun)(actor_system&, const actor_system_config&);
protected: protected:
virtual std::string make_help_text(const std::vector<message::cli_arg>&); virtual std::string make_help_text(const std::vector<message::cli_arg>&);
options_vector custom_options_; option_vector custom_options_;
private: private:
static std::string render_sec(uint8_t, atom_value, const message&); static std::string render_sec(uint8_t, atom_value, const message&);
...@@ -239,7 +275,7 @@ private: ...@@ -239,7 +275,7 @@ private:
std::string extract_config_file_name(message& args); std::string extract_config_file_name(message& args);
options_vector options_; option_vector options_;
}; };
} // namespace caf } // namespace caf
......
...@@ -116,6 +116,7 @@ ...@@ -116,6 +116,7 @@
# define CAF_DEPRECATED_MSG(msg) __attribute__((__deprecated__(msg))) # define CAF_DEPRECATED_MSG(msg) __attribute__((__deprecated__(msg)))
# define CAF_PUSH_WARNINGS # define CAF_PUSH_WARNINGS
_Pragma("GCC diagnostic push") \ _Pragma("GCC diagnostic push") \
_Pragma("GCC diagnostic ignored \"-Wc++14-extensions\"") \
_Pragma("GCC diagnostic ignored \"-Wfloat-equal\"") \ _Pragma("GCC diagnostic ignored \"-Wfloat-equal\"") \
_Pragma("GCC diagnostic ignored \"-Wconversion\"") \ _Pragma("GCC diagnostic ignored \"-Wconversion\"") \
_Pragma("GCC diagnostic ignored \"-Wcast-qual\"") \ _Pragma("GCC diagnostic ignored \"-Wcast-qual\"") \
......
...@@ -110,6 +110,7 @@ using actor_id = uint64_t; ...@@ -110,6 +110,7 @@ using actor_id = uint64_t;
namespace io { namespace io {
class hook;
class broker; class broker;
class middleman; class middleman;
......
...@@ -31,18 +31,18 @@ namespace caf { ...@@ -31,18 +31,18 @@ namespace caf {
namespace { namespace {
using options_vector = actor_system_config::options_vector; using option_vector = actor_system_config::option_vector;
class actor_system_config_reader { class actor_system_config_reader {
public: public:
using sink = std::function<void (size_t, config_value&)>; using sink = std::function<void (size_t, config_value&)>;
actor_system_config_reader(options_vector& xs, options_vector& ys) { actor_system_config_reader(option_vector& xs, option_vector& ys) {
add_opts(xs); add_opts(xs);
add_opts(ys); add_opts(ys);
} }
void add_opts(options_vector& xs) { void add_opts(option_vector& xs) {
for (auto& x : xs) for (auto& x : xs)
sinks_.emplace(x->full_name(), x->to_sink()); sinks_.emplace(x->full_name(), x->to_sink());
} }
...@@ -62,7 +62,7 @@ private: ...@@ -62,7 +62,7 @@ private:
} // namespace <anonymous> } // namespace <anonymous>
actor_system_config::opt_group::opt_group(options_vector& xs, actor_system_config::opt_group::opt_group(option_vector& xs,
const char* category) const char* category)
: xs_(xs), : xs_(xs),
cat_(category) { cat_(category) {
...@@ -276,7 +276,7 @@ actor_system_config& actor_system_config::parse(message& args, ...@@ -276,7 +276,7 @@ actor_system_config& actor_system_config::parse(message& args,
if (res.opts.count("caf#dump-config")) { if (res.opts.count("caf#dump-config")) {
cli_helptext_printed = true; cli_helptext_printed = true;
std::string category; std::string category;
options_vector* all_options[] = { &options_, &custom_options_ }; option_vector* all_options[] = { &options_, &custom_options_ };
for (auto& opt_vec : all_options) { for (auto& opt_vec : all_options) {
for (auto& opt : *opt_vec) { for (auto& opt : *opt_vec) {
if (category != opt->category()) { if (category != opt->category()) {
......
...@@ -44,6 +44,8 @@ using hook_uptr = std::unique_ptr<hook>; ...@@ -44,6 +44,8 @@ using hook_uptr = std::unique_ptr<hook>;
/// Interface to define hooks into the IO layer. /// Interface to define hooks into the IO layer.
class hook { class hook {
public: public:
hook(actor_system& sys);
virtual ~hook(); virtual ~hook();
/// Called whenever a message has arrived via the network. /// Called whenever a message has arrived via the network.
...@@ -138,22 +140,15 @@ public: ...@@ -138,22 +140,15 @@ public:
dispatch(event<Event>{}, std::forward<Ts>(ts)...); dispatch(event<Event>{}, std::forward<Ts>(ts)...);
} }
/// Forwards an event to the next hook. inline actor_system& system() const {
template <event_type Event, typename... Ts> return system_;
void call_next(Ts&&... ts) {
if (next) {
next->dispatch(event<Event>{}, std::forward<Ts>(ts)...);
}
} }
/// Intrusive pointer to the next hook. Hooks are stored as a simple,
/// singly linked list.
hook_uptr next;
private: private:
// ------------ convenience interface based on static dispatching ------------ // ------------ convenience interface based on static dispatching ------------
template <event_type Id> template <event_type Id>
using event = std::integral_constant<event_type, Id>; using event = std::integral_constant<event_type, Id>;
CAF_IO_HOOK_DISPATCH(message_received) CAF_IO_HOOK_DISPATCH(message_received)
CAF_IO_HOOK_DISPATCH(message_sent) CAF_IO_HOOK_DISPATCH(message_sent)
CAF_IO_HOOK_DISPATCH(message_forwarded) CAF_IO_HOOK_DISPATCH(message_forwarded)
...@@ -167,6 +162,8 @@ private: ...@@ -167,6 +162,8 @@ private:
CAF_IO_HOOK_DISPATCH(route_lost) CAF_IO_HOOK_DISPATCH(route_lost)
CAF_IO_HOOK_DISPATCH(invalid_message_received) CAF_IO_HOOK_DISPATCH(invalid_message_received)
CAF_IO_HOOK_DISPATCH(before_shutdown) CAF_IO_HOOK_DISPATCH(before_shutdown)
actor_system& system_;
}; };
} // namespace io } // namespace io
......
...@@ -43,6 +43,8 @@ class middleman : public actor_system::module { ...@@ -43,6 +43,8 @@ class middleman : public actor_system::module {
public: public:
friend class actor_system; friend class actor_system;
using hook_vector = std::vector<hook_uptr>;
~middleman(); ~middleman();
/// Publishes `whom` at `port`. /// Publishes `whom` at `port`.
...@@ -140,44 +142,18 @@ public: ...@@ -140,44 +142,18 @@ public:
/// Invokes the callback(s) associated with given event. /// Invokes the callback(s) associated with given event.
template <hook::event_type Event, typename... Ts> template <hook::event_type Event, typename... Ts>
void notify(Ts&&... ts) { void notify(Ts&&... ts) {
if (hooks_) for (auto& hook : hooks_)
hooks_->handle<Event>(std::forward<Ts>(ts)...); hook->handle<Event>(std::forward<Ts>(ts)...);
}
/// Adds a new hook to the middleman.
template<class C, typename... Ts>
C* add_hook(Ts&&... xs) {
// if only we could move a unique_ptr into a lambda in C++11
auto ptr = new C(system_, std::forward<Ts>(xs)...);
backend().dispatch([=] {
ptr->next.swap(hooks_);
hooks_.reset(ptr);
});
return ptr;
} }
/// Returns whether this middleman has any hooks installed. /// Returns whether this middleman has any hooks installed.
inline bool has_hook() const { inline bool has_hook() const {
return hooks_ != nullptr; return ! hooks_.empty();
} }
/// Adds a function object to this middleman that is called /// Returns all installed hooks.
/// when shutting down this middleman. const hook_vector& hooks() const {
template <class F> return hooks_;
void add_shutdown_cb(F fun) {
struct impl : hook {
impl(actor_system&, F&& f) : fun_(std::move(f)) {
// nop
}
void before_shutdown_cb() override {
fun_();
call_next<hook::before_shutdown>();
}
F fun_;
};
add_hook<impl>(std::move(fun));
} }
/// Returns the actor associated with `name` at `nid` or /// Returns the actor associated with `name` at `nid` or
...@@ -256,17 +232,6 @@ public: ...@@ -256,17 +232,6 @@ public:
return new impl(sys); return new impl(sys);
} }
/// Returns the heartbeat interval in milliseconds.
inline size_t heartbeat_interval() const {
return heartbeat_interval_;
}
/// Returns whether the middleman tries to establish
/// a direct connection to each of its peers.
inline bool enable_automatic_connections() const {
return enable_automatic_connections_;
}
protected: protected:
middleman(actor_system& ref); middleman(actor_system& ref);
...@@ -318,13 +283,9 @@ private: ...@@ -318,13 +283,9 @@ private:
// keeps track of "singleton-like" brokers // keeps track of "singleton-like" brokers
std::map<atom_value, actor> named_brokers_; std::map<atom_value, actor> named_brokers_;
// user-defined hooks // user-defined hooks
hook_uptr hooks_; hook_vector hooks_;
// actor offering asyncronous IO by managing this singleton instance // actor offering asyncronous IO by managing this singleton instance
middleman_actor manager_; middleman_actor manager_;
// heartbeat interval of BASP in milliseconds
size_t heartbeat_interval_;
// configures whether BASP tries to connect to all known peers
bool enable_automatic_connections_;
}; };
} // namespace io } // namespace io
......
...@@ -28,6 +28,7 @@ ...@@ -28,6 +28,7 @@
#include "caf/exception.hpp" #include "caf/exception.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/forwarding_actor_proxy.hpp" #include "caf/forwarding_actor_proxy.hpp"
#include "caf/actor_registry.hpp" #include "caf/actor_registry.hpp"
...@@ -441,7 +442,7 @@ basp_broker::basp_broker(actor_config& cfg) ...@@ -441,7 +442,7 @@ basp_broker::basp_broker(actor_config& cfg)
behavior basp_broker::make_behavior() { behavior basp_broker::make_behavior() {
CAF_LOG_TRACE(CAF_ARG(system().node())); CAF_LOG_TRACE(CAF_ARG(system().node()));
if (system().middleman().enable_automatic_connections()) { if (system().config().middleman_enable_automatic_connections) {
CAF_LOG_INFO("enable automatic connections"); CAF_LOG_INFO("enable automatic connections");
// open a random port and store a record for our peers how to // open a random port and store a record for our peers how to
// connect to this broker directly in the configuration server // connect to this broker directly in the configuration server
...@@ -453,7 +454,7 @@ behavior basp_broker::make_behavior() { ...@@ -453,7 +454,7 @@ behavior basp_broker::make_behavior() {
make_message(port.second, std::move(addrs))); make_message(port.second, std::move(addrs)));
state.enable_automatic_connections = true; state.enable_automatic_connections = true;
} }
auto heartbeat_interval = system().middleman().heartbeat_interval(); auto heartbeat_interval = system().config().middleman_heartbeat_interval;
if (heartbeat_interval > 0) { if (heartbeat_interval > 0) {
CAF_LOG_INFO("enable heartbeat" << CAF_ARG(heartbeat_interval)); CAF_LOG_INFO("enable heartbeat" << CAF_ARG(heartbeat_interval));
send(this, tick_atom::value, heartbeat_interval); send(this, tick_atom::value, heartbeat_interval);
......
...@@ -24,76 +24,73 @@ ...@@ -24,76 +24,73 @@
namespace caf { namespace caf {
namespace io { namespace io {
hook::hook(actor_system& sys) : system_(sys) {
// nop
}
hook::~hook() { hook::~hook() {
// nop // nop
} }
void hook::message_received_cb(const node_id& source, void hook::message_received_cb(const node_id&, const strong_actor_ptr&,
const strong_actor_ptr& from, const strong_actor_ptr&, message_id,
const strong_actor_ptr& dest, message_id mid, const message&) {
const message& msg) { // nop
call_next<message_received>(source, from, dest, mid, msg);
} }
void hook::message_sent_cb(const strong_actor_ptr& from, const node_id& dest_node, void hook::message_sent_cb(const strong_actor_ptr&, const node_id&,
const strong_actor_ptr& dest, message_id mid, const strong_actor_ptr&, message_id,
const message& payload) { const message&) {
call_next<message_sent>(from, dest_node, dest, mid, payload); // nop
} }
void hook::message_forwarded_cb(const basp::header& hdr, void hook::message_forwarded_cb(const basp::header&, const std::vector<char>*) {
const std::vector<char>* payload) { // nop
call_next<message_forwarded>(hdr, payload);
} }
void hook::message_forwarding_failed_cb(const basp::header& hdr, void hook::message_forwarding_failed_cb(const basp::header&,
const std::vector<char>* payload) { const std::vector<char>*) {
call_next<message_forwarding_failed>(hdr, payload); // nop
} }
void hook::message_sending_failed_cb(const strong_actor_ptr& from, void hook::message_sending_failed_cb(const strong_actor_ptr&,
const strong_actor_ptr& dest, const strong_actor_ptr&,
message_id mid, message_id, const message&) {
const message& payload) { // nop
call_next<message_sending_failed>(from, dest, mid, payload);
} }
void hook::actor_published_cb(const strong_actor_ptr& addr, void hook::actor_published_cb(const strong_actor_ptr&,
const std::set<std::string>& ifs, const std::set<std::string>&, uint16_t) {
uint16_t port) { // nop
call_next<actor_published>(addr, ifs, port);
} }
void hook::new_remote_actor_cb(const strong_actor_ptr& addr) { void hook::new_remote_actor_cb(const strong_actor_ptr&) {
call_next<new_remote_actor>(addr); // nop
} }
void hook::new_connection_established_cb(const node_id& node) { void hook::new_connection_established_cb(const node_id&) {
call_next<new_connection_established>(node); // nop
} }
void hook::new_route_added_cb(const node_id& via, const node_id& node) { void hook::new_route_added_cb(const node_id&, const node_id&) {
call_next<new_route_added>(via, node); // nop
} }
void hook::connection_lost_cb(const node_id& dest) { void hook::connection_lost_cb(const node_id&) {
call_next<connection_lost>(dest); // nop
} }
void hook::route_lost_cb(const node_id& hop, const node_id& dest) { void hook::route_lost_cb(const node_id&, const node_id&) {
call_next<route_lost>(hop, dest); // nop
} }
void hook::invalid_message_received_cb(const node_id& source, void hook::invalid_message_received_cb(const node_id&, const strong_actor_ptr&,
const strong_actor_ptr& sender, actor_id, message_id, const message&) {
actor_id invalid_dest, // nop
message_id mid,
const message& msg) {
call_next<invalid_message_received>(source, sender, invalid_dest, mid, msg);
} }
void hook::before_shutdown_cb() { void hook::before_shutdown_cb() {
call_next<before_shutdown>(); // nop
} }
} // namespace io } // namespace io
......
...@@ -103,9 +103,7 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) { ...@@ -103,9 +103,7 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
middleman::middleman(actor_system& sys) middleman::middleman(actor_system& sys)
: system_(sys), : system_(sys),
manager_(unsafe_actor_handle_init), manager_(unsafe_actor_handle_init) {
heartbeat_interval_(0),
enable_automatic_connections_(false) {
// nop // nop
} }
...@@ -155,9 +153,9 @@ uint16_t middleman::publish_local_groups(uint16_t port, const char* in) { ...@@ -155,9 +153,9 @@ uint16_t middleman::publish_local_groups(uint16_t port, const char* in) {
auto gn = system().spawn<hidden>(group_nameserver); auto gn = system().spawn<hidden>(group_nameserver);
try { try {
auto result = publish(gn, port, in); auto result = publish(gn, port, in);
add_shutdown_cb([gn] { // link gn to our manager
anon_send_exit(gn, exit_reason::user_shutdown); manager_->link_impl(abstract_actor::establish_link_op,
}); actor_cast<abstract_actor*>(gn));
return result; return result;
} }
catch (std::exception&) { catch (std::exception&) {
...@@ -272,6 +270,10 @@ strong_actor_ptr middleman::remote_lookup(atom_value name, const node_id& nid) { ...@@ -272,6 +270,10 @@ strong_actor_ptr middleman::remote_lookup(atom_value name, const node_id& nid) {
void middleman::start() { void middleman::start() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// create hooks
for (auto& f : system().config().hook_factories)
hooks_.emplace_back(f(system_));
// launch backend
backend_supervisor_ = backend().make_supervisor(); backend_supervisor_ = backend().make_supervisor();
if (! backend_supervisor_) { if (! backend_supervisor_) {
// the only backend that returns a `nullptr` is the `test_multiplexer` // the only backend that returns a `nullptr` is the `test_multiplexer`
...@@ -309,10 +311,10 @@ void middleman::stop() { ...@@ -309,10 +311,10 @@ void middleman::stop() {
backend_supervisor_.reset(); backend_supervisor_.reset();
if (thread_.joinable()) if (thread_.joinable())
thread_.join(); thread_.join();
hooks_.reset(); hooks_.clear();
named_brokers_.clear(); named_brokers_.clear();
scoped_actor self{system(), true}; scoped_actor self{system(), true};
self->send_exit(manager_, exit_reason::user_shutdown); self->send_exit(manager_, exit_reason::kill);
self->wait_for(manager_); self->wait_for(manager_);
destroy(manager_); destroy(manager_);
} }
...@@ -338,10 +340,7 @@ void middleman::init(actor_system_config& cfg) { ...@@ -338,10 +340,7 @@ void middleman::init(actor_system_config& cfg) {
// set scheduling parameters for multiplexer // set scheduling parameters for multiplexer
backend().max_throughput(cfg.scheduler_max_throughput); backend().max_throughput(cfg.scheduler_max_throughput);
backend().max_consecutive_reads(cfg.middleman_max_consecutive_reads); backend().max_consecutive_reads(cfg.middleman_max_consecutive_reads);
// set options relevant to BASP // give config access to slave mode implementation
heartbeat_interval_ = cfg.middleman_heartbeat_interval;
enable_automatic_connections_ = cfg.middleman_enable_automatic_connections;
// enable slave mode
cfg.slave_mode_fun = &middleman::exec_slave_mode; cfg.slave_mode_fun = &middleman::exec_slave_mode;
} }
......
...@@ -57,6 +57,10 @@ public: ...@@ -57,6 +57,10 @@ public:
++i; ++i;
} }
}); });
set_exit_handler([=](exit_msg&) {
// ignored, the MM links group nameservers
// to this actor for proper shutdown ordering
});
} }
void on_exit() override { void on_exit() override {
......
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