Commit cf9a69e3 authored by Dominik Charousset's avatar Dominik Charousset

Re-implement the local group module

parent c9cc5d42
...@@ -104,53 +104,42 @@ void run_client(actor_system& system, const config& cfg) { ...@@ -104,53 +104,42 @@ void run_client(actor_system& system, const config& cfg) {
return; return;
} }
} }
std::cout << "*** starting client, type '/help' for a list of commands\n";
auto client_actor = system.spawn(client, name); auto client_actor = system.spawn(client, name);
for (auto& uri : cfg.group_uris) { for (auto& uri : cfg.group_uris) {
auto tmp = system.groups().get(uri); if (auto grp = system.groups().get(uri)) {
if (tmp) std::cout << "*** join " << to_string(grp) << '\n';
anon_send(client_actor, join_atom_v, std::move(*tmp)); anon_send(client_actor, join_atom_v, std::move(*grp));
else } else {
std::cerr << R"(*** failed to parse ")" << uri << R"(" as group URI: )" std::cerr << R"(*** failed to parse ")" << uri << R"(" as group URI: )"
<< to_string(tmp.error()) << std::endl; << to_string(grp.error()) << std::endl;
}
} }
std::cout << "*** starting client, type '/help' for a list of commands\n";
std::istream_iterator<line> eof; std::istream_iterator<line> eof;
std::vector<std::string> words; std::vector<std::string> words;
for (std::istream_iterator<line> i{std::cin}; i != eof; ++i) { for (std::istream_iterator<line> i{std::cin}; i != eof; ++i) {
auto send_input = [&] { if (i->str.empty()) {
if (!i->str.empty()) // Ignore empty lines.
anon_send(client_actor, broadcast_atom_v, i->str); } else if (i->str[0] == '/') {
}; words.clear();
words.clear(); split(words, i->str, is_any_of(" "));
split(words, i->str, is_any_of(" ")); if (words.size() == 3 && words[0] == "/join") {
message_handler f{ if (auto grp = system.groups().get(words[1], words[2]))
[&](const std::string& cmd, const std::string& mod, anon_send(client_actor, join_atom_v, *grp);
const std::string& id) { else
if (cmd == "/join") { std::cerr << "*** failed to join group: " << to_string(grp.error())
auto grp = system.groups().get(mod, id); << std::endl;
if (grp) } else if (words.size() == 1 && words[0] == "quit") {
anon_send(client_actor, join_atom_v, *grp); std::cin.setstate(std::ios_base::eofbit);
} else { } else {
send_input(); std::cout << "*** available commands:\n"
} " /join <module> <group> join a new chat channel\n"
}, " /quit quit the program\n"
[&](const std::string& cmd) { " /help print this text\n";
if (cmd == "/quit") { }
std::cin.setstate(std::ios_base::eofbit); } else {
} else if (cmd[0] == '/') { anon_send(client_actor, broadcast_atom_v, i->str);
std::cout << "*** available commands:\n" }
" /join <module> <group> join a new chat channel\n"
" /quit quit the program\n"
" /help print this text\n";
} else {
send_input();
}
},
};
auto msg = message_builder(words.begin(), words.end()).move_to_message();
auto res = f(msg);
if (!res)
send_input();
} }
// force actor to quit // force actor to quit
anon_send_exit(client_actor, exit_reason::user_shutdown); anon_send_exit(client_actor, exit_reason::user_shutdown);
......
...@@ -87,6 +87,7 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS} ...@@ -87,6 +87,7 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS}
src/detail/get_root_uuid.cpp src/detail/get_root_uuid.cpp
src/detail/glob_match.cpp src/detail/glob_match.cpp
src/detail/invoke_result_visitor.cpp src/detail/invoke_result_visitor.cpp
src/detail/local_group_module.cpp
src/detail/message_builder_element.cpp src/detail/message_builder_element.cpp
src/detail/message_data.cpp src/detail/message_data.cpp
src/detail/meta_object.cpp src/detail/meta_object.cpp
...@@ -248,6 +249,7 @@ caf_add_test_suites(caf-core-test ...@@ -248,6 +249,7 @@ caf_add_test_suites(caf-core-test
detail.encode_base64 detail.encode_base64
detail.ieee_754 detail.ieee_754
detail.limited_vector detail.limited_vector
detail.local_group_module
detail.meta_object detail.meta_object
detail.parse detail.parse
detail.parser.read_bool detail.parser.read_bool
...@@ -290,7 +292,6 @@ caf_add_test_suites(caf-core-test ...@@ -290,7 +292,6 @@ caf_add_test_suites(caf-core-test
ipv6_endpoint ipv6_endpoint
ipv6_subnet ipv6_subnet
load_inspector load_inspector
local_group
logger logger
mailbox_element mailbox_element
message message
......
...@@ -60,13 +60,16 @@ public: ...@@ -60,13 +60,16 @@ public:
// -- observers -------------------------------------------------------------- // -- observers --------------------------------------------------------------
/// Returns the hosting system. /// Returns the hosting system.
actor_system& system() const { actor_system& system() const noexcept;
return system_;
}
/// Returns the parent module. /// Returns the parent module.
group_module& module() const { group_module& module() const noexcept {
return parent_; return *parent_;
}
/// Returns the origin node of the group if applicable.
node_id origin() const noexcept {
return origin_;
} }
/// Returns a string representation of the group identifier, e.g., /// Returns a string representation of the group identifier, e.g.,
...@@ -75,19 +78,18 @@ public: ...@@ -75,19 +78,18 @@ public:
return identifier_; return identifier_;
} }
/// @private /// Returns a human-readable string representation of the group ID.
const actor& dispatcher() { virtual std::string to_string() const;
return dispatcher_;
} /// Returns the intermediary actor for the group if applicable.
virtual actor intermediary() const noexcept;
protected: protected:
abstract_group(group_module& mod, std::string id, node_id nid); abstract_group(group_module_ptr mod, std::string id, node_id nid);
actor_system& system_; group_module_ptr parent_;
group_module& parent_;
std::string identifier_;
node_id origin_; node_id origin_;
actor dispatcher_; std::string identifier_;
}; };
/// A smart pointer type that manages instances of {@link group}. /// A smart pointer type that manages instances of {@link group}.
......
...@@ -5,7 +5,7 @@ ...@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework * * | |___ / ___ \| _| Framework *
* \____/_/ \_|_| * * \____/_/ \_|_| *
* * * *
* Copyright 2011-2018 Dominik Charousset * * Copyright 2011-2020 Dominik Charousset *
* * * *
* Distributed under the terms and conditions of the BSD 3-Clause License or * * 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 * * (at your option) under the terms and conditions of the Boost Software *
...@@ -16,71 +16,88 @@ ...@@ -16,71 +16,88 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#define CAF_SUITE local_group #pragma once
#include "caf/all.hpp" #include <functional>
#include <mutex>
#include <set>
#include <unordered_map>
#include "core-test.hpp" #include "caf/abstract_group.hpp"
#include "caf/group_module.hpp"
#include <array> namespace caf::detail {
#include <chrono>
#include <algorithm>
using namespace caf; /// Simple group implementation that allows arbitrary strings as group names.
/// However, the group names must use a string representation of the local node
/// ID as prefix. Each group instance spins up an intermediary actor to enable
/// remote access to the group. The default module is used internally by the
/// "local" as well as the "remote" module.
class local_group_module : public group_module {
public:
using super = group_module;
namespace { struct intermediary_actor_state {
static inline const char* name = "caf.detail.group-intermediary";
using testee_if intermediary_actor_state(event_based_actor* self, abstract_group_ptr gptr);
= typed_actor<replies_to<get_atom>::with<int>, reacts_to<put_atom, int>>;
struct testee_state { behavior make_behavior();
int x = 0;
}; event_based_actor* self;
behavior testee_impl(stateful_actor<testee_state>* self) { abstract_group_ptr gptr;
auto subscriptions = self->joined_groups();
return {
[=](put_atom, int x) {
self->state.x = x;
},
[=](get_atom) {
return self->state.x;
}
}; };
}
struct fixture { /// A group intermediary enables remote actors to join and leave groups on
actor_system_config config; /// this endpoint as well as sending message to it.
actor_system system{config}; using intermediary_actor = stateful_actor<intermediary_actor_state>;
scoped_actor self{system};
}; /// Implementation of the group interface for instances of this module.
class impl : public abstract_group {
public:
friend local_group_module;
using super = abstract_group;
using subscriber_set = std::set<strong_actor_ptr, std::less<>>;
impl(group_module_ptr mod, std::string id);
~impl() override;
void enqueue(strong_actor_ptr sender, message_id mid, message content,
execution_unit* host) override;
bool subscribe(strong_actor_ptr who) override;
void unsubscribe(const actor_control_block* who) override;
} // namespace actor intermediary() const noexcept override;
CAF_TEST_FIXTURE_SCOPE(group_tests, fixture) void stop() override;
CAF_TEST(class_based_joined_at_spawn) { protected:
auto grp = system.groups().get_local("test"); mutable std::mutex mtx_;
// initialize all testee actors, spawning them in the group actor intermediary_;
std::array<actor, 10> xs; bool stopped_ = false;
for (auto& x : xs) subscriber_set subscribers_;
x = system.spawn_in_group(grp, testee_impl); };
// get a function view for all testees
std::array<function_view<testee_if>, 10> fs; using instances_map = std::unordered_map<std::string, intrusive_ptr<impl>>;
std::transform(xs.begin(), xs.end(), fs.begin(), [](const actor& x) {
return make_function_view(actor_cast<testee_if>(x)); explicit local_group_module(actor_system& sys);
});
// make sure all actors start at 0 ~local_group_module() override;
for (auto& f : fs)
CAF_CHECK_EQUAL(f(get_atom_v), 0); expected<group> get(const std::string& group_name) override;
// send a put to all actors via the group and make sure they change state
self->send(grp, put_atom_v, 42); void stop() override;
for (auto& f : fs)
CAF_CHECK_EQUAL(f(get_atom_v), 42); private:
// shutdown all actors std::mutex mtx_;
for (auto& x : xs) bool stopped_ = false;
self->send_exit(x, exit_reason::user_shutdown); instances_map instances_;
} };
CAF_TEST_FIXTURE_SCOPE_END()
} // namespace caf::detail
...@@ -381,8 +381,9 @@ using weak_actor_ptr = weak_intrusive_ptr<actor_control_block>; ...@@ -381,8 +381,9 @@ using weak_actor_ptr = weak_intrusive_ptr<actor_control_block>;
// -- intrusive pointer aliases ------------------------------------------------ // -- intrusive pointer aliases ------------------------------------------------
using strong_actor_ptr = intrusive_ptr<actor_control_block>; using group_module_ptr = intrusive_ptr<group_module>;
using stream_manager_ptr = intrusive_ptr<stream_manager>; using stream_manager_ptr = intrusive_ptr<stream_manager>;
using strong_actor_ptr = intrusive_ptr<actor_control_block>;
// -- unique pointer aliases --------------------------------------------------- // -- unique pointer aliases ---------------------------------------------------
......
...@@ -42,7 +42,8 @@ struct invalid_group_t { ...@@ -42,7 +42,8 @@ struct invalid_group_t {
constexpr invalid_group_t invalid_group = invalid_group_t{}; constexpr invalid_group_t invalid_group = invalid_group_t{};
class CAF_CORE_EXPORT group : detail::comparable<group>, class CAF_CORE_EXPORT group : detail::comparable<group>,
detail::comparable<group, invalid_group_t> { detail::comparable<group, invalid_group_t>,
detail::comparable<group, std::nullptr_t> {
public: public:
using signatures = none_t; using signatures = none_t;
...@@ -60,7 +61,7 @@ public: ...@@ -60,7 +61,7 @@ public:
group& operator=(const invalid_group_t&); group& operator=(const invalid_group_t&);
group(abstract_group*); explicit group(abstract_group*);
group(intrusive_ptr<abstract_group> gptr); group(intrusive_ptr<abstract_group> gptr);
...@@ -76,7 +77,11 @@ public: ...@@ -76,7 +77,11 @@ public:
intptr_t compare(const group& other) const noexcept; intptr_t compare(const group& other) const noexcept;
intptr_t compare(const invalid_group_t&) const noexcept { intptr_t compare(invalid_group_t) const noexcept {
return ptr_ ? 1 : 0;
}
intptr_t compare(std::nullptr_t) const noexcept {
return ptr_ ? 1 : 0; return ptr_ ? 1 : 0;
} }
...@@ -116,22 +121,20 @@ public: ...@@ -116,22 +121,20 @@ public:
template <class Inspector> template <class Inspector>
friend bool inspect(Inspector& f, group& x) { friend bool inspect(Inspector& f, group& x) {
std::string module_name; node_id origin;
std::string group_name; std::string mod;
actor dispatcher; std::string id;
if constexpr (!Inspector::is_loading) { if constexpr (!Inspector::is_loading) {
if (x) { if (x) {
module_name = x.get()->module().name(); origin = x.get()->origin();
group_name = x.get()->identifier(); mod = x.get()->module().name();
dispatcher = x.get()->dispatcher(); id = x.get()->identifier();
} }
} }
auto load_cb = [&] { auto load_cb = [&] {
if constexpr (detail::has_context<Inspector>::value) { if constexpr (detail::has_context<Inspector>::value) {
auto ctx = f.context(); if (auto ctx = f.context()) {
if (ctx != nullptr) { if (auto grp = load_impl(ctx->system(), origin, mod, id)) {
if (auto grp = ctx->system().groups().get(module_name, group_name,
dispatcher)) {
x = std::move(*grp); x = std::move(*grp);
return true; return true;
} else { } else {
...@@ -144,15 +147,19 @@ public: ...@@ -144,15 +147,19 @@ public:
return false; return false;
}; };
return f.object(x) return f.object(x)
.on_load(load_cb) // .on_load(load_cb) //
.fields(f.field("module_name", module_name), .fields(f.field("origin", origin), //
f.field("group_name", group_name), f.field("module", mod), //
f.field("dispatcher", dispatcher)); f.field("identifier", id));
} }
/// @endcond /// @endcond
private: private:
static expected<group> load_impl(actor_system& sys, const node_id& origin,
const std::string& mod,
const std::string& id);
abstract_group* release() noexcept { abstract_group* release() noexcept {
return ptr_.release(); return ptr_.release();
} }
......
...@@ -18,18 +18,15 @@ ...@@ -18,18 +18,15 @@
#pragma once #pragma once
#include <map> #include <atomic>
#include <mutex> #include <functional>
#include <thread>
#include <unordered_map> #include <unordered_map>
#include "caf/abstract_group.hpp" #include "caf/abstract_group.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/shared_spinlock.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/group_module.hpp" #include "caf/group_module.hpp"
#include "caf/optional.hpp"
namespace caf { namespace caf {
...@@ -38,11 +35,17 @@ public: ...@@ -38,11 +35,17 @@ public:
// -- friends ---------------------------------------------------------------- // -- friends ----------------------------------------------------------------
friend class actor_system; friend class actor_system;
friend class group;
friend class io::middleman;
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using modules_map using modules_map = std::unordered_map<std::string, group_module_ptr>;
= std::unordered_map<std::string, std::unique_ptr<group_module>>;
using get_remote_fun
= std::function<expected<group>(const node_id& origin,
const std::string& module_name,
const std::string& group_identifier)>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
...@@ -57,34 +60,31 @@ public: ...@@ -57,34 +60,31 @@ public:
/// Get a handle to the group associated with given URI scheme. /// Get a handle to the group associated with given URI scheme.
/// @threadsafe /// @threadsafe
/// @experimental /// @experimental
expected<group> get(std::string group_uri) const; expected<group> get(std::string group_uri);
/// Get a handle to the group associated with /// Get a handle to the group associated with
/// `identifier` from the module `mod_name`. /// `identifier` from the module `mod_name`.
/// @threadsafe /// @threadsafe
expected<group> get(const std::string& module_name, expected<group> get(const std::string& module_name,
const std::string& group_identifier) const; const std::string& group_identifier);
/// Get a pointer to the group associated with /// Get a pointer to the group associated with
/// `identifier` from the module `local`. /// `identifier` from the module `local`.
/// @threadsafe /// @threadsafe
group get_local(const std::string& group_identifier) const; group get_local(const std::string& group_identifier);
/// Returns an anonymous group. /// Returns an anonymous group.
/// Each calls to this member function returns a new instance /// Each calls to this member function returns a new instance
/// of an anonymous group. Anonymous groups can be used whenever /// of an anonymous group. Anonymous groups can be used whenever
/// a set of actors wants to communicate using an exclusive channel. /// a set of actors wants to communicate using an exclusive channel.
group anonymous() const; group anonymous();
/// Returns the module named `name` if it exists, otherwise `none`. /// Returns the module named `name` if it exists, otherwise `none`.
optional<group_module&> get_module(const std::string& x) const; group_module_ptr get_module(const std::string& x) const;
/// @private
expected<group> get(const std::string& module_name,
const std::string& group_identifier,
const caf::actor& dispatcher) const;
private: private:
get_remote_fun get_remote;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit group_manager(actor_system& sys); explicit group_manager(actor_system& sys);
...@@ -100,7 +100,8 @@ private: ...@@ -100,7 +100,8 @@ private:
// -- data members ----------------------------------------------------------- // -- data members -----------------------------------------------------------
modules_map mmap_; modules_map mmap_;
actor_system& system_; actor_system* system_;
std::atomic<size_t> ad_hoc_id_;
}; };
} // namespace caf } // namespace caf
...@@ -31,7 +31,7 @@ ...@@ -31,7 +31,7 @@
namespace caf { namespace caf {
/// Interface for user-defined multicast implementations. /// Interface for user-defined multicast implementations.
class CAF_CORE_EXPORT group_module { class CAF_CORE_EXPORT group_module : public ref_counted {
public: public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
...@@ -41,7 +41,7 @@ public: ...@@ -41,7 +41,7 @@ public:
group_module& operator=(const group_module&) = delete; group_module& operator=(const group_module&) = delete;
virtual ~group_module(); ~group_module() override;
// -- pure virtual member functions ------------------------------------------ // -- pure virtual member functions ------------------------------------------
...@@ -52,26 +52,23 @@ public: ...@@ -52,26 +52,23 @@ public:
/// @threadsafe /// @threadsafe
virtual expected<group> get(const std::string& group_name) = 0; virtual expected<group> get(const std::string& group_name) = 0;
/// Returns a pointer to the group associated with the name `group_name`.
/// @threadsafe
virtual expected<group>
get(const std::string& group_name, const caf::actor& dispatcher) = 0;
// -- observers -------------------------------------------------------------- // -- observers --------------------------------------------------------------
/// Returns the hosting actor system. /// Returns the hosting actor system.
actor_system& system() const { actor_system& system() const noexcept {
return system_; return *system_;
} }
/// Returns the name of this module implementation. /// Returns the name of this module implementation.
const std::string& name() const { const std::string& name() const noexcept {
return name_; return name_;
} }
private: private:
actor_system& system_; actor_system* system_;
std::string name_; std::string name_;
}; };
using group_module_ptr = intrusive_ptr<group_module>;
} // namespace caf } // namespace caf
...@@ -114,6 +114,14 @@ public: ...@@ -114,6 +114,14 @@ public:
return !data_; return !data_;
} }
/// Checks whether this messages contains the types `Ts...` with values
/// `values...`. Users may pass `std::ignore` as a wildcard for individual
/// elements. Elements are compared using `operator==`.
template <class... Ts>
bool matches(const Ts&... values) const {
return matches_impl(std::index_sequence_for<Ts...>{}, values...);
}
// -- serialization ---------------------------------------------------------- // -- serialization ----------------------------------------------------------
bool save(serializer& sink) const; bool save(serializer& sink) const;
...@@ -174,6 +182,19 @@ public: ...@@ -174,6 +182,19 @@ public:
} }
private: private:
template <size_t Pos, class T>
bool matches_at(const T& value) const {
if constexpr (std::is_same<T, decltype(std::ignore)>::value)
return true;
else
return match_element<T>(Pos) && get_as<T>(Pos) == value;
}
template <size_t... Is, class... Ts>
bool matches_impl(std::index_sequence<Is...>, const Ts&... values) const {
return (matches_at<Is>(values) && ...);
}
data_ptr data_; data_ptr data_;
}; };
......
...@@ -150,10 +150,15 @@ public: ...@@ -150,10 +150,15 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Queries whether this node is not default-constructed. /// Queries whether this node is not default-constructed.
explicit operator bool() const { explicit operator bool() const noexcept {
return static_cast<bool>(data_); return static_cast<bool>(data_);
} }
/// Queries whether this node is default-constructed.
bool operator!() const noexcept {
return !data_;
}
/// Compares this instance to `other`. /// Compares this instance to `other`.
/// @returns -1 if `*this < other`, 0 if `*this == other`, and 1 otherwise. /// @returns -1 if `*this < other`, 0 if `*this == other`, and 1 otherwise.
int compare(const node_id& other) const noexcept; int compare(const node_id& other) const noexcept;
......
...@@ -18,21 +18,21 @@ ...@@ -18,21 +18,21 @@
#include "caf/abstract_group.hpp" #include "caf/abstract_group.hpp"
#include "caf/group.hpp"
#include "caf/message.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/group_module.hpp"
#include "caf/group_manager.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
#include "caf/group.hpp"
#include "caf/group_manager.hpp"
#include "caf/group_module.hpp"
#include "caf/message.hpp"
namespace caf { namespace caf {
abstract_group::abstract_group(group_module& mod, std::string id, node_id nid) abstract_group::abstract_group(group_module_ptr mod, std::string id,
: abstract_channel(abstract_channel::is_abstract_group_flag), node_id nid)
system_(mod.system()), : abstract_channel(abstract_channel::is_abstract_group_flag),
parent_(mod), parent_(std::move(mod)),
identifier_(std::move(id)), origin_(std::move(nid)),
origin_(std::move(nid)) { identifier_(std::move(id)) {
// nop // nop
} }
...@@ -40,4 +40,19 @@ abstract_group::~abstract_group() { ...@@ -40,4 +40,19 @@ abstract_group::~abstract_group() {
// nop // nop
} }
actor_system& abstract_group::system() const noexcept {
return module().system();
}
std::string abstract_group::to_string() const {
auto result = module().name();
result += ':';
result += identifier();
return result;
}
actor abstract_group::intermediary() const noexcept {
return {};
}
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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/detail/local_group_module.hpp"
#include "caf/actor_system.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/make_counted.hpp"
#include "caf/send.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/string_algorithms.hpp"
namespace caf::detail {
// -- local group intermediary -------------------------------------------------
local_group_module::intermediary_actor_state::intermediary_actor_state(
event_based_actor* self, abstract_group_ptr gptr)
: self(self), gptr(std::move(gptr)) {
// nop
}
behavior local_group_module::intermediary_actor_state::make_behavior() {
self->set_down_handler([this](const down_msg& dm) {
if (auto ptr = dm.source.get())
gptr->unsubscribe(ptr);
});
return {
[this](join_atom, const strong_actor_ptr& other) {
CAF_LOG_TRACE(CAF_ARG(other));
if (other) {
gptr->subscribe(other);
self->monitor(other);
}
},
[this](leave_atom, const strong_actor_ptr& other) {
CAF_LOG_TRACE(CAF_ARG(other));
if (other) {
gptr->unsubscribe(other.get());
self->demonitor(other);
}
},
[this](forward_atom, const message& what) {
CAF_LOG_TRACE(CAF_ARG(what));
gptr->enqueue(self->current_sender(), make_message_id(), what,
self->context());
},
};
}
// -- local group impl ---------------------------------------------------------
local_group_module::impl::impl(group_module_ptr mod, std::string id)
: super(mod, std::move(id), mod->system().node()) {
CAF_LOG_DEBUG("created new local group:" << identifier_);
}
local_group_module::impl::~impl() {
CAF_LOG_DEBUG("destroyed local group:" << identifier_);
}
void local_group_module::impl::enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* host) {
std::unique_lock<std::mutex> guard{mtx_};
for (auto subscriber : subscribers_)
subscriber->enqueue(sender, mid, content, host);
}
bool local_group_module::impl::subscribe(strong_actor_ptr who) {
std::unique_lock<std::mutex> guard{mtx_};
return !stopped_ && subscribers_.emplace(who).second;
}
void local_group_module::impl::unsubscribe(const actor_control_block* who) {
std::unique_lock<std::mutex> guard{mtx_};
// Note: can't call erase with `who` directly, because only set::find has an
// overload for any K that is less-comparable to Key.
if (auto i = subscribers_.find(who); i != subscribers_.end())
subscribers_.erase(i);
}
actor local_group_module::impl::intermediary() const noexcept {
std::unique_lock<std::mutex> guard{mtx_};
return intermediary_;
}
void local_group_module::impl::stop() {
CAF_LOG_DEBUG("stop local group:" << identifier_);
auto hdl = actor{};
auto subs = subscriber_set{};
{
using std::swap;
std::unique_lock<std::mutex> guard{mtx_};
if (stopped_)
return;
stopped_ = true;
swap(subs, subscribers_);
swap(hdl, intermediary_);
}
anon_send_exit(hdl, exit_reason::user_shutdown);
}
// -- local group module -------------------------------------------------------
local_group_module::local_group_module(actor_system& sys)
: super(sys, "local") {
// nop
}
local_group_module::~local_group_module() {
stop();
}
expected<group> local_group_module::get(const std::string& group_name) {
std::unique_lock<std::mutex> guard{mtx_};
if (stopped_) {
return make_error(sec::runtime_error,
"cannot get a group from on a stopped module");
} else if (auto i = instances_.find(group_name); i != instances_.end()) {
return group{i->second.get()};
} else {
auto ptr = make_counted<impl>(this, group_name);
ptr->intermediary_ = system().spawn<intermediary_actor, hidden>(ptr);
instances_.emplace(group_name, ptr);
return group{ptr.release()};
}
}
void local_group_module::stop() {
instances_map tmp;
{
using std::swap;
std::unique_lock<std::mutex> guard{mtx_};
if (stopped_)
return;
swap(instances_, tmp);
stopped_ = true;
}
for (auto& kvp : tmp)
kvp.second->stop();
}
} // namespace caf::detail
...@@ -58,13 +58,23 @@ intptr_t group::compare(const group& other) const noexcept { ...@@ -58,13 +58,23 @@ intptr_t group::compare(const group& other) const noexcept {
return compare(ptr_.get(), other.ptr_.get()); return compare(ptr_.get(), other.ptr_.get());
} }
expected<group> group::load_impl(actor_system& sys, const node_id& origin,
const std::string& mod,
const std::string& id) {
if (!origin || origin == sys.node())
return sys.groups().get(mod, id);
else if (auto& get_remote = sys.groups().get_remote)
return get_remote(origin, mod, id);
else
return make_error(sec::feature_disabled,
"cannot access remote group: middleman not loaded");
}
std::string to_string(const group& x) { std::string to_string(const group& x) {
if (x == invalid_group) if (x == invalid_group)
return "<invalid-group>"; return "<invalid-group>";
std::string result = x.get()->module().name(); else
result += ":"; return x.get()->to_string();
result += x.get()->identifier();
return result;
} }
} // namespace caf } // namespace caf
This diff is collapsed.
...@@ -21,8 +21,7 @@ ...@@ -21,8 +21,7 @@
namespace caf { namespace caf {
group_module::group_module(actor_system& sys, std::string mname) group_module::group_module(actor_system& sys, std::string mname)
: system_(sys), : system_(&sys), name_(std::move(mname)) {
name_(std::move(mname)) {
// nop // nop
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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. *
******************************************************************************/
#define CAF_SUITE detail.local_group_module
#include "caf/all.hpp"
#include "core-test.hpp"
#include <algorithm>
#include <array>
#include <chrono>
#include "caf/detail/local_group_module.hpp"
using namespace caf;
namespace {
struct testee_state {
int x = 0;
static inline const char* name = "testee";
};
behavior testee_impl(stateful_actor<testee_state>* self) {
return {
[=](put_atom, int x) { self->state.x = x; },
[=](get_atom) { return self->state.x; },
};
}
struct fixture : test_coordinator_fixture<> {
fixture() {
auto ptr = sys.groups().get_module("local");
uut.reset(dynamic_cast<detail::local_group_module*>(ptr.get()));
}
~fixture() {
// Groups keep their subscribers alive (on purpose). Since we don't want to
// manually kill all our testee actors, we simply force the group module to
// stop here.
uut->stop();
}
intrusive_ptr<detail::local_group_module> uut;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(group_tests, fixture)
CAF_TEST(local groups are singletons) {
auto ptr1 = unbox(uut->get("test"));
auto ptr2 = unbox(uut->get("test"));
CAF_CHECK_EQUAL(ptr1.get(), ptr2.get());
auto ptr3 = sys.groups().get_local("test");
CAF_CHECK_EQUAL(ptr1.get(), ptr3.get());
}
CAF_TEST(local groups forward messages to all subscribers) {
CAF_MESSAGE("Given two subscribers to the group 'test'.");
auto grp = unbox(uut->get("test"));
auto t1 = sys.spawn_in_group(grp, testee_impl);
auto t2 = sys.spawn_in_group(grp, testee_impl);
{ // Subtest.
CAF_MESSAGE("When an actors sends to the group.");
self->send(grp, put_atom_v, 42);
CAF_MESSAGE("Then both subscribers receive the message.");
expect((put_atom, int), from(self).to(t1).with(_, 42));
expect((put_atom, int), from(self).to(t2).with(_, 42));
}
{ // Subtest.
CAF_MESSAGE("When an actors leaves the group.");
CAF_MESSAGE("And an actors sends to the group.");
grp->unsubscribe(actor_cast<actor_control_block*>(t1));
self->send(grp, put_atom_v, 23);
CAF_MESSAGE("Then only one remaining actor receives the message.");
disallow((put_atom, int), from(self).to(t1).with(_, 23));
expect((put_atom, int), from(self).to(t2).with(_, 23));
}
}
CAF_TEST(local group intermediaries manage groups) {
CAF_MESSAGE("Given two subscribers to the group 'test'.");
auto grp = unbox(uut->get("test"));
auto intermediary = grp.get()->intermediary();
auto t1 = sys.spawn_in_group(grp, testee_impl);
auto t2 = sys.spawn_in_group(grp, testee_impl);
{ // Subtest.
CAF_MESSAGE("When an actors sends to the group's intermediary.");
inject((forward_atom, message),
from(self)
.to(intermediary)
.with(forward_atom_v, make_message(put_atom_v, 42)));
CAF_MESSAGE("Then both subscribers receive the message.");
expect((put_atom, int), from(self).to(t1).with(_, 42));
expect((put_atom, int), from(self).to(t2).with(_, 42));
}
auto t3 = sys.spawn(testee_impl);
{ // Subtest.
CAF_MESSAGE("When an actor sends 'join' to the group's intermediary.");
CAF_MESSAGE("And an actors sends to the group's intermediary.");
inject((join_atom, strong_actor_ptr),
from(self)
.to(intermediary)
.with(join_atom_v, actor_cast<strong_actor_ptr>(t3)));
self->send(grp, put_atom_v, 23);
CAF_MESSAGE("Then all three subscribers receive the message.");
expect((put_atom, int), from(self).to(t1).with(_, 23));
expect((put_atom, int), from(self).to(t2).with(_, 23));
expect((put_atom, int), from(self).to(t3).with(_, 23));
}
{ // Subtest.
CAF_MESSAGE("When an actor sends 'leave' to the group's intermediary.");
CAF_MESSAGE("And an actors sends to the group's intermediary.");
inject((leave_atom, strong_actor_ptr),
from(self)
.to(intermediary)
.with(leave_atom_v, actor_cast<strong_actor_ptr>(t3)));
self->send(grp, put_atom_v, 37337);
CAF_MESSAGE("Then only the two remaining subscribers receive the message.");
self->send(grp, put_atom_v, 37337);
expect((put_atom, int), from(self).to(t1).with(_, 37337));
expect((put_atom, int), from(self).to(t2).with(_, 37337));
disallow((put_atom, int), from(self).to(t3).with(_, 37337));
}
}
CAF_TEST_FIXTURE_SCOPE_END()
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