Commit ab865a9c authored by Dominik Charousset's avatar Dominik Charousset

Store config in the actor system

parent 7e2fcfa1
......@@ -165,6 +165,8 @@ public:
actor_system(int argc, char** argv);
/// @warning The system stores a reference to `cfg`, which means the
/// config object must outlive the actor system.
explicit actor_system(actor_system_config& cfg);
explicit actor_system(actor_system_config&& cfg);
......@@ -409,6 +411,11 @@ public:
await_actors_before_shutdown_ = x;
}
/// Returns the configuration of this actor system.
const actor_system_config& config() const {
return *cfg_;
}
/// @cond PRIVATE
/// Increases running-detached-threads-count by one.
......@@ -420,13 +427,11 @@ public:
/// Blocks the caller until all detached threads are done.
void await_detached_threads();
inline atom_value backend_name() const {
return backend_name_;
}
/// @endcond
private:
actor_system(actor_system_config* cfg, bool owns_cfg);
template <class T>
void check_invariants() {
static_assert(! std::is_base_of<prohibit_top_level_spawn_marker, T>::value,
......@@ -465,7 +470,6 @@ private:
module_array modules_;
io::middleman* middleman_;
scoped_execution_unit dummy_execution_unit_;
atom_value backend_name_;
opencl::manager* opencl_manager_;
riac::probe* probe_;
bool await_actors_before_shutdown_;
......@@ -475,6 +479,8 @@ private:
mutable std::mutex detached_mtx;
mutable std::condition_variable detached_cv;
actor_system_config* cfg_;
bool owns_cfg_;
};
} // namespace caf
......
......@@ -43,25 +43,23 @@ namespace caf {
/// Configures an `actor_system` on startup.
class actor_system_config {
public:
friend class actor_system;
using module_factory = std::function<actor_system::module* (actor_system&)>;
using module_factories = std::vector<module_factory>;
using module_factory_vector = std::vector<module_factory>;
using value_factory = std::function<type_erased_value_ptr ()>;
using value_factories_by_name = std::unordered_map<std::string, value_factory>;
using value_factory_string_map = std::unordered_map<std::string, value_factory>;
using value_factories_by_rtti = std::unordered_map<std::type_index, value_factory>;
using value_factory_rtti_map = std::unordered_map<std::type_index, value_factory>;
using actor_factories = std::unordered_map<std::string, actor_factory>;
using actor_factory_map = std::unordered_map<std::string, actor_factory>;
using portable_names = std::unordered_map<std::type_index, std::string>;
using portable_name_map = std::unordered_map<std::type_index, std::string>;
using error_renderer = std::function<std::string (uint8_t, atom_value, const message&)>;
using error_renderers = std::unordered_map<atom_value, error_renderer>;
using error_renderer_map = std::unordered_map<atom_value, error_renderer>;
using option_ptr = std::unique_ptr<config_option>;
......@@ -86,9 +84,15 @@ public:
actor_system_config();
actor_system_config(actor_system_config&&) = default;
actor_system_config(const actor_system_config&) = delete;
actor_system_config& operator=(const actor_system_config&) = delete;
actor_system_config& parse(message& args, std::istream& ini_stream);
actor_system_config& parse(int argc, char** argv, std::istream& ini_stream);
actor_system_config& parse(int argc, char** argv,
const char* config_file_name = nullptr);
......@@ -122,9 +126,9 @@ public:
&& std::is_copy_constructible<T>::value),
"T must provide default and copy constructors");
static_assert(detail::is_serializable<T>::value, "T must be serializable");
type_names_by_rtti_.emplace(std::type_index(typeid(T)), name);
value_factories_by_name_.emplace(std::move(name), &make_type_erased_value<T>);
value_factories_by_rtti_.emplace(std::type_index(typeid(T)),
type_names_by_rtti.emplace(std::type_index(typeid(T)), name);
value_factories_by_name.emplace(std::move(name), &make_type_erased_value<T>);
value_factories_by_rtti.emplace(std::type_index(typeid(T)),
&make_type_erased_value<T>);
return *this;
}
......@@ -156,7 +160,7 @@ public:
/// Loads module `T` with optional template parameters `Ts...`.
template <class T, class... Ts>
actor_system_config& load() {
module_factories_.push_back([](actor_system& sys) -> actor_system::module* {
module_factories.push_back([](actor_system& sys) -> actor_system::module* {
return T::make(sys, detail::type_list<Ts...>{});
});
return *this;
......@@ -205,9 +209,12 @@ public:
// Config parameters of the OpenCL module.
std::string opencl_device_ids;
// System parameters that are set while initializing modules.
node_id network_id;
proxy_registry* network_proxies;
value_factory_string_map value_factories_by_name;
value_factory_rtti_map value_factories_by_rtti;
portable_name_map type_names_by_rtti;
actor_factory_map actor_factories;
module_factory_vector module_factories;
error_renderer_map error_renderers;
int (*slave_mode_fun)(actor_system&, const actor_system_config&);
......@@ -221,12 +228,8 @@ private:
static std::string render_exit_reason(uint8_t, atom_value, const message&);
value_factories_by_name value_factories_by_name_;
value_factories_by_rtti value_factories_by_rtti_;
portable_names type_names_by_rtti_;
actor_factories actor_factories_;
module_factories module_factories_;
error_renderers error_renderers_;
std::string extract_config_file_name(message& args);
options_vector options_;
};
......
......@@ -104,20 +104,11 @@ private:
// message types
std::array<value_factory_kvp, type_nrs - 1> builtin_;
value_factories_by_name custom_by_name_;
value_factories_by_rtti custom_by_rtti_;
value_factories_by_name ad_hoc_;
mutable detail::shared_spinlock ad_hoc_mtx_;
// message type names
std::array<std::string, type_nrs - 1> builtin_names_;
portable_names custom_names_;
// actor types
actor_factories factories_;
// error types
error_renderers error_renderers_;
};
} // namespace caf
......
......@@ -168,21 +168,33 @@ actor_system::module::~module() {
// nop
}
actor_system::actor_system() : actor_system(actor_system_config{}) {
actor_system::actor_system(actor_system_config&& cfg)
: actor_system(new actor_system_config(std::move(cfg)), true) {
// nop
}
actor_system::actor_system()
: actor_system(new actor_system_config, true) {
// nop
}
actor_system_config* new_parsed_config(int argc, char** argv) {
auto ptr = new actor_system_config;
ptr->parse(argc, argv);
return ptr;
}
actor_system::actor_system(int argc, char** argv)
: actor_system(actor_system_config{}.parse(argc, argv)) {
: actor_system(new_parsed_config(argc, argv), true) {
// nop
}
actor_system::actor_system(actor_system_config& cfg)
: actor_system(std::move(cfg)) {
: actor_system(&cfg, false) {
// nop
}
actor_system::actor_system(actor_system_config&& cfg)
actor_system::actor_system(actor_system_config* ptr, bool owns_ptr)
: ids_(0),
types_(*this),
logger_(*this),
......@@ -191,12 +203,15 @@ actor_system::actor_system(actor_system_config&& cfg)
middleman_(nullptr),
dummy_execution_unit_(this),
await_actors_before_shutdown_(true),
detached(0) {
detached(0),
cfg_(ptr),
owns_cfg_(owns_ptr) {
CAF_ASSERT(ptr != nullptr);
CAF_SET_LOGGER_SYS(this);
backend_name_ = cfg.middleman_network_backend;
for (auto& f : cfg.module_factories_) {
auto ptr = f(*this);
modules_[ptr->id()].reset(ptr);
auto& cfg = *ptr;
for (auto& f : cfg.module_factories) {
auto mod_ptr = f(*this);
modules_[mod_ptr->id()].reset(mod_ptr);
}
auto& mmptr = modules_[module::middleman];
if (mmptr)
......@@ -251,14 +266,6 @@ actor_system::actor_system(actor_system_config&& cfg)
for (auto& mod : modules_)
if (mod)
mod->init(cfg);
// move all custom factories into our map
types_.custom_names_ = std::move(cfg.type_names_by_rtti_);
types_.custom_by_name_ = std::move(cfg.value_factories_by_name_);
types_.custom_by_rtti_ = std::move(cfg.value_factories_by_rtti_);
types_.factories_ = std::move(cfg.actor_factories_);
types_.error_renderers_ = std::move(cfg.error_renderers_);
// move remaining config
node_.swap(cfg.network_id);
// spawn config and spawn servers (lazily to not access the scheduler yet)
static constexpr auto Flags = hidden + lazy_init;
spawn_serv_ = actor_cast<strong_actor_ptr>(spawn<Flags>(spawn_serv_impl));
......@@ -294,6 +301,8 @@ actor_system::~actor_system() {
await_detached_threads();
registry_.stop();
logger_.stop();
if (owns_cfg_)
delete cfg_;
CAF_SET_LOGGER_SYS(nullptr);
}
......
......@@ -49,11 +49,11 @@ public:
void operator()(size_t ln, std::string name, config_value& cv) {
auto i = sinks_.find(name);
if (i == sinks_.end())
if (i != sinks_.end())
(i->second)(ln, cv);
else
std::cerr << "error in line " << ln
<< ": unrecognized parameter name \"" << name << "\"";
else
(i->second)(ln, cv);
}
private:
......@@ -124,8 +124,8 @@ actor_system_config::actor_system_config()
.add(opencl_device_ids, "device-ids",
"restricts which OpenCL devices are accessed by CAF");
// add renderers for default error categories
error_renderers_.emplace(atom("system"), render_sec);
error_renderers_.emplace(atom("exit"), render_exit_reason);
error_renderers.emplace(atom("system"), render_sec);
error_renderers.emplace(atom("exit"), render_exit_reason);
}
std::string
......@@ -165,29 +165,46 @@ actor_system_config::make_help_text(const std::vector<message::cli_arg>& xs) {
actor_system_config& actor_system_config::parse(int argc, char** argv,
const char* ini_file_cstr) {
auto args = message_builder(argv + 1, argv + argc).move_to_message();
// extract config file name first, since INI files are overruled by CLI args
message args;
if (argc > 1)
args = message_builder(argv + 1, argv + argc).move_to_message();
// set default config file name if not set by user
if (! ini_file_cstr)
ini_file_cstr = "caf-application.ini";
std::string config_file_name;
if (ini_file_cstr)
config_file_name = ini_file_cstr;
// CLI file name has priority over default file name
args.extract_opts({
{"caf#config-file", "", config_file_name}
});
// (2) content of the INI file overrides hard-coded defaults
if (! config_file_name.empty()) {
if (config_file_name.empty())
config_file_name = ini_file_cstr;
std::ifstream ini{config_file_name};
return parse(args, ini);
}
actor_system_config& actor_system_config::parse(int argc, char** argv,
std::istream& ini) {
message args;
if (argc > 1)
args = message_builder(argv + 1, argv + argc).move_to_message();
return parse(args, ini);
}
actor_system_config& actor_system_config::parse(message& args,
std::istream& ini) {
// (2) content of the INI file overrides hard-coded defaults
if (ini.good()) {
actor_system_config_reader consumer{options_, custom_options_};
detail::parse_ini(ini, consumer, std::cerr);
}
}
// (3) CLI options override the content of the INI file
std::string dummy; // caf#config-file either ignored or already open
std::vector<message::cli_arg> cargs;
for (auto& x : options_)
cargs.emplace_back(x->to_cli_arg(true));
cargs.emplace_back("caf#dump-config", "print config in INI format to stdout");
//cargs.emplace_back("caf#help", "print this text");
cargs.emplace_back("caf#config-file", "parse INI file", config_file_name);
cargs.emplace_back("caf#config-file", "parse INI file", dummy);
cargs.emplace_back("caf#slave-mode", "run in slave mode");
cargs.emplace_back("caf#slave-name", "set name for this slave", slave_name);
cargs.emplace_back("caf#bootstrap-node", "set bootstrapping", bootstrap_node);
......@@ -253,13 +270,13 @@ actor_system_config& actor_system_config::parse(int argc, char** argv,
actor_system_config&
actor_system_config::add_actor_factory(std::string name, actor_factory fun) {
actor_factories_.emplace(std::move(name), std::move(fun));
actor_factories.emplace(std::move(name), std::move(fun));
return *this;
}
actor_system_config&
actor_system_config::add_error_category(atom_value x, error_renderer y) {
error_renderers_.emplace(x, y);
error_renderers.emplace(x, y);
return *this;
}
......
......@@ -42,6 +42,7 @@
#include "caf/abstract_group.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/message_builder.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/type_nr.hpp"
#include "caf/detail/safe_equal.hpp"
......@@ -124,16 +125,18 @@ uniform_type_info_map::make_value(const std::string& x) const {
auto i = std::find_if(builtin_.begin(), e, pred);
if (i != e)
return i->second();
auto j = custom_by_name_.find(x);
if (j != custom_by_name_.end())
auto& custom_names = system().config().value_factories_by_name;
auto j = custom_names.find(x);
if (j != custom_names.end())
return j->second();
return nullptr;
}
type_erased_value_ptr
uniform_type_info_map::make_value(const std::type_info& x) const {
auto i = custom_by_rtti_.find(std::type_index(x));
if (i != custom_by_rtti_.end())
auto& custom_by_rtti = system().config().value_factories_by_rtti;
auto i = custom_by_rtti.find(std::type_index(x));
if (i != custom_by_rtti.end())
return i->second();
return nullptr;
}
......@@ -145,16 +148,18 @@ uniform_type_info_map::portable_name(uint16_t nr,
return &builtin_names_[nr - 1];
if (! ti)
return nullptr;
auto i = custom_names_.find(std::type_index(*ti));
if (i != custom_names_.end())
auto& custom_names = system().config().type_names_by_rtti;
auto i = custom_names.find(std::type_index(*ti));
if (i != custom_names.end())
return &(i->second);
return nullptr;
}
uniform_type_info_map::error_renderer
uniform_type_info_map::renderer(atom_value x) const {
auto i = error_renderers_.find(x);
if (i != error_renderers_.end())
auto& error_renderers = system().config().error_renderers;
auto i = error_renderers.find(x);
if (i != error_renderers.end())
return i->second;
return nullptr;
}
......@@ -164,8 +169,9 @@ actor_factory_result uniform_type_info_map::make_actor(const std::string& name,
message& msg) const {
strong_actor_ptr res;
std::set<std::string> ifs;
auto i = factories_.find(name);
if (i != factories_.end())
auto& factories = system().config().actor_factories;
auto i = factories.find(name);
if (i != factories.end())
std::tie(res, ifs) = i->second(cfg, msg);
return std::make_pair(std::move(res), std::move(ifs));
}
......
......@@ -127,6 +127,16 @@ void serialize(Processor&, test_empty_non_pod&, const unsigned int) {
// nop
}
class config : public actor_system_config {
public:
config() {
add_message_type<test_enum>("test_enum");
add_message_type<raw_struct>("raw_struct");
add_message_type<test_array>("test_array");
add_message_type<test_empty_non_pod>("test_empty_non_pod");
}
};
struct fixture {
int32_t i32 = -345;
float f32 = 3.45f;
......@@ -143,6 +153,7 @@ struct fixture {
};
int ra[3] = {1, 2, 3};
config cfg;
actor_system system;
scoped_execution_unit context;
message msg;
......@@ -192,11 +203,7 @@ struct fixture {
}
fixture()
: system(actor_system_config{}
.add_message_type<test_enum>("test_enum")
.add_message_type<raw_struct>("raw_struct")
.add_message_type<test_array>("test_array")
.add_message_type<test_empty_non_pod>("test_empty_non_pod")),
: system(cfg),
context(&system) {
rs.str.assign(string(str.rbegin(), str.rend()));
msg = make_message(i32, te, str, rs);
......
......@@ -296,11 +296,11 @@ behavior foo2(event_based_actor* self) {
}
struct fixture {
actor_system_config cfg;
actor_system system;
scoped_actor self;
fixture() : system(actor_system_config()
.add_message_type<get_state_msg>("get_state_msg")),
fixture() : system(cfg.add_message_type<get_state_msg>("get_state_msg")),
self(system) {
// nop
}
......
......@@ -95,7 +95,7 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
private:
network::asio_multiplexer backend_;
};
if (sys.backend_name() == atom("asio"))
if (sys.config().middleman_network_backend == atom("asio"))
return new asio_impl(sys);
# endif // CAF_USE_ASIO
return new impl(sys);
......@@ -333,7 +333,7 @@ void middleman::init(actor_system_config& cfg) {
.add_message_type<new_data_msg>("@new_data_msg");
// compute and set ID for this network node
node_id this_node{node_id::data::create_singleton()};
cfg.network_id.swap(this_node);
system().node_.swap(this_node);
// set scheduling parameters for multiplexer
backend().max_throughput(cfg.scheduler_max_throughput);
backend().max_consecutive_reads(cfg.middleman_max_consecutive_reads);
......
......@@ -117,8 +117,7 @@ string hexstr(const buffer& buf) {
class fixture {
public:
fixture(bool autoconn = false)
: system(actor_system_config{}
.load<io::middleman, network::test_multiplexer>()
: system(cfg.load<io::middleman, network::test_multiplexer>()
.set("middleman.enable-automatic-connections", autoconn)) {
auto& mm = system.middleman();
mpx_ = dynamic_cast<network::test_multiplexer*>(&mm.backend());
......@@ -435,6 +434,7 @@ public:
return {this};
}
actor_system_config cfg;
actor_system system;
private:
......
......@@ -39,18 +39,16 @@ public:
config() {
load<caf::io::middleman>();
add_message_type<std::vector<int>>("std::vector<int>");
}
config& parse() {
actor_system_config::parse(caf::test::engine::argc(),
caf::test::engine::argv());
return *this;
}
};
struct fixture {
caf::actor_system server_side{config{}.parse()};
caf::actor_system client_side{config{}.parse()};
config server_side_config;
caf::actor_system server_side{server_side_config};
config client_side_config;
caf::actor_system client_side{client_side_config};
caf::io::middleman& server_side_mm = server_side.middleman();
caf::io::middleman& client_side_mm = client_side.middleman();
};
......
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