Commit ca325fea authored by Dominik Charousset's avatar Dominik Charousset

WIP

parent cabb2d83
......@@ -356,13 +356,8 @@ public:
atom_value middleman_network_backend;
std::string middleman_app_identifier;
bool middleman_enable_automatic_connections;
size_t middleman_max_consecutive_reads;
size_t middleman_heartbeat_interval;
bool middleman_detach_utility_actors;
bool middleman_detach_multiplexer;
bool middleman_enable_tcp;
bool middleman_enable_udp;
size_t middleman_cached_udp_buffers;
size_t middleman_max_pending_msgs;
......
......@@ -73,13 +73,10 @@ extern const bool inline_output;
namespace middleman {
extern const atom_value network_backend;
extern const bool enable_automatic_connections;
extern const size_t max_consecutive_reads;
extern const size_t heartbeat_interval;
extern const bool detach_utility_actors;
extern const bool detach_multiplexer;
extern const bool enable_tcp;
extern const bool enable_udp;
extern const size_t cached_udp_buffers;
extern const size_t max_pending_msgs;
......
......@@ -94,10 +94,16 @@ public:
/// Deletes all proxies.
void clear();
/// Returns the hosting actor system.
inline actor_system& system() {
return system_;
}
/// Returns the hosting actor system.
inline const actor_system& system() const {
return system_;
}
inline size_t size() const {
return proxies_.size();
}
......
......@@ -84,13 +84,8 @@ actor_system_config::actor_system_config()
logger_verbosity = atom("trace");
logger_inline_output = false;
middleman_network_backend = atom("default");
middleman_enable_automatic_connections = false;
middleman_max_consecutive_reads = 50;
middleman_heartbeat_interval = 0;
middleman_detach_utility_actors = true;
middleman_detach_multiplexer = true;
middleman_enable_tcp = true;
middleman_enable_udp = false;
middleman_cached_udp_buffers = 10;
middleman_max_pending_msgs = 10;
// fill our options vector for creating INI and CLI parsers
......@@ -154,24 +149,22 @@ actor_system_config::actor_system_config()
.add(logger_component_filter, "filter",
"deprecated (use console-component-filter instead)");
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",
"sets the network backend to either 'default' or 'asio' (if available)")
.add(middleman_app_identifier, "app-identifier",
"sets the application identifier of this node")
.add(middleman_enable_automatic_connections, "enable-automatic-connections",
"enables or disables automatic connection management (off per default)")
.add(middleman_max_consecutive_reads, "max-consecutive-reads",
"sets the maximum number of consecutive I/O reads per broker")
.add(middleman_heartbeat_interval, "heartbeat-interval",
"sets the interval (ms) of heartbeat, 0 (default) means disabling it")
.add(middleman_detach_utility_actors, "detach-utility-actors",
"enables or disables detaching of utility actors")
.add(middleman_detach_multiplexer, "detach-multiplexer",
"enables or disables background activity of the multiplexer")
.add(middleman_enable_tcp, "enable-tcp",
"enable communication via TCP (on by default)")
.add(middleman_enable_udp, "enable-udp",
"enable communication via UDP (off by default)")
.add(middleman_cached_udp_buffers, "cached-udp-buffers",
"sets the max number of UDP send buffers that will be cached for reuse "
"(default: 10)")
......@@ -377,7 +370,7 @@ actor_system_config& actor_system_config::set_impl(const char* name,
auto opt = custom_options_.qualified_name_lookup(name);
if (opt != nullptr && opt->check(value) == none) {
opt->store(value);
content[opt->category()][name] = std::move(value);
content[opt->category()][opt->long_name()] = std::move(value);
}
return *this;
}
......
......@@ -85,13 +85,8 @@ const bool inline_output = false;
namespace middleman {
const atom_value network_backend = atom("default");
const bool enable_automatic_connections = false;
const size_t max_consecutive_reads = 50;
const size_t heartbeat_interval = 0;
const bool detach_utility_actors = true;
const bool detach_multiplexer = true;
const bool enable_tcp = true;
const bool enable_udp = false;
const size_t cached_udp_buffers = 10;
const size_t max_pending_msgs = 10;
......
......@@ -410,10 +410,10 @@ CAF_TEST(event_testee_series) {
self->send(et, "hello again event testee!");
self->send(et, "goodbye event testee!");
typed_actor<replies_to<get_state_msg>::with<string>> sub_et = et;
set<string> iface{"caf::replies_to<get_state_msg>::with<@str>",
"caf::replies_to<@str>::with<void>",
"caf::replies_to<float>::with<void>",
"caf::replies_to<@i32>::with<@i32>"};
std::set<string> iface{"caf::replies_to<get_state_msg>::with<@str>",
"caf::replies_to<@str>::with<void>",
"caf::replies_to<float>::with<void>",
"caf::replies_to<@i32>::with<@i32>"};
CAF_CHECK_EQUAL(join(sub_et->message_types(), ","), join(iface, ","));
self->send(sub_et, get_state_msg{});
// we expect three 42s
......
......@@ -99,10 +99,16 @@ public:
return namespace_;
}
/// Returns the hosting actor system.
inline actor_system& system() {
return namespace_.system();
}
/// Returns the system-wide configuration.
inline const actor_system_config& config() const {
return namespace_.system().config();
}
/// Returns the next outgoing sequence number for a connection.
virtual sequence_type next_sequence_number(connection_handle hdl) = 0;
......
......@@ -163,12 +163,15 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// get a 'SpawnServ' instance on the remote side
std::unordered_map<node_id, actor> spawn_servers;
// can be enabled by the user to let CAF automatically try
// to establish new connections at runtime to optimize
// routing paths by forming a mesh between all nodes
bool enable_automatic_connections = false;
bool enable_tcp = true;
bool enable_udp = false;
/// Configures whether BASP automatically open new connections to optimize
/// routing paths by forming a mesh between all nodes.
bool automatic_connections = false;
/// Configures whether BASP allows TCP connections.
bool allow_tcp = true;
/// Configures whether BASP allows UDP connections.
bool allow_udp = false;
// reusable send buffers for UDP communication
const size_t max_buffers;
......
......@@ -156,6 +156,11 @@ public:
return system_;
}
/// Returns the systemw-wide configuration.
inline const actor_system_config& config() const {
return system_.config();
}
/// Returns a handle to the actor managing the middleman singleton.
middleman_actor actor_handle();
......
......@@ -77,8 +77,8 @@ basp_broker_state::basp_broker_state(broker* selfptr)
static_cast<proxy_registry::backend&>(*this)),
self(selfptr),
instance(selfptr, *this),
max_buffers(self->system().config().middleman_cached_udp_buffers),
max_pending_messages(self->system().config().middleman_max_pending_msgs) {
max_buffers(self->config().middleman_cached_udp_buffers),
max_pending_messages(self->config().middleman_max_pending_msgs) {
CAF_ASSERT(this_node() != none);
}
......@@ -395,7 +395,7 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
CAF_ASSERT(this_context != nullptr);
CAF_LOG_TRACE(CAF_ARG(nid));
learned_new_node(nid);
if (!enable_automatic_connections)
if (!automatic_connections)
return;
// this member function gets only called once, after adding a new
// indirect connection to the routing table; hence, spawning
......@@ -412,9 +412,9 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
}
using namespace detail;
auto try_connect = [&](std::string item) {
auto tmp = system().config().middleman_detach_utility_actors
? system().spawn<detached + hidden>(connection_helper, self)
: system().spawn<hidden>(connection_helper, self);
auto tmp = get_or(config(), "middleman.attach-utility-actors", false)
? system().spawn<hidden>(connection_helper, self)
: system().spawn<detached + hidden>(connection_helper, self);
system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp));
auto writer = make_callback([&item](serializer& sink) -> error {
auto name_atm = atom("ConfigServ");
......@@ -430,9 +430,9 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
hdr, &writer);
instance.flush(*path);
};
if (enable_tcp)
if (allow_tcp)
try_connect("basp.default-connectivity-tcp");
if (enable_udp)
if (allow_udp)
try_connect("basp.default-connectivity-udp");
}
......@@ -640,13 +640,13 @@ basp_broker::basp_broker(actor_config& cfg)
behavior basp_broker::make_behavior() {
CAF_LOG_TRACE(CAF_ARG(system().node()));
state.enable_tcp = system().config().middleman_enable_tcp;
state.enable_udp = system().config().middleman_enable_udp;
if (system().config().middleman_enable_automatic_connections) {
state.allow_tcp = !get_or(config(), "middleman.disable-tcp", false);
state.allow_udp = get_or(config(), "middleman.enable-udp", false);
if (get_or(config(), "middleman.enable-automatic-connections", false)) {
CAF_LOG_INFO("enable automatic connections");
// open a random port and store a record for our peers how to
// connect to this broker directly in the configuration server
if (state.enable_tcp) {
if (state.allow_tcp) {
auto res = add_tcp_doorman(uint16_t{0});
if (res) {
auto port = res->second;
......@@ -657,7 +657,7 @@ behavior basp_broker::make_behavior() {
make_message(port, std::move(addrs)));
}
}
if (state.enable_udp) {
if (state.allow_udp) {
auto res = add_udp_datagram_servant(uint16_t{0});
if (res) {
auto port = res->second;
......@@ -668,9 +668,9 @@ behavior basp_broker::make_behavior() {
make_message(port, std::move(addrs)));
}
}
state.enable_automatic_connections = true;
state.automatic_connections = true;
}
auto heartbeat_interval = system().config().middleman_heartbeat_interval;
auto heartbeat_interval = config().middleman_heartbeat_interval;
if (heartbeat_interval > 0) {
CAF_LOG_INFO("enable heartbeat" << CAF_ARG(heartbeat_interval));
send(this, tick_atom::value, heartbeat_interval);
......
......@@ -99,16 +99,16 @@ behavior connection_helper(stateful_actor<connection_helper_state>* self,
}
CAF_LOG_INFO("could not connect to node directly");
} else if (item == "basp.default-connectivity-udp") {
auto& sys = self->system();
// create new broker to try addresses for communication via UDP
if (self->system().config().middleman_detach_utility_actors) {
self->system().middleman().spawn_broker<detached + hidden>(
if (get_or(sys.config(), "middleman.attach-utility-actors", false))
self->system().middleman().spawn_broker<hidden>(
datagram_connection_broker, port, std::move(addresses), b
);
} else {
self->system().middleman().spawn_broker<hidden>(
else
self->system().middleman().spawn_broker<detached + hidden>(
datagram_connection_broker, port, std::move(addresses), b
);
}
} else {
CAF_LOG_INFO("aborted direct connection attempt, unknown item: "
<< CAF_ARG(item));
......
......@@ -292,7 +292,7 @@ void middleman::start() {
for (auto& f : system().config().hook_factories)
hooks_.emplace_back(f(system_));
// Launch backend.
if (system_.config().middleman_detach_multiplexer)
if (!get_or(config(), "middleman.manual-multiplexing", false))
backend_supervisor_ = backend().make_supervisor();
// The only backend that returns a `nullptr` by default is the
// `test_multiplexer` which does not have its own thread but uses the main
......@@ -342,7 +342,7 @@ void middleman::stop() {
}
}
});
if (system_.config().middleman_detach_multiplexer) {
if (!get_or(config(), "middleman.manual-multiplexing", false)) {
backend_supervisor_.reset();
if (thread_.joinable())
thread_.join();
......@@ -354,7 +354,7 @@ void middleman::stop() {
named_brokers_.clear();
scoped_actor self{system(), true};
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_);
destroy(manager_);
}
......@@ -362,7 +362,7 @@ void middleman::stop() {
void middleman::init(actor_system_config& cfg) {
// never detach actors when using the testing multiplexer
if (cfg.middleman_network_backend == atom("testing"))
cfg.middleman_detach_utility_actors = false;
cfg.set("middleman.attach-utility-actors", true);
// add remote group module to config
struct remote_groups : group_module {
public:
......
......@@ -32,9 +32,9 @@ namespace caf {
namespace io {
middleman_actor make_middleman_actor(actor_system& sys, actor db) {
return sys.config().middleman_detach_utility_actors
? sys.spawn<middleman_actor_impl, detached + hidden>(std::move(db))
: sys.spawn<middleman_actor_impl, hidden>(std::move(db));
return get_or(sys.config(), "middleman.attach-utility-actors", false)
? sys.spawn<middleman_actor_impl, hidden>(std::move(db))
: sys.spawn<middleman_actor_impl, detached + hidden>(std::move(db));
}
} // namespace io
......
......@@ -79,17 +79,23 @@ const char* middleman_actor_impl::name() const {
auto middleman_actor_impl::make_behavior() -> behavior_type {
CAF_LOG_TRACE("");
auto tcp_disabled = [=] {
return get_or(config(), "middleman.disable-tcp", false);
};
auto udp_disabled = [=] {
return !get_or(config(), "middleman.enable-udp", false);
};
return {
[=](publish_atom, uint16_t port, strong_actor_ptr& whom, mpi_set& sigs,
std::string& addr, bool reuse) -> put_res {
CAF_LOG_TRACE("");
if (!system().config().middleman_enable_tcp)
if (tcp_disabled())
return make_error(sec::feature_disabled);
return put(port, whom, sigs, addr.c_str(), reuse);
},
[=](open_atom, uint16_t port, std::string& addr, bool reuse) -> put_res {
CAF_LOG_TRACE("");
if (!system().config().middleman_enable_tcp)
if (tcp_disabled())
return make_error(sec::feature_disabled);
strong_actor_ptr whom;
mpi_set sigs;
......@@ -97,7 +103,7 @@ auto middleman_actor_impl::make_behavior() -> behavior_type {
},
[=](connect_atom, std::string& hostname, uint16_t port) -> get_res {
CAF_LOG_TRACE(CAF_ARG(hostname) << CAF_ARG(port));
if (!system().config().middleman_enable_tcp)
if (tcp_disabled())
return make_error(sec::feature_disabled);
auto rp = make_response_promise();
endpoint key{std::move(hostname), port};
......@@ -153,13 +159,13 @@ auto middleman_actor_impl::make_behavior() -> behavior_type {
[=](publish_udp_atom, uint16_t port, strong_actor_ptr& whom,
mpi_set& sigs, std::string& addr, bool reuse) -> put_res {
CAF_LOG_TRACE("");
if (!system().config().middleman_enable_udp)
if (udp_disabled())
return make_error(sec::feature_disabled);
return put_udp(port, whom, sigs, addr.c_str(), reuse);
},
[=](contact_atom, std::string& hostname, uint16_t port) -> get_res {
CAF_LOG_TRACE(CAF_ARG(hostname) << CAF_ARG(port));
if (!system().config().middleman_enable_udp)
if (udp_disabled())
return make_error(sec::feature_disabled);
auto rp = make_response_promise();
endpoint key{std::move(hostname), port};
......
......@@ -129,7 +129,7 @@ public:
.set("middleman.enable-automatic-connections", autoconn)
.set("scheduler.policy", autoconn ? caf::atom("testing")
: caf::atom("stealing"))
.set("middleman.detach-utility-actors", !autoconn)) {
.set("middleman.attach-utility-actors", autoconn)) {
auto& mm = sys.middleman();
mpx_ = dynamic_cast<network::test_multiplexer*>(&mm.backend());
CAF_REQUIRE(mpx_ != nullptr);
......
......@@ -114,12 +114,12 @@ public:
: sys(cfg.load<io::middleman, network::test_multiplexer>()
.set("middleman.enable-automatic-connections", autoconn)
.set("middleman.enable-udp", true)
.set("middleman.enable-tcp", false)
.set("middleman.disable-tcp", true)
.set("scheduler.policy", autoconn || use_test_coordinator
? caf::atom("testing")
: caf::atom("stealing"))
.set("middleman.detach-utility-actors",
!(autoconn || use_test_coordinator))) {
.set("middleman.attach-utility-actors",
autoconn || use_test_coordinator)) {
auto& mm = sys.middleman();
mpx_ = dynamic_cast<network::test_multiplexer*>(&mm.backend());
CAF_REQUIRE(mpx_ != nullptr);
......
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