Commit 20047855 authored by Dominik Charousset's avatar Dominik Charousset

Receive BASP config directly from middleman

parent 80f74146
......@@ -22,6 +22,7 @@
#include <atomic>
#include <string>
#include <memory>
#include <functional>
#include <type_traits>
#include <unordered_map>
......@@ -57,22 +58,67 @@ public:
using error_renderers = std::unordered_map<atom_value, error_renderer>;
/// A variant type for config parameters.
using config_value = variant<std::string, double, int64_t, bool, atom_value>;
/// Helper class to generate config readers for different input types.
class option {
public:
using config_reader_sink = std::function<void (size_t, config_value&)>;
option(const char* category, const char* name, const char* explanation);
virtual ~option();
inline const char* name() const {
return name_;
}
inline const char* category() const {
return category_;
}
inline const char* explanation() const {
return explanation_;
}
virtual std::string to_string() const = 0;
virtual config_reader_sink to_sink() const = 0;
virtual message::cli_arg to_cli_arg() const = 0;
protected:
const char* category_;
const char* name_;
const char* explanation_;
};
using option_ptr = std::unique_ptr<option>;
using options_vector = std::vector<option_ptr>;
actor_system_config();
actor_system_config(int argc, char** argv);
actor_system_config& add_actor_factory(std::string name, actor_factory fun);
/// Allows others to spawn actors of type `T`
/// dynamically by using `name` as identifier.
template <class T, class... Ts>
actor_system_config& add_actor_type(std::string name) {
return add_actor_factory(std::move(name), make_actor_factory<T, Ts...>());
}
/// Allows others to spawn actors implemented by function `f`
/// dynamically by using `name` as identifier.
template <class F>
actor_system_config& add_actor_type(std::string name, F f) {
return add_actor_factory(std::move(name), make_actor_factory(std::move(f)));
}
/// Adds message type `T` with runtime type info `name`.
template <class T>
actor_system_config& add_message_type(std::string name) {
static_assert(std::is_empty<T>::value
......@@ -89,10 +135,8 @@ public:
return *this;
}
/**
* Enables the actor system to convert errors of this error category
* to human-readable strings via `renderer`.
*/
/// Enables the actor system to convert errors of this error category
/// to human-readable strings via `renderer`.
actor_system_config& add_error_category(atom_value category,
error_renderer renderer);
......@@ -115,6 +159,7 @@ public:
return add_error_category(category, f);
}
/// 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* {
......@@ -126,8 +171,8 @@ public:
/// Stores CLI arguments that were not consumed by CAF.
message args_remainder;
/// Sets the parameter `name` to `val`.
using config_value = variant<std::string, double, int64_t, bool, atom_value>;
/// Sets a config by using its INI name `config_name` to `config_value`.
actor_system_config& set(const char* config_name, config_value config_value);
// Config parameters of scheduler.
atom_value scheduler_policy;
......@@ -141,7 +186,7 @@ public:
atom_value middleman_network_backend;
bool middleman_enable_automatic_connections;
size_t middleman_max_consecutive_reads;
size_t middleman_basp_heartbeat_interval;
size_t middleman_heartbeat_interval;
// System parameters that are set while initializing modules.
node_id network_id;
......@@ -158,6 +203,7 @@ private:
actor_factories actor_factories_;
module_factories module_factories_;
error_renderers error_renderers_;
options_vector options_;
};
} // namespace caf
......
This diff is collapsed.
......@@ -135,8 +135,6 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
}
};
/// A broker implementation for the Binary Actor System Protocol (BASP).
class basp_broker : public stateful_actor<basp_broker_state, broker> {
public:
......
......@@ -261,6 +261,17 @@ public:
return new impl(sys);
}
/// Returns the heartbeat interval in milliseconds.
inline size_t heartbeat_interval() const {
return heartbeat_interval_;
}
/// Retruns 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:
middleman(actor_system& ref);
......@@ -315,8 +326,10 @@ private:
hook_uptr hooks_;
// actor offering asyncronous IO by managing this singleton instance
middleman_actor manager_;
// configure parameters
size_t basp_heartbeat_interval_;
// 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
......
......@@ -444,10 +444,23 @@ basp_broker::basp_broker(actor_config& cfg)
behavior basp_broker::make_behavior() {
CAF_LOG_TRACE("");
// TODO: query this config from middleman directly
// ask the configuration server whether we should open a default port
auto config_server = system().registry().get(atom("ConfigServ"));
send(config_server, get_atom::value, "middleman.enable-automatic-connections");
if (system().middleman().enable_automatic_connections()) {
printf("enable automatic connections\n");
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
auto port = add_tcp_doorman(uint16_t{0});
auto addrs = network::interfaces::list_addresses(false);
auto config_server = system().registry().get(atom("ConfigServ"));
send(config_server, put_atom::value, "basp.default-connectivity",
make_message(port.second, std::move(addrs)));
state.enable_automatic_connections = true;
}
auto heartbeat_interval = system().middleman().heartbeat_interval();
if (heartbeat_interval > 0) {
CAF_LOG_INFO("enable heartbeat" << CAF_ARG(heartbeat_interval));
send(this, tick_atom::value, heartbeat_interval);
}
return {
// received from underlying broker implementation
[=](new_data_msg& msg) {
......@@ -639,22 +652,6 @@ behavior basp_broker::make_behavior() {
}
return {};
},
[=](ok_atom, const std::string& key, message& value) {
if (key == "middleman.enable-automatic-connections") {
value.apply([&](bool enabled) {
if (! enabled)
return;
CAF_LOG_INFO("enable automatic connection");
// open a random port and store a record for others
// how to connect to this port in the configuration server
auto port = add_tcp_doorman(uint16_t{0});
auto addrs = network::interfaces::list_addresses(false);
send(config_server, put_atom::value, "basp.default-connectivity",
make_message(port.second, std::move(addrs)));
state.enable_automatic_connections = true;
});
}
},
[=](get_atom, const node_id& x)
-> std::tuple<node_id, std::string, uint16_t> {
std::string addr;
......
......@@ -100,7 +100,10 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
return new impl(sys);
}
middleman::middleman(actor_system& sys) : system_(sys) {
middleman::middleman(actor_system& sys)
: system_(sys),
heartbeat_interval_(0),
enable_automatic_connections_(false) {
// nop
}
......@@ -286,10 +289,6 @@ void middleman::start() {
backend().thread_id(thread_.get_id());
}
auto basp = named_broker<basp_broker>(atom("BASP"));
if (basp_heartbeat_interval_ > 0) {
CAF_LOG_INFO("enable basp-heartbeat: " << CAF_ARG(basp_heartbeat_interval_));
anon_send(basp, tick_atom::value, basp_heartbeat_interval_);
}
manager_ = make_middleman_actor(system(), basp);
}
......@@ -347,7 +346,9 @@ void middleman::init(actor_system_config& cfg) {
// set scheduling parameters for multiplexer
backend().max_throughput(cfg.scheduler_max_throughput);
backend().max_consecutive_reads(cfg.middleman_max_consecutive_reads);
basp_heartbeat_interval_ = cfg.middleman_basp_heartbeat_interval;
// set options relevant to BASP
heartbeat_interval_ = cfg.middleman_heartbeat_interval;
enable_automatic_connections_ = cfg.middleman_enable_automatic_connections;
}
actor_system::module::id_t middleman::id() const {
......
......@@ -97,8 +97,10 @@ string hexstr(const buffer& buf) {
class fixture {
public:
fixture() : system(actor_system_config{}
.load<io::middleman, network::test_multiplexer>()) {
fixture(bool autoconn = false)
: system(actor_system_config{}
.load<io::middleman, network::test_multiplexer>()
.set("middleman.enable-automatic-connections", autoconn)) {
auto& mm = system.middleman();
mpx_ = dynamic_cast<network::test_multiplexer*>(&mm.backend());
CAF_REQUIRE(mpx_ != nullptr);
......@@ -427,6 +429,13 @@ private:
actor_registry* registry_;
};
class autoconn_enabled_fixture : public fixture {
public:
autoconn_enabled_fixture() : fixture(true) {
// nop
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(basp_tests, fixture)
......@@ -669,8 +678,9 @@ CAF_TEST(actor_serialize_and_deserialize) {
}
CAF_TEST(indirect_connections) {
// jupiter [remote hdl 0] -> mars [remote hdl 1] -> earth [this_node]
// (this node receives a message from jupiter via mars and responds via mars)
// jupiter [remote hdl 0] -> mars [remote hdl 1] -> earth [this_node];
// this node receives a message from jupiter via mars and responds via mars
// and any ad-hoc automatic connection requests are ignored
CAF_MESSAGE("self: " << to_string(self()->address()));
auto ax = accept_handle::from_int(4242);
mpx()->provide_acceptor(4242, ax);
......@@ -715,11 +725,15 @@ CAF_TEST(indirect_connections) {
make_message("hello from earth!"));
}
CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(basp_tests_with_autoconn, autoconn_enabled_fixture)
CAF_TEST(automatic_connection) {
// this tells our BASP broker to enable the automatic connection feature
anon_send(aut(), ok_atom::value,
"middleman.enable-automatic-connections", make_message(true));
mpx()->exec_runnable(); // process publish message in basp_broker
//anon_send(aut(), ok_atom::value,
// "middleman.enable-automatic-connections", make_message(true));
//mpx()->exec_runnable(); // process publish message in basp_broker
// jupiter [remote hdl 0] -> mars [remote hdl 1] -> earth [this_node]
// (this node receives a message from jupiter via mars and responds via mars,
// but then also establishes a connection to jupiter directly)
......
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