Commit 80da1c4f authored by Dominik Charousset's avatar Dominik Charousset

Add group modules via system config, relates #480

parent 89995905
......@@ -55,6 +55,7 @@ set (LIBCAF_CORE_SRCS
src/get_root_uuid.cpp
src/group.cpp
src/group_manager.cpp
src/group_module.cpp
src/invoke_result_visitor.cpp
src/match_case.cpp
src/merged_tuple.cpp
......
......@@ -34,66 +34,49 @@ namespace caf {
/// A multicast group.
class abstract_group : public ref_counted, public abstract_channel {
public:
// -- member types -----------------------------------------------------------
friend class local_actor;
friend class subscription;
friend class detail::group_manager;
~abstract_group();
/// Interface for user-defined multicast implementations.
class module {
public:
module(actor_system& sys, std::string module_name);
virtual ~module();
/// Stops all groups from this module.
virtual void stop() = 0;
inline actor_system& system() const {
return system_;
}
/// Returns the name of this module implementation.
/// @threadsafe
const std::string& name() const;
/// Returns a pointer to the group associated with the name `group_name`.
/// @threadsafe
virtual group get(const std::string& group_name) = 0;
// -- constructors, destructors, and assignment operators --------------------
virtual group load(deserializer& source) = 0;
private:
actor_system& system_;
std::string name_;
};
~abstract_group();
using module_ptr = module*;
using unique_module_ptr = std::unique_ptr<module>;
// -- pure virtual member functions ------------------------------------------
/// Serialize this group to `sink`.
virtual error save(serializer& sink) const = 0;
/// Returns a string representation of the group identifier, e.g.,
/// "224.0.0.1" for IPv4 multicast or a user-defined string for local groups.
const std::string& identifier() const;
module_ptr get_module() const;
/// Returns the name of the module.
const std::string& module_name() const;
/// Subscribes `who` to this group and returns `true` on success
/// or `false` if `who` is already subscribed.
virtual bool subscribe(strong_actor_ptr who) = 0;
/// Unsubscribes `who` from this group.
virtual void unsubscribe(const actor_control_block* who) = 0;
/// Stops any background actors or threads and IO handles.
virtual void stop() = 0;
inline actor_system& system() {
// -- observers --------------------------------------------------------------
/// Returns the parent module.
inline group_module& module() const {
return parent_;
}
/// Returns the hosting system.
inline actor_system& system() const {
return system_;
}
/// Returns a string representation of the group identifier, e.g.,
/// "224.0.0.1" for IPv4 multicast or a user-defined string for local groups.
const std::string& identifier() const {
return identifier_;
}
/// @cond PRIVATE
template <class... Ts>
......@@ -104,16 +87,13 @@ public:
make_message(std::forward<Ts>(xs)...), ctx);
}
virtual void unsubscribe(const actor_control_block* who) = 0;
/// @endcond
protected:
abstract_group(actor_system& sys, module_ptr module,
std::string group_id, const node_id& nid);
abstract_group(group_module& parent, std::string id, node_id origin);
actor_system& system_;
module_ptr module_;
group_module& parent_;
std::string identifier_;
node_id origin_;
};
......
......@@ -71,6 +71,10 @@ public:
using option_vector = std::vector<option_ptr>;
using group_module_factory = std::function<group_module* ()>;
using group_module_factory_vector = std::vector<group_module_factory>;
// -- nested classes ---------------------------------------------------------
class opt_group {
......@@ -223,7 +227,7 @@ public:
/// 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.
// -- config parameters of the scheduler -------------------------------------
atom_value scheduler_policy;
size_t scheduler_max_threads;
size_t scheduler_max_throughput;
......@@ -231,7 +235,8 @@ public:
size_t scheduler_profiling_ms_resolution;
std::string scheduler_profiling_output_file;
// Config parameters for work-stealing strategy
// -- config parameters for work-stealing ------------------------------------
size_t work_stealing_aggressive_poll_attempts;
size_t work_stealing_aggressive_steal_interval;
size_t work_stealing_moderate_poll_attempts;
......@@ -240,23 +245,36 @@ public:
size_t work_stealing_relaxed_steal_interval;
size_t work_stealing_relaxed_sleep_duration_us;
// Config parameters of middleman.
// -- config parameters of the middleman -------------------------------------
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;
// Config parameters of the OpenCL module.
// -- config parameters of the OpenCL module ---------------------------------
std::string opencl_device_ids;
// -- factories --------------------------------------------------------------
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;
hook_factory_vector hook_factories;
group_module_factory_vector group_module_factories;
// -- run-time type information ----------------------------------------------
portable_name_map type_names_by_rtti;
// -- rendering of user-defined types ----------------------------------------
error_renderer_map error_renderers;
// -- utility for caf-run ----------------------------------------------------
int (*slave_mode_fun)(actor_system&, const actor_system_config&);
......
......@@ -43,6 +43,7 @@ template <class... Ts>
class type_erased_tuple_view : public type_erased_tuple {
public:
// -- member types -----------------------------------------------------------
template <size_t X>
using num_token = std::integral_constant<size_t, X>;
......
......@@ -67,6 +67,7 @@ class local_actor;
class actor_config;
class actor_system;
class deserializer;
class group_module;
class message_view;
class scoped_actor;
class abstract_actor;
......
......@@ -23,9 +23,12 @@
#include <map>
#include <mutex>
#include <thread>
#include <unordered_map>
#include "caf/fwd.hpp"
#include "caf/optional.hpp"
#include "caf/expected.hpp"
#include "caf/group_module.hpp"
#include "caf/abstract_group.hpp"
#include "caf/detail/shared_spinlock.hpp"
......@@ -35,43 +38,57 @@ namespace caf {
class group_manager {
public:
// -- friends ----------------------------------------------------------------
friend class actor_system;
void start();
// -- member types -----------------------------------------------------------
void stop();
using modules_map = std::unordered_map<std::string,
std::unique_ptr<group_module>>;
// -- constructors, destructors, and assignment operators --------------------
~group_manager();
/// Get a pointer to the group associated with
// -- observers --------------------------------------------------------------
/// Get a handle to the group associated with
/// `identifier` from the module `mod_name`.
/// @threadsafe
expected<group> get(const std::string& module_name,
const std::string& group_identifier);
const std::string& group_identifier) const;
/// Get a pointer to the group associated with
/// `identifier` from the module `local`.
/// @threadsafe
group get_local(const std::string& group_identifier);
group get_local(const std::string& identifier) const;
/// Returns an anonymous group.
/// Each calls to this member function returns a new instance
/// of an anonymous group. Anonymous groups can be used whenever
/// a set of actors wants to communicate using an exclusive channel.
group anonymous();
void add_module(abstract_group::unique_module_ptr);
group anonymous() const;
abstract_group::module_ptr get_module(const std::string& module_name);
/// Returns the module named `name` if it exists, otherwise `none`.
optional<group_module&> get_module(const std::string& name) const;
private:
using modules_map = std::map<std::string, abstract_group::unique_module_ptr>;
// -- constructors, destructors, and assignment operators --------------------
group_manager(actor_system& sys);
// -- member functions required by actor_system ------------------------------
void init(actor_system_config& cfg);
void start();
void stop();
// -- data members -----------------------------------------------------------
modules_map mmap_;
std::mutex mmap_mtx_;
actor_system& system_;
};
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_GROUP_MODULE_HPP
#define CAF_GROUP_MODULE_HPP
#include <string>
#include <memory>
#include "caf/fwd.hpp"
#include "caf/actor_addr.hpp"
#include "caf/attachable.hpp"
#include "caf/ref_counted.hpp"
#include "caf/abstract_channel.hpp"
namespace caf {
/// Interface for user-defined multicast implementations.
class group_module {
public:
// -- constructors, destructors, and assignment operators --------------------
group_module(actor_system& sys, std::string module_name);
virtual ~group_module();
// -- pure virtual member functions ------------------------------------------
/// Stops all groups from this module.
virtual void stop() = 0;
/// Returns a pointer to the group associated with the name `group_name`.
/// @threadsafe
virtual group get(const std::string& group_name) = 0;
/// Loads a group of this module from `source` and stores it in `storage`.
virtual error load(deserializer& source, group& storage) = 0;
// -- observers --------------------------------------------------------------
/// Returns the hosting actor system.
inline actor_system& system() const {
return system_;
}
/// Returns the name of this module implementation.
inline const std::string& name() const {
return name_;
}
private:
actor_system& system_;
std::string name_;
};
} // namespace caf
#endif // CAF_GROUP_MODULE_HPP
......@@ -22,49 +22,18 @@
#include "caf/group.hpp"
#include "caf/message.hpp"
#include "caf/actor_cast.hpp"
#include "caf/group_module.hpp"
#include "caf/group_manager.hpp"
#include "caf/detail/shared_spinlock.hpp"
namespace caf {
abstract_group::module::module(actor_system& sys, std::string mname)
: system_(sys),
name_(std::move(mname)) {
// nop
}
void abstract_group::module::stop() {
// nop
}
const std::string& abstract_group::module::name() const {
return name_;
}
abstract_group::abstract_group(actor_system& sys,
abstract_group::module_ptr mod,
std::string id, const node_id& nid)
abstract_group::abstract_group(group_module& mod, std::string id, node_id nid)
: abstract_channel(abstract_channel::is_abstract_group_flag),
system_(sys),
module_(mod),
system_(mod.system()),
parent_(mod),
identifier_(std::move(id)),
origin_(nid) {
// nop
}
const std::string& abstract_group::identifier() const {
return identifier_;
}
abstract_group::module_ptr abstract_group::get_module() const {
return module_;
}
const std::string& abstract_group::module_name() const {
return get_module()->name();
}
abstract_group::module::~module() {
origin_(std::move(nid)) {
// nop
}
......
......@@ -242,6 +242,7 @@ actor_system::actor_system(actor_system_config& cfg)
for (auto& mod : modules_)
if (mod)
mod->init(cfg);
groups_.init(cfg);
// 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));
......
......@@ -61,7 +61,7 @@ void serialize(serializer& sink, const group& x, const unsigned int) {
std::string dummy;
sink << dummy;
} else {
sink << ptr->module_name();
sink << ptr->module().name();
ptr->save(sink);
}
}
......@@ -80,13 +80,13 @@ void serialize(deserializer& source, group& x, const unsigned int) {
if (! mod)
CAF_RAISE_ERROR("Cannot deserialize a group for unknown module: "
+ module_name);
x = mod->load(source);
mod->load(source, x);
}
std::string to_string(const group& x) {
if (x == invalid_group)
return "<invalid-group>";
std::string result = x->get_module()->name();
std::string result = x->module().name();
result += "/";
result += x->identifier();
return result;
......
......@@ -121,8 +121,8 @@ public:
return broker_;
}
local_group(actor_system& sys, optional<actor> local_broker,
local_group_module* mod, std::string id, const node_id& nid);
local_group(local_group_module& mod, std::string id, node_id nid,
optional<actor> local_broker);
~local_group();
......@@ -236,11 +236,9 @@ private:
class local_group_proxy : public local_group {
public:
using super = local_group;
template <class... Ts>
local_group_proxy(actor_system& sys, actor remote_broker, Ts&&... xs)
: super(sys, std::move(remote_broker), std::forward<Ts>(xs)...),
local_group_proxy(actor_system& sys, actor remote_broker,
local_group_module& mod, std::string id, node_id nid)
: local_group(mod, std::move(id), std::move(nid), std::move(remote_broker)),
proxy_broker_{sys.spawn<proxy_broker, hidden>(this)},
monitor_{sys.spawn<hidden>(broker_monitor_actor, this)} {
// nop
......@@ -323,11 +321,9 @@ behavior proxy_broker::make_behavior() {
};
}
class local_group_module : public abstract_group::module {
class local_group_module : public group_module {
public:
using super = abstract_group::module;
local_group_module(actor_system& sys) : super(sys, "local") {
local_group_module(actor_system& sys) : group_module(sys, "local") {
CAF_LOG_TRACE("");
}
......@@ -337,8 +333,8 @@ public:
auto i = instances_.find(identifier);
if (i != instances_.end())
return {i->second};
auto tmp = make_counted<local_group>(system(), none, this, identifier,
system().node());
auto tmp = make_counted<local_group>(*this, identifier,
system().node(), none);
upgrade_to_unique_guard uguard(guard);
auto p = instances_.emplace(identifier, tmp);
auto result = p.first->second;
......@@ -349,29 +345,36 @@ public:
return {result};
}
group load(deserializer& source) override {
error load(deserializer& source, group& storage) override {
CAF_LOG_TRACE("");
// deserialize identifier and broker
std::string identifier;
strong_actor_ptr broker_ptr;
source >> identifier >> broker_ptr;
CAF_LOG_DEBUG(CAF_ARG(identifier) << CAF_ARG(broker_ptr));
if (! broker_ptr)
return invalid_group;
if (! broker_ptr) {
storage = invalid_group;
return {};
}
auto broker = actor_cast<actor>(broker_ptr);
if (broker->node() == system().node())
return this->get(identifier);
if (broker->node() == system().node()) {
storage = this->get(identifier);
return {};
}
upgrade_guard guard(proxies_mtx_);
auto i = proxies_.find(broker);
if (i != proxies_.end())
return {i->second};
if (i != proxies_.end()) {
storage = group{i->second};
return {};
}
local_group_ptr tmp = make_counted<local_group_proxy>(system(), broker,
this, identifier,
*this, identifier,
broker->node());
upgrade_to_unique_guard uguard(guard);
auto p = proxies_.emplace(broker, tmp);
// someone might preempt us
return {p.first->second};
storage = group{p.first->second};
return {};
}
error save(const local_group* ptr, serializer& sink) const {
......@@ -405,11 +408,10 @@ private:
std::map<actor, local_group_ptr> proxies_;
};
local_group::local_group(actor_system& sys, optional<actor> lb,
local_group_module* mod, std::string id,
const node_id& nid)
: abstract_group(sys, mod, std::move(id), nid),
broker_(lb ? *lb : sys.spawn<local_broker, hidden>(this)) {
local_group::local_group(local_group_module& mod, std::string id, node_id nid,
optional<actor> lb)
: abstract_group(mod, std::move(id), std::move(nid)),
broker_(lb ? *lb : mod.system().spawn<local_broker, hidden>(this)) {
CAF_LOG_TRACE(CAF_ARG(id) << CAF_ARG(nid));
}
......@@ -421,7 +423,7 @@ error local_group::save(serializer& sink) const {
CAF_LOG_TRACE("");
// this cast is safe, because the only available constructor accepts
// local_group_module* as module pointer
static_cast<local_group_module*>(module_)->save(this, sink);
static_cast<local_group_module&>(parent_).save(this, sink);
// TODO: refactor after visit API is in place (#470)
return {};
}
......@@ -430,20 +432,25 @@ std::atomic<size_t> s_ad_hoc_id;
} // namespace <anonymous>
void group_manager::init(actor_system_config& cfg) {
CAF_LOG_TRACE("");
using ptr_type = std::unique_ptr<group_module>;
mmap_.emplace("local", ptr_type{new local_group_module(system_)});
for (auto& fac : cfg.group_module_factories) {
ptr_type ptr{fac()};
std::string name = ptr->name();
mmap_.emplace(std::move(name), std::move(ptr));
}
}
void group_manager::start() {
CAF_LOG_TRACE("");
}
void group_manager::stop() {
CAF_LOG_TRACE("");
modules_map mm;
{ // critical section
std::lock_guard<std::mutex> guard(mmap_mtx_);
mm.swap(mmap_);
}
for (auto& kvp : mm) {
for (auto& kvp : mmap_)
kvp.second->stop();
}
}
group_manager::~group_manager() {
......@@ -451,12 +458,10 @@ group_manager::~group_manager() {
}
group_manager::group_manager(actor_system& sys) : system_(sys) {
CAF_LOG_TRACE("");
abstract_group::unique_module_ptr ptr{new local_group_module(sys)};
mmap_.emplace(std::string("local"), std::move(ptr));
// nop
}
group group_manager::anonymous() {
group group_manager::anonymous() const {
CAF_LOG_TRACE("");
std::string id = "__#";
id += std::to_string(++s_ad_hoc_id);
......@@ -464,44 +469,27 @@ group group_manager::anonymous() {
}
expected<group> group_manager::get(const std::string& module_name,
const std::string& group_identifier) {
const std::string& group_identifier) const {
CAF_LOG_TRACE(CAF_ARG(module_name) << CAF_ARG(group_identifier));
auto mod = get_module(module_name);
if (mod) {
if (mod)
return mod->get(group_identifier);
}
std::string error_msg = "no module named \"";
error_msg += module_name;
error_msg += "\" found";
return make_error(sec::no_such_group_module, std::move(error_msg));
}
group group_manager::get_local(const std::string& group_identifier) {
// guaranteed to never return an error
return *get("local", group_identifier);
}
void group_manager::add_module(std::unique_ptr<abstract_group::module> mptr) {
CAF_LOG_TRACE("");
if (! mptr)
return;
auto& mname = mptr->name();
{ // lifetime scope of guard
std::lock_guard<std::mutex> guard(mmap_mtx_);
if (mmap_.emplace(mname, std::move(mptr)).second)
return; // success; don't throw
}
std::string error_msg = "module name \"";
error_msg += mname;
error_msg += "\" already defined";
CAF_RAISE_ERROR(std::move(error_msg));
optional<group_module&> group_manager::get_module(const std::string& x) const {
auto i = mmap_.find(x);
if (i != mmap_.end())
return *(i->second);
return none;
}
abstract_group::module* group_manager::get_module(const std::string& mname) {
CAF_LOG_TRACE(CAF_ARG(mname));
std::lock_guard<std::mutex> guard(mmap_mtx_);
auto i = mmap_.find(mname);
return (i != mmap_.end()) ? i->second.get() : nullptr;
group group_manager::get_local(const std::string& group_identifier) const {
// guaranteed to never return an error
return *get("local", group_identifier);
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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/group_module.hpp"
namespace caf {
group_module::group_module(actor_system& sys, std::string mname)
: system_(sys),
name_(std::move(mname)) {
// nop
}
group_module::~group_module() {
// nop
}
} // namespace caf
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