Commit 751f86fd authored by Dominik Charousset's avatar Dominik Charousset

Deprecate middleman_* member variables in config

parent ca325fea
...@@ -66,6 +66,7 @@ set(LIBCAF_CORE_SRCS ...@@ -66,6 +66,7 @@ set(LIBCAF_CORE_SRCS
src/local_actor.cpp src/local_actor.cpp
src/logger.cpp src/logger.cpp
src/mailbox_element.cpp src/mailbox_element.cpp
src/make_config_option.cpp
src/match_case.cpp src/match_case.cpp
src/memory_managed.cpp src/memory_managed.cpp
src/merged_tuple.cpp src/merged_tuple.cpp
......
...@@ -103,6 +103,9 @@ public: ...@@ -103,6 +103,9 @@ public:
return *this; return *this;
} }
opt_group& add_neg(bool& storage, const char* name,
const char* description);
private: private:
config_option_set& xs_; config_option_set& xs_;
const char* category_; const char* category_;
...@@ -354,12 +357,15 @@ public: ...@@ -354,12 +357,15 @@ public:
// -- middleman parameters --------------------------------------------------- // -- middleman parameters ---------------------------------------------------
atom_value middleman_network_backend; atom_value middleman_network_backend CAF_DEPRECATED;
std::string middleman_app_identifier; std::string middleman_app_identifier CAF_DEPRECATED;
size_t middleman_max_consecutive_reads; bool middleman_enable_automatic_connections CAF_DEPRECATED;
size_t middleman_heartbeat_interval; size_t middleman_max_consecutive_reads CAF_DEPRECATED;
size_t middleman_cached_udp_buffers; size_t middleman_heartbeat_interval CAF_DEPRECATED;
size_t middleman_max_pending_msgs; bool middleman_detach_utility_actors CAF_DEPRECATED;
bool middleman_detach_multiplexer CAF_DEPRECATED;
size_t middleman_cached_udp_buffers CAF_DEPRECATED;
size_t middleman_max_pending_msgs CAF_DEPRECATED;
// -- OpenCL parameters ------------------------------------------------------ // -- OpenCL parameters ------------------------------------------------------
......
...@@ -113,6 +113,9 @@ public: ...@@ -113,6 +113,9 @@ public:
return *this; return *this;
} }
/// @private
config_option_set& add(config_option&& opt);
/// Generates human-readable help text for all options. /// Generates human-readable help text for all options.
std::string help_text(bool global_only = true) const; std::string help_text(bool global_only = true) const;
......
...@@ -435,7 +435,7 @@ T get(const config_value::dictionary& xs, const std::string& name) { ...@@ -435,7 +435,7 @@ T get(const config_value::dictionary& xs, const std::string& name) {
/// Retrieves the value associated to `name` from `xs` or returns /// Retrieves the value associated to `name` from `xs` or returns
/// `default_value`. /// `default_value`.
/// @relates config_value /// @relates config_value
template <class T> template <class T, class E = detail::enable_if_t<!std::is_pointer<T>::value>>
T get_or(const config_value::dictionary& xs, const std::string& name, T get_or(const config_value::dictionary& xs, const std::string& name,
const T& default_value) { const T& default_value) {
auto result = get_if<T>(&xs, name); auto result = get_if<T>(&xs, name);
...@@ -444,6 +444,12 @@ T get_or(const config_value::dictionary& xs, const std::string& name, ...@@ -444,6 +444,12 @@ T get_or(const config_value::dictionary& xs, const std::string& name,
return default_value; return default_value;
} }
/// Retrieves the value associated to `name` from `xs` or returns
/// `default_value`.
/// @relates config_value
std::string get_or(const config_value::dictionary& xs, const std::string& name,
const char* default_value);
/// Tries to retrieve the value associated to `name` from `xs`. /// Tries to retrieve the value associated to `name` from `xs`.
/// @relates config_value /// @relates config_value
template <class T> template <class T>
...@@ -473,7 +479,7 @@ T get(const std::map<std::string, config_value::dictionary>& xs, ...@@ -473,7 +479,7 @@ T get(const std::map<std::string, config_value::dictionary>& xs,
/// Retrieves the value associated to `name` from `xs` or returns /// Retrieves the value associated to `name` from `xs` or returns
/// `default_value`. /// `default_value`.
/// @relates config_value /// @relates config_value
template <class T> template <class T, class E = detail::enable_if_t<!std::is_pointer<T>::value>>
T get_or(const std::map<std::string, config_value::dictionary>& xs, T get_or(const std::map<std::string, config_value::dictionary>& xs,
const std::string& name, const T& default_value) { const std::string& name, const T& default_value) {
auto result = get_if<T>(&xs, name); auto result = get_if<T>(&xs, name);
...@@ -482,6 +488,12 @@ T get_or(const std::map<std::string, config_value::dictionary>& xs, ...@@ -482,6 +488,12 @@ T get_or(const std::map<std::string, config_value::dictionary>& xs,
return default_value; return default_value;
} }
/// Retrieves the value associated to `name` from `xs` or returns
/// `default_value`.
/// @relates config_value
std::string get_or(const std::map<std::string, config_value::dictionary>& xs,
const std::string& name, const char* default_value);
/// Tries to retrieve the value associated to `name` from `cfg`. /// Tries to retrieve the value associated to `name` from `cfg`.
/// @relates config_value /// @relates config_value
template <class T> template <class T>
...@@ -499,12 +511,15 @@ T get(const actor_system_config& cfg, const std::string& name) { ...@@ -499,12 +511,15 @@ T get(const actor_system_config& cfg, const std::string& name) {
/// Retrieves the value associated to `name` from `cfg` or returns /// Retrieves the value associated to `name` from `cfg` or returns
/// `default_value`. /// `default_value`.
/// @relates config_value /// @relates config_value
template <class T> template <class T, class E = detail::enable_if_t<!std::is_pointer<T>::value>>
T get_or(const actor_system_config& cfg, const std::string& name, T get_or(const actor_system_config& cfg, const std::string& name,
const T& default_value) { const T& default_value) {
return get_or(content(cfg), name, default_value); return get_or(content(cfg), name, default_value);
} }
std::string get_or(const actor_system_config& cfg, const std::string& name,
const char* default_value);
/// @relates config_value /// @relates config_value
bool operator<(const config_value& x, const config_value& y); bool operator<(const config_value& x, const config_value& y);
......
...@@ -72,6 +72,7 @@ extern const bool inline_output; ...@@ -72,6 +72,7 @@ extern const bool inline_output;
namespace middleman { namespace middleman {
extern const char* app_identifier;
extern const atom_value network_backend; extern const atom_value network_backend;
extern const size_t max_consecutive_reads; extern const size_t max_consecutive_reads;
extern const size_t heartbeat_interval; extern const size_t heartbeat_interval;
......
...@@ -40,7 +40,7 @@ config_option make_config_option(const char* category, const char* name, ...@@ -40,7 +40,7 @@ config_option make_config_option(const char* category, const char* name,
return detail::type_name<T>(); return detail::type_name<T>();
} }
}; };
return {category, name, description, std::is_same<T, bool>::value, vtbl}; return {category, name, description, false, vtbl};
} }
/// Creates a config option that synchronizes with `storage`. /// Creates a config option that synchronizes with `storage`.
...@@ -60,8 +60,25 @@ config_option make_config_option(T& storage, const char* category, ...@@ -60,8 +60,25 @@ config_option make_config_option(T& storage, const char* category,
return detail::type_name<T>(); return detail::type_name<T>();
} }
}; };
return {category, name, description, std::is_same<T, bool>::value, return {category, name, description, false, vtbl, &storage};
vtbl, &storage};
} }
// -- backward compatbility, do not use for new code ! -------------------------
// Inverts the value when writing to `storage`.
config_option make_negated_config_option(bool& storage, const char* category,
const char* name,
const char* description);
// -- specializations for common types.
template <>
config_option make_config_option<bool>(const char* category, const char* name,
const char* description);
template <>
config_option make_config_option<bool>(bool& storage, const char* category,
const char* name,
const char* description);
} // namespace caf } // namespace caf
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <fstream> #include <fstream>
#include <sstream> #include <sstream>
#include "caf/defaults.hpp"
#include "caf/detail/gcd.hpp" #include "caf/detail/gcd.hpp"
#include "caf/detail/ini_consumer.hpp" #include "caf/detail/ini_consumer.hpp"
#include "caf/detail/parser/read_ini.hpp" #include "caf/detail/parser/read_ini.hpp"
...@@ -40,6 +41,13 @@ actor_system_config::opt_group::opt_group(config_option_set& xs, ...@@ -40,6 +41,13 @@ actor_system_config::opt_group::opt_group(config_option_set& xs,
// nop // nop
} }
actor_system_config::opt_group&
actor_system_config::opt_group::add_neg(bool& storage, const char* name,
const char* description) {
xs_.add(make_negated_config_option(storage, category_, name, description));
return *this;
}
actor_system_config::~actor_system_config() { actor_system_config::~actor_system_config() {
// nop // nop
} }
...@@ -83,11 +91,14 @@ actor_system_config::actor_system_config() ...@@ -83,11 +91,14 @@ actor_system_config::actor_system_config()
logger_console_format = "%m"; logger_console_format = "%m";
logger_verbosity = atom("trace"); logger_verbosity = atom("trace");
logger_inline_output = false; logger_inline_output = false;
middleman_network_backend = atom("default"); middleman_network_backend = defaults::middleman::network_backend;
middleman_max_consecutive_reads = 50; middleman_enable_automatic_connections = false;
middleman_heartbeat_interval = 0; middleman_max_consecutive_reads = defaults::middleman::max_consecutive_reads;
middleman_cached_udp_buffers = 10; middleman_heartbeat_interval = defaults::middleman::heartbeat_interval;
middleman_max_pending_msgs = 10; middleman_detach_utility_actors = true;
middleman_detach_multiplexer = true;
middleman_cached_udp_buffers = defaults::middleman::cached_udp_buffers;
middleman_max_pending_msgs = defaults::middleman::max_pending_msgs;
// fill our options vector for creating INI and CLI parsers // fill our options vector for creating INI and CLI parsers
opt_group{custom_options_, "global"} opt_group{custom_options_, "global"}
.add<bool>(cli_helptext_printed, "help,h?", "print help and exit") .add<bool>(cli_helptext_printed, "help,h?", "print help and exit")
...@@ -149,28 +160,30 @@ actor_system_config::actor_system_config() ...@@ -149,28 +160,30 @@ actor_system_config::actor_system_config()
.add(logger_component_filter, "filter", .add(logger_component_filter, "filter",
"deprecated (use console-component-filter instead)"); "deprecated (use console-component-filter instead)");
opt_group{custom_options_, "middleman"} opt_group{custom_options_, "middleman"}
.add<bool>("enable-automatic-connections",
"enables automatic connection management")
.add<bool>("attach-utility-actors",
"schedule utility actors instead of dedicating individual threads")
.add<bool>("manual-multiplexing",
"disables background activity of the multiplexer")
.add<bool>("disable-tcp", "disables communication via TCP")
.add<bool>("enable-udp", "enable communication via UDP")
.add(middleman_network_backend, "network-backend", .add(middleman_network_backend, "network-backend",
"sets the network backend to either 'default' or 'asio' (if available)") "sets the network backend to either 'default' or 'asio' (if available)")
.add(middleman_app_identifier, "app-identifier", .add(middleman_app_identifier, "app-identifier",
"sets the application identifier of this node") "sets the application identifier of this node")
.add(middleman_enable_automatic_connections, "enable-automatic-connections",
"enables automatic connection management")
.add(middleman_max_consecutive_reads, "max-consecutive-reads", .add(middleman_max_consecutive_reads, "max-consecutive-reads",
"sets the maximum number of consecutive I/O reads per broker") "sets the maximum number of consecutive I/O reads per broker")
.add(middleman_heartbeat_interval, "heartbeat-interval", .add(middleman_heartbeat_interval, "heartbeat-interval",
"sets the interval (ms) of heartbeat, 0 (default) means disabling it") "sets the interval (ms) of heartbeat, 0 (default) means disabling it")
.add(middleman_detach_utility_actors, "detach-utility-actors",
"deprecated, see attach-utility-actors instead")
.add_neg(middleman_detach_utility_actors, "attach-utility-actors",
"schedule utility actors instead of dedicating individual threads")
.add(middleman_detach_multiplexer, "detach-multiplexer",
"deprecated, see manual-multiplexing instead")
.add_neg(middleman_detach_multiplexer, "manual-multiplexing",
"disables background activity of the multiplexer")
.add(middleman_cached_udp_buffers, "cached-udp-buffers", .add(middleman_cached_udp_buffers, "cached-udp-buffers",
"sets the max number of UDP send buffers that will be cached for reuse " "sets the maximum for cached UDP send buffers (default: 10)")
"(default: 10)")
.add(middleman_max_pending_msgs, "max-pending-messages", .add(middleman_max_pending_msgs, "max-pending-messages",
"sets the max number of UDP pending messages due to ordering " "sets the maximum for reordering of UDP receive buffers (default: 10)")
"(default: 10)"); .add<bool>("disable-tcp", "disables communication via TCP")
.add<bool>("enable-udp", "enable communication via UDP");
opt_group(custom_options_, "opencl") opt_group(custom_options_, "opencl")
.add(opencl_device_ids, "device-ids", .add(opencl_device_ids, "device-ids",
"restricts which OpenCL devices are accessed by CAF"); "restricts which OpenCL devices are accessed by CAF");
......
...@@ -62,6 +62,11 @@ config_option_set::config_option_set() { ...@@ -62,6 +62,11 @@ config_option_set::config_option_set() {
// nop // nop
} }
config_option_set& config_option_set::add(config_option&& opt) {
opts_.emplace_back(std::move(opt));
return *this;
}
std::string config_option_set::help_text(bool global_only) const { std::string config_option_set::help_text(bool global_only) const {
//<--- argument --------> <---- desciption ----> //<--- argument --------> <---- desciption ---->
// (-w|--write) <string> : output file // (-w|--write) <string> : output file
......
...@@ -124,5 +124,26 @@ std::string to_string(const config_value& x) { ...@@ -124,5 +124,26 @@ std::string to_string(const config_value& x) {
return deep_to_string(x.get_data()); return deep_to_string(x.get_data());
} }
std::string get_or(const config_value::dictionary& xs, const std::string& name,
const char* default_value) {
auto result = get_if<std::string>(&xs, name);
if (result)
return std::move(*result);
return default_value;
}
std::string get_or(const std::map<std::string, config_value::dictionary>& xs,
const std::string& name, const char* default_value) {
auto result = get_if<std::string>(&xs, name);
if (result)
return std::move(*result);
return default_value;
}
std::string get_or(const actor_system_config& cfg, const std::string& name,
const char* default_value) {
return get_or(content(cfg), name, default_value);
}
} // namespace caf } // namespace caf
...@@ -84,6 +84,7 @@ const bool inline_output = false; ...@@ -84,6 +84,7 @@ const bool inline_output = false;
namespace middleman { namespace middleman {
const char* app_identifier = "";
const atom_value network_backend = atom("default"); const atom_value network_backend = atom("default");
const size_t max_consecutive_reads = 50; const size_t max_consecutive_reads = 50;
const size_t heartbeat_interval = 0; const size_t heartbeat_interval = 0;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#include "caf/make_config_option.hpp"
namespace caf {
namespace {
using vtbl_type = config_option::vtbl_type;
error bool_check(const config_value& x) {
if (holds_alternative<bool>(x))
return none;
return make_error(pec::type_mismatch);
}
void bool_store(void* ptr, const config_value& x) {
*static_cast<bool*>(ptr) = get<bool>(x);
}
void bool_store_neg(void* ptr, const config_value& x) {
*static_cast<bool*>(ptr) = !get<bool>(x);
}
std::string bool_name() {
return detail::type_name<bool>();
}
constexpr vtbl_type bool_vtbl{bool_check, bool_store, bool_name};
constexpr vtbl_type bool_neg_vtbl{bool_check, bool_store_neg, bool_name};
} // namespace anonymous
config_option make_negated_config_option(bool& storage, const char* category,
const char* name,
const char* description) {
return {category, name, description, true, bool_neg_vtbl, &storage};
}
template <>
config_option make_config_option<bool>(const char* category, const char* name,
const char* description) {
return {category, name, description, true, bool_vtbl};
}
template <>
config_option make_config_option<bool>(bool& storage, const char* category,
const char* name,
const char* description) {
return {category, name, description, true, bool_vtbl, &storage};
}
} // namespace caf
...@@ -202,12 +202,17 @@ CAF_TEST(heterogeneous dictionary) { ...@@ -202,12 +202,17 @@ CAF_TEST(heterogeneous dictionary) {
} }
CAF_TEST(successful parsing) { CAF_TEST(successful parsing) {
auto parse = [](const std::string& str) { // Store the parsed value on the stack, because the unit test framework takes
// references when comparing values. Since we call get<T>() on the result of
// parse(), we would end up with a reference to a temporary.
config_value parsed;
auto parse = [&](const std::string& str) -> config_value& {
auto x = config_value::parse(str); auto x = config_value::parse(str);
if (!x) if (!x)
CAF_FAIL("cannot parse " << str << ": assumed a result but error " CAF_FAIL("cannot parse " << str << ": assumed a result but error "
<< to_string(x.error())); << to_string(x.error()));
return std::move(*x); parsed = std::move(*x);
return parsed;
}; };
using di = std::map<string, int>; // Dictionary-of-integers. using di = std::map<string, int>; // Dictionary-of-integers.
using ls = std::vector<string>; // List-of-strings. using ls = std::vector<string>; // List-of-strings.
......
...@@ -317,7 +317,10 @@ public: ...@@ -317,7 +317,10 @@ public:
auto e = bd(remote_appid); auto e = bd(remote_appid);
if (e) if (e)
return false; return false;
if (remote_appid != callee_.system().config().middleman_app_identifier) { auto appid = get_if<std::string>(&callee_.config(),
"middleman.app-identifier");
if ((appid && *appid != remote_appid)
|| (!appid && !remote_appid.empty())) {
CAF_LOG_ERROR("app identifier mismatch"); CAF_LOG_ERROR("app identifier mismatch");
return false; return false;
} }
...@@ -368,7 +371,10 @@ public: ...@@ -368,7 +371,10 @@ public:
auto e = bd(remote_appid); auto e = bd(remote_appid);
if (e) if (e)
return false; return false;
if (remote_appid != callee_.system().config().middleman_app_identifier) { auto appid = get_if<std::string>(&callee_.config(),
"middleman.app-identifier");
if ((appid && *appid != remote_appid)
|| (!appid && !remote_appid.empty())) {
CAF_LOG_ERROR("app identifier mismatch"); CAF_LOG_ERROR("app identifier mismatch");
return false; return false;
} }
......
...@@ -584,7 +584,7 @@ protected: ...@@ -584,7 +584,7 @@ protected:
template <class Policy> template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) { void handle_event_impl(io::network::operation op, Policy& policy) {
CAF_LOG_TRACE(CAF_ARG(op)); CAF_LOG_TRACE(CAF_ARG(op));
auto mcr = max_consecutive_reads(); auto mcr = max_consecutive_reads_;
switch (op) { switch (op) {
case io::network::operation::read: { case io::network::operation::read: {
// Loop until an error occurs or we have nothing more to read // Loop until an error occurs or we have nothing more to read
...@@ -651,12 +651,13 @@ protected: ...@@ -651,12 +651,13 @@ protected:
} }
private: private:
size_t max_consecutive_reads();
void prepare_next_read(); void prepare_next_read();
void prepare_next_write(); void prepare_next_write();
size_t max_consecutive_reads_;
// state for reading // state for reading
manager_ptr reader_; manager_ptr reader_;
size_t read_threshold_; size_t read_threshold_;
...@@ -840,7 +841,7 @@ protected: ...@@ -840,7 +841,7 @@ protected:
template <class Policy> template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) { void handle_event_impl(io::network::operation op, Policy& policy) {
CAF_LOG_TRACE(CAF_ARG(op)); CAF_LOG_TRACE(CAF_ARG(op));
auto mcr = max_consecutive_reads(); auto mcr = max_consecutive_reads_;
switch (op) { switch (op) {
case io::network::operation::read: { case io::network::operation::read: {
// Loop until an error occurs or we have nothing more to read // Loop until an error occurs or we have nothing more to read
...@@ -910,7 +911,7 @@ protected: ...@@ -910,7 +911,7 @@ protected:
} }
private: private:
size_t max_consecutive_reads(); size_t max_consecutive_reads_;
void prepare_next_read(); void prepare_next_read();
......
...@@ -18,25 +18,23 @@ ...@@ -18,25 +18,23 @@
#include "caf/io/basp_broker.hpp" #include "caf/io/basp_broker.hpp"
#include <limits>
#include <chrono> #include <chrono>
#include <limits>
#include "caf/sec.hpp" #include "caf/actor_registry.hpp"
#include "caf/send.hpp" #include "caf/actor_system_config.hpp"
#include "caf/after.hpp" #include "caf/after.hpp"
#include "caf/make_counted.hpp" #include "caf/defaults.hpp"
#include "caf/detail/sync_request_bouncer.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/detail/sync_request_bouncer.hpp"
#include "caf/io/basp/all.hpp" #include "caf/io/basp/all.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/connection_helper.hpp" #include "caf/io/connection_helper.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/network/interfaces.hpp" #include "caf/io/network/interfaces.hpp"
#include "caf/make_counted.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -77,8 +75,10 @@ basp_broker_state::basp_broker_state(broker* selfptr) ...@@ -77,8 +75,10 @@ basp_broker_state::basp_broker_state(broker* selfptr)
static_cast<proxy_registry::backend&>(*this)), static_cast<proxy_registry::backend&>(*this)),
self(selfptr), self(selfptr),
instance(selfptr, *this), instance(selfptr, *this),
max_buffers(self->config().middleman_cached_udp_buffers), max_buffers(get_or(self->config(), "middleman.cached-udp-buffers",
max_pending_messages(self->config().middleman_max_pending_msgs) { defaults::middleman::cached_udp_buffers)),
max_pending_messages(get_or(self->config(), "middleman.max-pending-msgs",
defaults::middleman::max_pending_msgs)) {
CAF_ASSERT(this_node() != none); CAF_ASSERT(this_node() != none);
} }
...@@ -670,7 +670,8 @@ behavior basp_broker::make_behavior() { ...@@ -670,7 +670,8 @@ behavior basp_broker::make_behavior() {
} }
state.automatic_connections = true; state.automatic_connections = true;
} }
auto heartbeat_interval = config().middleman_heartbeat_interval; auto heartbeat_interval = get_or(config(), "middleman.heartbeat-interval",
defaults::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);
......
...@@ -16,10 +16,12 @@ ...@@ -16,10 +16,12 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/io/connection_helper.hpp"
#include <chrono> #include <chrono>
#include "caf/defaults.hpp"
#include "caf/io/basp/instance.hpp" #include "caf/io/basp/instance.hpp"
#include "caf/io/connection_helper.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -37,7 +39,8 @@ behavior datagram_connection_broker(broker* self, uint16_t port, ...@@ -37,7 +39,8 @@ behavior datagram_connection_broker(broker* self, uint16_t port,
actor system_broker) { actor system_broker) {
auto& mx = self->system().middleman().backend(); auto& mx = self->system().middleman().backend();
auto& this_node = self->system().node(); auto& this_node = self->system().node();
auto& app_id = self->system().config().middleman_app_identifier; auto app_id = get_or(self->config(), "middleman.app-identifier",
defaults::middleman::app_identifier);
for (auto& kvp : addresses) { for (auto& kvp : addresses) {
for (auto& addr : kvp.second) { for (auto& addr : kvp.second) {
auto eptr = mx.new_remote_udp_endpoint(addr, port); auto eptr = mx.new_remote_udp_endpoint(addr, port);
......
...@@ -18,18 +18,16 @@ ...@@ -18,18 +18,16 @@
#include "caf/io/network/default_multiplexer.hpp" #include "caf/io/network/default_multiplexer.hpp"
#include "caf/config.hpp"
#include "caf/optional.hpp"
#include "caf/make_counted.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/config.hpp"
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/defaults.hpp"
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/network/protocol.hpp"
#include "caf/io/network/interfaces.hpp" #include "caf/io/network/interfaces.hpp"
#include "caf/io/network/protocol.hpp"
#include "caf/make_counted.hpp"
#include "caf/optional.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
# include <winsock2.h> # include <winsock2.h>
...@@ -1067,6 +1065,9 @@ void pipe_reader::init(native_socket sock_fd) { ...@@ -1067,6 +1065,9 @@ void pipe_reader::init(native_socket sock_fd) {
stream::stream(default_multiplexer& backend_ref, native_socket sockfd) stream::stream(default_multiplexer& backend_ref, native_socket sockfd)
: event_handler(backend_ref, sockfd), : event_handler(backend_ref, sockfd),
max_consecutive_reads_(
get_or(backend().system().config(), "middleman.max-consecutive-reads",
defaults::middleman::max_consecutive_reads)),
read_threshold_(1), read_threshold_(1),
collected_(0), collected_(0),
ack_writes_(false), ack_writes_(false),
...@@ -1130,10 +1131,6 @@ void stream::removed_from_loop(operation op) { ...@@ -1130,10 +1131,6 @@ void stream::removed_from_loop(operation op) {
} }
} }
size_t stream::max_consecutive_reads() {
return backend().system().config().middleman_max_consecutive_reads;
}
void stream::prepare_next_read() { void stream::prepare_next_read() {
collected_ = 0; collected_ = 0;
switch (rd_flag_) { switch (rd_flag_) {
...@@ -1204,6 +1201,9 @@ void acceptor::removed_from_loop(operation op) { ...@@ -1204,6 +1201,9 @@ void acceptor::removed_from_loop(operation op) {
datagram_handler::datagram_handler(default_multiplexer& backend_ref, datagram_handler::datagram_handler(default_multiplexer& backend_ref,
native_socket sockfd) native_socket sockfd)
: event_handler(backend_ref, sockfd), : event_handler(backend_ref, sockfd),
max_consecutive_reads_(
get_or(backend().system().config(), "middleman.max-consecutive-reads",
defaults::middleman::max_consecutive_reads)),
max_datagram_size_(receive_buffer_size), max_datagram_size_(receive_buffer_size),
rd_buf_(receive_buffer_size), rd_buf_(receive_buffer_size),
send_buffer_size_(0), send_buffer_size_(0),
...@@ -1305,10 +1305,6 @@ void datagram_handler::removed_from_loop(operation op) { ...@@ -1305,10 +1305,6 @@ void datagram_handler::removed_from_loop(operation op) {
}; };
} }
size_t datagram_handler::max_consecutive_reads() {
return backend().system().config().middleman_max_consecutive_reads;
}
void datagram_handler::prepare_next_read() { void datagram_handler::prepare_next_read() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.second.size()) CAF_LOG_TRACE(CAF_ARG(wr_buf_.second.size())
<< CAF_ARG(wr_offline_buf_.size())); << CAF_ARG(wr_offline_buf_.size()));
......
...@@ -18,12 +18,12 @@ ...@@ -18,12 +18,12 @@
#include "caf/io/basp/instance.hpp" #include "caf/io/basp/instance.hpp"
#include "caf/streambuf.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/defaults.hpp"
#include "caf/io/basp/version.hpp" #include "caf/io/basp/version.hpp"
#include "caf/streambuf.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -364,8 +364,9 @@ void instance::write_server_handshake(execution_unit* ctx, ...@@ -364,8 +364,9 @@ void instance::write_server_handshake(execution_unit* ctx,
} }
CAF_LOG_DEBUG_IF(!pa && port, "no actor published"); CAF_LOG_DEBUG_IF(!pa && port, "no actor published");
auto writer = make_callback([&](serializer& sink) -> error { auto writer = make_callback([&](serializer& sink) -> error {
auto& ref = callee_.system().config().middleman_app_identifier; auto id = get_or(callee_.config(), "middleman.app-identifier",
auto e = sink(const_cast<std::string&>(ref)); defaults::middleman::app_identifier);
auto e = sink(id);
if (e) if (e)
return e; return e;
if (pa != nullptr) { if (pa != nullptr) {
...@@ -404,7 +405,8 @@ void instance::write_client_handshake(execution_unit* ctx, ...@@ -404,7 +405,8 @@ void instance::write_client_handshake(execution_unit* ctx,
const node_id& remote_side, const node_id& remote_side,
uint16_t sequence_number) { uint16_t sequence_number) {
write_client_handshake(ctx, buf, remote_side, this_node_, write_client_handshake(ctx, buf, remote_side, this_node_,
callee_.system().config().middleman_app_identifier, get_or(callee_.config(), "middleman.app-identifier",
defaults::middleman::app_identifier),
sequence_number); sequence_number);
} }
......
...@@ -32,6 +32,7 @@ ...@@ -32,6 +32,7 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/defaults.hpp"
#include "caf/actor_proxy.hpp" #include "caf/actor_proxy.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
...@@ -89,7 +90,9 @@ private: ...@@ -89,7 +90,9 @@ private:
} // namespace <anonymous> } // namespace <anonymous>
actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) { actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
switch (atom_uint(sys.config().middleman_network_backend)) { auto atm = get_or(sys.config(), "middleman.network-backend",
defaults::middleman::network_backend);
switch (atom_uint(atm)) {
# ifdef CAF_USE_ASIO # ifdef CAF_USE_ASIO
case atom_uint(atom("asio")): case atom_uint(atom("asio")):
return new mm_impl<network::asio_multiplexer>(sys); return new mm_impl<network::asio_multiplexer>(sys);
...@@ -361,7 +364,9 @@ void middleman::stop() { ...@@ -361,7 +364,9 @@ void middleman::stop() {
void middleman::init(actor_system_config& cfg) { void middleman::init(actor_system_config& cfg) {
// never detach actors when using the testing multiplexer // never detach actors when using the testing multiplexer
if (cfg.middleman_network_backend == atom("testing")) auto network_backend = get_or(cfg, "middleman.network-backend",
defaults::middleman::network_backend);
if (network_backend == atom("testing"))
cfg.set("middleman.attach-utility-actors", true); cfg.set("middleman.attach-utility-actors", true);
// add remote group module to config // add remote group module to config
struct remote_groups : group_module { struct remote_groups : group_module {
......
...@@ -53,6 +53,11 @@ public: ...@@ -53,6 +53,11 @@ public:
return system_; return system_;
} }
/// Returns the system-wide configuration.
inline const actor_system_config& config() const {
return system_.config();
}
/// Returns true if configured to require certificate-based authentication /// Returns true if configured to require certificate-based authentication
/// of peers. /// of peers.
bool authentication_enabled(); bool authentication_enabled();
......
...@@ -52,7 +52,7 @@ void manager::stop() { ...@@ -52,7 +52,7 @@ void manager::stop() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
scoped_actor self{system(), true}; scoped_actor self{system(), true};
self->send_exit(manager_, exit_reason::kill); self->send_exit(manager_, exit_reason::kill);
if (system().config().middleman_detach_utility_actors) if (!get_or(config(), "middleman.attach-utility-actors", false))
self->wait_for(manager_); self->wait_for(manager_);
manager_ = nullptr; manager_ = nullptr;
} }
......
...@@ -245,7 +245,7 @@ private: ...@@ -245,7 +245,7 @@ private:
} // namespace <anonymous> } // namespace <anonymous>
io::middleman_actor make_middleman_actor(actor_system& sys, actor db) { io::middleman_actor make_middleman_actor(actor_system& sys, actor db) {
return sys.config().middleman_detach_utility_actors return !get_or(sys.config(), "middleman.attach-utility-actors", false)
? sys.spawn<middleman_actor_impl, detached + hidden>(std::move(db)) ? sys.spawn<middleman_actor_impl, detached + hidden>(std::move(db))
: sys.spawn<middleman_actor_impl, hidden>(std::move(db)); : sys.spawn<middleman_actor_impl, hidden>(std::move(db));
} }
......
...@@ -55,9 +55,9 @@ public: ...@@ -55,9 +55,9 @@ public:
load<openssl::manager>(); load<openssl::manager>();
add_message_type<std::vector<int>>("std::vector<int>"); add_message_type<std::vector<int>>("std::vector<int>");
actor_system_config::parse(test::engine::argc(), test::engine::argv()); actor_system_config::parse(test::engine::argc(), test::engine::argv());
middleman_detach_multiplexer = false; set("middleman.manual-multiplexing", true);
middleman_detach_utility_actors = false; set("middleman.attach-utility-actors", true);
scheduler_policy = atom("testing"); set("scheduler.policy", atom("testing"));
} }
static std::string data_dir() { static std::string data_dir() {
......
...@@ -244,34 +244,39 @@ void bootstrap(actor_system& system, ...@@ -244,34 +244,39 @@ void bootstrap(actor_system& system,
return 1; \ return 1; \
} while (true) } while (true)
struct config : actor_system_config {
config() {
opt_group{custom_options_, "global"}
.add(hostfile, "hostfile", "path to hostfile")
.add(wdir, "wdir", "working directory");
}
string hostfile;
string wdir;
};
int main(int argc, char** argv) { int main(int argc, char** argv) {
actor_system_config cfg; config cfg;
cfg.parse(argc, argv); cfg.parse(argc, argv);
if (cfg.cli_helptext_printed) if (cfg.cli_helptext_printed)
return 0; return 0;
if (cfg.slave_mode) if (cfg.slave_mode)
RETURN_WITH_ERROR("cannot use slave mode in caf-run tool"); RETURN_WITH_ERROR("cannot use slave mode in caf-run tool");
string hostfile;
std::unique_ptr<char, void (*)(void*)> pwd{getcwd(nullptr, 0), ::free}; std::unique_ptr<char, void (*)(void*)> pwd{getcwd(nullptr, 0), ::free};
string wdir; if (cfg.hostfile.empty())
auto res = cfg.args_remainder.extract_opts({
{"hostfile", "path to the hostfile", hostfile},
{"wdir", wdir}
});
if (hostfile.empty())
RETURN_WITH_ERROR("no hostfile specified or hostfile is empty"); RETURN_WITH_ERROR("no hostfile specified or hostfile is empty");
auto& remainder = res.remainder; auto& remainder = cfg.remainder;
if (remainder.empty()) if (remainder.empty())
RETURN_WITH_ERROR("empty command line"); RETURN_WITH_ERROR("empty command line");
auto cmd = std::move(remainder.get_mutable_as<std::string>(0)); auto cmd = std::move(remainder.front());
vector<string> xs; vector<string> xs;
remainder.drop(1).extract([&](string& x) { xs.emplace_back(std::move(x)); }); for (auto i = cfg.remainder.begin() + 1; i != cfg.remainder.end(); ++i)
auto hosts = read_hostfile(hostfile); xs.emplace_back(std::move(*i));
auto hosts = read_hostfile(cfg.hostfile);
if (hosts.empty()) if (hosts.empty())
RETURN_WITH_ERROR("no valid entry in hostfile"); RETURN_WITH_ERROR("no valid entry in hostfile");
actor_system system{cfg}; actor_system system{cfg};
auto master = hosts.front(); auto master = hosts.front();
hosts.erase(hosts.begin()); hosts.erase(hosts.begin());
bootstrap(system, (wdir.empty()) ? pwd.get() : wdir.c_str(), master, bootstrap(system, (cfg.wdir.empty()) ? pwd.get() : cfg.wdir.c_str(), master,
std::move(hosts), cmd, xs); std::move(hosts), cmd, xs);
} }
...@@ -805,9 +805,9 @@ void caf_main(actor_system& sys, const config& cfg) { ...@@ -805,9 +805,9 @@ void caf_main(actor_system& sys, const config& cfg) {
}; };
// do a first pass on all files to extract node IDs and entities // do a first pass on all files to extract node IDs and entities
vector<intermediate_res> intermediate_results; vector<intermediate_res> intermediate_results;
intermediate_results.resize(cfg.args_remainder.size()); intermediate_results.resize(cfg.remainder.size());
for (size_t i = 0; i < cfg.args_remainder.size(); ++i) { for (size_t i = 0; i < cfg.remainder.size(); ++i) {
auto& file = cfg.args_remainder.get_as<string>(i); auto& file = cfg.remainder[i];
auto ptr = &intermediate_results[i]; auto ptr = &intermediate_results[i];
ptr->fname = file; ptr->fname = file;
ptr->fstream.reset(new std::ifstream(file)); ptr->fstream.reset(new std::ifstream(file));
......
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