Commit 60cbcf5f authored by Dominik Charousset's avatar Dominik Charousset

Add RIAC probe module to actor system

parent ca84509a
...@@ -22,3 +22,10 @@ profiling-output-file="/dev/null" ...@@ -22,3 +22,10 @@ profiling-output-file="/dev/null"
enable-automatic-connections=false enable-automatic-connections=false
; accepted alternative: 'asio' (only when compiling CAF with ASIO) ; accepted alternative: 'asio' (only when compiling CAF with ASIO)
network-backend='default' network-backend='default'
; when loading riac::probe
[probe]
; denotes the hostname or IP address to reach the Nexus
nexus-host=""
; denotes the port where the Nexus actor is published
nexus-port=0
...@@ -61,7 +61,10 @@ struct mpi_field_access { ...@@ -61,7 +61,10 @@ struct mpi_field_access {
auto ptr = types.portable_name(0, &typeid(T)); auto ptr = types.portable_name(0, &typeid(T));
if (ptr) if (ptr)
return *ptr; return *ptr;
return "INVALID-TYPE"; std::string result = "<invalid-type[typeid ";
result += typeid(T).name();
result += "]>";
return result;
} }
}; };
...@@ -126,6 +129,7 @@ public: ...@@ -126,6 +129,7 @@ public:
scheduler, scheduler,
middleman, middleman,
opencl_manager, opencl_manager,
riac_probe,
num_ids num_ids
}; };
...@@ -202,9 +206,16 @@ public: ...@@ -202,9 +206,16 @@ public:
bool has_middleman() const; bool has_middleman() const;
/// Returns the middleman instance from the I/O module. /// Returns the middleman instance from the I/O module.
/// @throws `std::logic_error` if the I/O module has not been loaded. /// @throws `std::logic_error` if module is not loaded.
io::middleman& middleman(); io::middleman& middleman();
/// Returns `true` if the RIAC probe module is available, `false` otherwise.
bool has_probe() const;
/// Returns the RIAC probe instance.
/// @throws `std::logic_error` if module is not loaded.
riac::probe& probe();
/// Returns a dummy execution unit that forwards /// Returns a dummy execution unit that forwards
/// everything to the scheduler. /// everything to the scheduler.
scoped_execution_unit* dummy_execution_unit(); scoped_execution_unit* dummy_execution_unit();
...@@ -403,6 +414,7 @@ private: ...@@ -403,6 +414,7 @@ private:
io::middleman* middleman_; io::middleman* middleman_;
scoped_execution_unit dummy_execution_unit_; scoped_execution_unit dummy_execution_unit_;
atom_value backend_name_; atom_value backend_name_;
riac::probe* probe_;
}; };
} // namespace caf } // namespace caf
......
...@@ -123,6 +123,10 @@ public: ...@@ -123,6 +123,10 @@ public:
node_id network_id; node_id network_id;
proxy_registry* network_proxies; proxy_registry* network_proxies;
// Config parameters of RIAC probes.
std::string nexus_host;
uint16_t nexus_port;
private: private:
value_factories_by_name value_factories_by_name_; value_factories_by_name value_factories_by_name_;
value_factories_by_rtti value_factories_by_rtti_; value_factories_by_rtti value_factories_by_rtti_;
......
...@@ -104,16 +104,26 @@ struct header; ...@@ -104,16 +104,26 @@ struct header;
} // namespace io } // namespace io
namespace riac {
class probe;
} // namespace riac
namespace scheduler { namespace scheduler {
class abstract_worker;
class abstract_coordinator; class abstract_worker;
class abstract_coordinator;
} // namespace scheduler } // namespace scheduler
namespace detail { namespace detail {
class disposer;
class message_data; class disposer;
class group_manager; class message_data;
class dynamic_message_data; class group_manager;
class dynamic_message_data;
} // namespace detail } // namespace detail
using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>; using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
......
...@@ -54,7 +54,9 @@ enum class sec : uint8_t { ...@@ -54,7 +54,9 @@ enum class sec : uint8_t {
/// Middleman could not open requested port. /// Middleman could not open requested port.
cannot_open_port, cannot_open_port,
/// A remote spawn failed because the provided types did not match. /// A remote spawn failed because the provided types did not match.
cannot_spawn_actor_from_arguments cannot_spawn_actor_from_arguments,
/// Requested RIAC information about a node that does not exist.
no_such_riac_node
}; };
/// @relates sec /// @relates sec
......
...@@ -66,6 +66,9 @@ actor_system::actor_system(actor_system_config&& cfg) ...@@ -66,6 +66,9 @@ actor_system::actor_system(actor_system_config&& cfg)
auto& mmptr = modules_[module::middleman]; auto& mmptr = modules_[module::middleman];
if (mmptr) if (mmptr)
middleman_ = reinterpret_cast<io::middleman*>(mmptr->subtype_ptr()); middleman_ = reinterpret_cast<io::middleman*>(mmptr->subtype_ptr());
auto& pptr = modules_[module::riac_probe];
if (pptr)
probe_ = reinterpret_cast<riac::probe*>(pptr->subtype_ptr());
auto& sched = modules_[module::scheduler]; auto& sched = modules_[module::scheduler];
using share = scheduler::coordinator<policy::work_sharing>; using share = scheduler::coordinator<policy::work_sharing>;
using steal = scheduler::coordinator<policy::work_stealing>; using steal = scheduler::coordinator<policy::work_stealing>;
...@@ -186,10 +189,21 @@ bool actor_system::has_middleman() const { ...@@ -186,10 +189,21 @@ bool actor_system::has_middleman() const {
io::middleman& actor_system::middleman() { io::middleman& actor_system::middleman() {
if (! middleman_) if (! middleman_)
throw std::logic_error("cannot access middleman: I/O module not loaded"); throw std::logic_error("cannot access middleman: module not loaded");
return *middleman_; return *middleman_;
} }
bool actor_system::has_probe() const {
return probe_ != nullptr;
}
riac::probe& actor_system::probe() {
if (! probe_)
throw std::logic_error("cannot access RIAC probe: module not loaded");
return *probe_;
}
scoped_execution_unit* actor_system::dummy_execution_unit() { scoped_execution_unit* actor_system::dummy_execution_unit() {
return &dummy_execution_unit_; return &dummy_execution_unit_;
} }
......
...@@ -44,6 +44,9 @@ struct type_name_visitor : static_visitor<const char*> { ...@@ -44,6 +44,9 @@ struct type_name_visitor : static_visitor<const char*> {
const char* operator()(const size_t) const { const char* operator()(const size_t) const {
return "an unsigned integer"; return "an unsigned integer";
} }
const char* operator()(const uint16_t) const {
return "an unsigned short integer";
}
const char* operator()(const bool) const { const char* operator()(const bool) const {
return "a boolean"; return "a boolean";
} }
...@@ -79,6 +82,13 @@ bool assign_config_value(size_t& x, int64_t& y) { ...@@ -79,6 +82,13 @@ bool assign_config_value(size_t& x, int64_t& y) {
return true; return true;
} }
bool assign_config_value(uint16_t& x, int64_t& y) {
if (y < 0 || y > std::numeric_limits<uint16_t>::max())
return false;
x = std::move(y);
return true;
}
class actor_system_config_reader { class actor_system_config_reader {
public: public:
using config_value = actor_system_config::config_value; using config_value = actor_system_config::config_value;
...@@ -91,7 +101,7 @@ public: ...@@ -91,7 +101,7 @@ public:
// the INI parser accepts all integers as int64_t // the INI parser accepts all integers as int64_t
using cfg_type = using cfg_type =
typename std::conditional< typename std::conditional<
std::is_same<T, size_t>::value, std::is_integral<T>::value,
int64_t, int64_t,
T T
>::type; >::type;
...@@ -131,6 +141,7 @@ actor_system_config::actor_system_config() { ...@@ -131,6 +141,7 @@ actor_system_config::actor_system_config() {
scheduler_profiling_ms_resolution = 100; scheduler_profiling_ms_resolution = 100;
middleman_network_backend = atom("default"); middleman_network_backend = atom("default");
middleman_enable_automatic_connections = false; middleman_enable_automatic_connections = false;
nexus_port = 0;
} }
actor_system_config::actor_system_config(int argc, char** argv) actor_system_config::actor_system_config(int argc, char** argv)
...@@ -157,7 +168,9 @@ actor_system_config::actor_system_config(int argc, char** argv) ...@@ -157,7 +168,9 @@ actor_system_config::actor_system_config(int argc, char** argv)
scheduler_profiling_output_file) scheduler_profiling_output_file)
.bind("middleman.network-backend", middleman_network_backend) .bind("middleman.network-backend", middleman_network_backend)
.bind("middleman.enable-automatic-connections", .bind("middleman.enable-automatic-connections",
middleman_enable_automatic_connections); middleman_enable_automatic_connections)
.bind("probe.nexus-host", nexus_host)
.bind("probe.nexus-port", nexus_port);
detail::parse_ini(ini, consumer, std::cerr); detail::parse_ini(ini, consumer, std::cerr);
} }
} }
...@@ -187,6 +200,12 @@ actor_system_config::actor_system_config(int argc, char** argv) ...@@ -187,6 +200,12 @@ actor_system_config::actor_system_config(int argc, char** argv)
{"caf#middleman.enable-automatic-connections", {"caf#middleman.enable-automatic-connections",
"enables or disables automatic connection management (off per default)", "enables or disables automatic connection management (off per default)",
middleman_enable_automatic_connections}, middleman_enable_automatic_connections},
{"caf#probe.nexus-host",
"sets the hostname or IP address for connecting to the Nexus",
nexus_host},
{"caf#probe.nexus-port",
"sets the port for connecting to the Nexus",
nexus_port},
{"caf-dump-config", "print config in INI format to stdout"}, {"caf-dump-config", "print config in INI format to stdout"},
{"caf-help", "print this text"} {"caf-help", "print this text"}
}, nullptr, true); }, nullptr, true);
......
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