Unverified Commit 72e62b1a authored by Noir's avatar Noir Committed by GitHub

Merge pull request #1159

Fix communiation with remote groups
parents cc68df1c 0af8c8cc
...@@ -39,6 +39,7 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -39,6 +39,7 @@ is based on [Keep a Changelog](https://keepachangelog.com).
failed due to undefined references to `__atomic_fetch` symbols. Adding a CMake failed due to undefined references to `__atomic_fetch` symbols. Adding a CMake
dependency for `caf_core` to libatomic gets executables to compile and link as dependency for `caf_core` to libatomic gets executables to compile and link as
expected (#1153). expected (#1153).
- Fixed a regression for remote groups introduced in 0.18.0-rc.1 (#1157).
## [0.18.0-rc.1] - 2020-09-09 ## [0.18.0-rc.1] - 2020-09-09
......
...@@ -45,10 +45,11 @@ behavior client(event_based_actor* self, const std::string& name) { ...@@ -45,10 +45,11 @@ behavior client(event_based_actor* self, const std::string& name) {
} }
}, },
[=](join_atom, const group& what) { [=](join_atom, const group& what) {
for (const auto& g : self->joined_groups()) { auto groups = self->joined_groups();
std::cout << "*** leave " << to_string(g) << std::endl; for (auto&& grp : groups) {
self->send(g, name + " has left the chatroom"); std::cout << "*** leave " << to_string(grp) << std::endl;
self->leave(g); self->send(grp, name + " has left the chatroom");
self->leave(grp);
} }
std::cout << "*** join " << to_string(what) << std::endl; std::cout << "*** join " << to_string(what) << std::endl;
self->join(what); self->join(what);
...@@ -62,27 +63,31 @@ behavior client(event_based_actor* self, const std::string& name) { ...@@ -62,27 +63,31 @@ behavior client(event_based_actor* self, const std::string& name) {
[=](const group_down_msg& g) { [=](const group_down_msg& g) {
std::cout << "*** chatroom offline: " << to_string(g.source) << std::endl; std::cout << "*** chatroom offline: " << to_string(g.source) << std::endl;
}, },
[=](leave_atom) {
auto groups = self->joined_groups();
for (auto&& grp : groups) {
std::cout << "*** leave " << to_string(grp) << std::endl;
self->send(grp, name + " has left the chatroom");
self->leave(grp);
}
},
}; };
} }
class config : public actor_system_config { class config : public actor_system_config {
public: public:
std::string name;
std::vector<std::string> group_uris;
uint16_t port = 0;
bool server_mode = false;
config() { config() {
opt_group{custom_options_, "global"} opt_group{custom_options_, "global"}
.add(name, "name,n", "set name") .add<std::string>("name,n", "set name")
.add(group_uris, "group,g", "join group") .add<std::string>("group,g", "join group")
.add(server_mode, "server,s", "run in server mode") .add<bool>("server,s", "run in server mode")
.add(port, "port,p", "set port (ignored in client mode)"); .add<uint16_t>("port,p", "set port (ignored in client mode)");
} }
}; };
void run_server(actor_system& system, const config& cfg) { void run_server(actor_system& sys) {
auto res = system.middleman().publish_local_groups(cfg.port); auto port = get_or(sys.config(), "port", uint16_t{0});
auto res = sys.middleman().publish_local_groups(port);
if (!res) { if (!res) {
std::cerr << "*** publishing local groups failed: " std::cerr << "*** publishing local groups failed: "
<< to_string(res.error()) << std::endl; << to_string(res.error()) << std::endl;
...@@ -95,8 +100,10 @@ void run_server(actor_system& system, const config& cfg) { ...@@ -95,8 +100,10 @@ void run_server(actor_system& system, const config& cfg) {
std::cout << "... cya" << std::endl; std::cout << "... cya" << std::endl;
} }
void run_client(actor_system& system, const config& cfg) { void run_client(actor_system& sys) {
auto name = cfg.name; std::string name;
if (auto config_name = get_if<std::string>(&sys.config(), "name"))
name = *config_name;
while (name.empty()) { while (name.empty()) {
std::cout << "please enter your name: " << std::flush; std::cout << "please enter your name: " << std::flush;
if (!std::getline(std::cin, name)) { if (!std::getline(std::cin, name)) {
...@@ -104,61 +111,50 @@ void run_client(actor_system& system, const config& cfg) { ...@@ -104,61 +111,50 @@ 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 = sys.spawn(client, name);
auto client_actor = system.spawn(client, name); if (auto locator = get_if<std::string>(&sys.config(), "group")) {
for (auto& uri : cfg.group_uris) { if (auto grp = sys.groups().get(*locator)) {
auto tmp = system.groups().get(uri); anon_send(client_actor, join_atom_v, std::move(*grp));
if (tmp) } else {
anon_send(client_actor, join_atom_v, std::move(*tmp)); std::cerr << R"(*** failed to parse ")" << *locator
else << R"(" as group locator: )" << to_string(grp.error())
std::cerr << R"(*** failed to parse ")" << uri << R"(" as group URI: )" << std::endl;
<< to_string(tmp.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(" "));
message_handler f{ if (words.size() == 3 && words[0] == "/join") {
[&](const std::string& cmd, const std::string& mod, if (auto grp = sys.groups().get(words[1], words[2]))
const std::string& id) {
if (cmd == "/join") {
auto grp = system.groups().get(mod, id);
if (grp)
anon_send(client_actor, join_atom_v, *grp); anon_send(client_actor, join_atom_v, *grp);
} else { else
send_input(); std::cerr << "*** failed to join group: " << to_string(grp.error())
} << std::endl;
}, } else if (words.size() == 1 && words[0] == "/quit") {
[&](const std::string& cmd) {
if (cmd == "/quit") {
std::cin.setstate(std::ios_base::eofbit); std::cin.setstate(std::ios_base::eofbit);
} else if (cmd[0] == '/') { } else {
std::cout << "*** available commands:\n" std::cout << "*** available commands:\n"
" /join <module> <group> join a new chat channel\n" " /join <module> <group> join a new chat channel\n"
" /quit quit the program\n" " /quit quit the program\n"
" /help print this text\n"; " /help print this text\n";
}
} else { } else {
send_input(); anon_send(client_actor, broadcast_atom_v, i->str);
} }
},
};
auto msg = message_builder(words.begin(), words.end()).move_to_message();
auto res = f(msg);
if (!res)
send_input();
} }
// force actor to quit anon_send(client_actor, leave_atom_v);
anon_send_exit(client_actor, exit_reason::user_shutdown); anon_send_exit(client_actor, exit_reason::user_shutdown);
} }
void caf_main(actor_system& system, const config& cfg) { void caf_main(actor_system& sys, const config& cfg) {
auto f = cfg.server_mode ? run_server : run_client; auto f = get_or(cfg, "server", false) ? run_server : run_client;
f(system, cfg); f(sys);
} }
CAF_MAIN(id_block::group_chat, io::middleman) CAF_MAIN(id_block::group_chat, io::middleman)
...@@ -86,7 +86,9 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS} ...@@ -86,7 +86,9 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS}
src/detail/get_process_id.cpp src/detail/get_process_id.cpp
src/detail/get_root_uuid.cpp src/detail/get_root_uuid.cpp
src/detail/glob_match.cpp src/detail/glob_match.cpp
src/detail/group_tunnel.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
...@@ -265,8 +267,10 @@ caf_add_test_suites(caf-core-test ...@@ -265,8 +267,10 @@ caf_add_test_suites(caf-core-test
detail.bounds_checker detail.bounds_checker
detail.config_consumer detail.config_consumer
detail.encode_base64 detail.encode_base64
detail.group_tunnel
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
...@@ -309,7 +313,6 @@ caf_add_test_suites(caf-core-test ...@@ -309,7 +313,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 stringify() 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,66 @@ ...@@ -16,71 +16,66 @@
* 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 "caf/abstract_group.hpp"
#include "caf/actor.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/local_group_module.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/fwd.hpp"
#include "caf/stateful_actor.hpp"
#include "core-test.hpp" #include <tuple>
#include <array> namespace caf::detail {
#include <chrono>
#include <algorithm>
using namespace caf; /// Represents a group that runs on a remote CAF node.
class CAF_CORE_EXPORT group_tunnel : public local_group_module::impl {
public:
using super = local_group_module::impl;
namespace { using cached_message = std::tuple<strong_actor_ptr, message_id, message>;
using testee_if using cached_message_list = std::vector<cached_message>;
= typed_actor<replies_to<get_atom>::with<int>, reacts_to<put_atom, int>>;
struct testee_state { // Creates a connected tunnel.
int x = 0; group_tunnel(group_module_ptr mod, std::string id,
}; actor upstream_intermediary);
// Creates an unconnected tunnel that caches incoming messages until it
// becomes connected to the upstream intermediary.
group_tunnel(group_module_ptr mod, std::string id, node_id origin);
~group_tunnel() override;
bool subscribe(strong_actor_ptr who) override;
void unsubscribe(const actor_control_block* who) override;
// Locally enqueued message, forwarded via worker_.
void enqueue(strong_actor_ptr sender, message_id mid, message content,
execution_unit* host) override;
void stop() override;
std::string stringify() const override;
// Messages received from the upstream group, forwarded to local subscribers.
void upstream_enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* host);
bool connect(actor upstream_intermediary);
bool connected() const noexcept;
actor worker() const noexcept;
behavior testee_impl(stateful_actor<testee_state>* self) { private:
auto subscriptions = self->joined_groups(); actor worker_;
return { cached_message_list cached_messages_;
[=](put_atom, int x) {
self->state.x = x;
},
[=](get_atom) {
return self->state.x;
}
};
}
struct fixture {
actor_system_config config;
actor_system system{config};
scoped_actor self{system};
}; };
} // namespace using group_tunnel_ptr = intrusive_ptr<group_tunnel>;
CAF_TEST_FIXTURE_SCOPE(group_tests, fixture)
CAF_TEST(class_based_joined_at_spawn) {
auto grp = system.groups().get_local("test");
// initialize all testee actors, spawning them in the group
std::array<actor, 10> xs;
for (auto& x : xs)
x = system.spawn_in_group(grp, testee_impl);
// get a function view for all testees
std::array<function_view<testee_if>, 10> fs;
std::transform(xs.begin(), xs.end(), fs.begin(), [](const actor& x) {
return make_function_view(actor_cast<testee_if>(x));
});
// make sure all actors start at 0
for (auto& f : fs)
CAF_CHECK_EQUAL(f(get_atom_v), 0);
// send a put to all actors via the group and make sure they change state
self->send(grp, put_atom_v, 42);
for (auto& f : fs)
CAF_CHECK_EQUAL(f(get_atom_v), 42);
// shutdown all actors
for (auto& x : xs)
self->send_exit(x, exit_reason::user_shutdown);
}
CAF_TEST_FIXTURE_SCOPE_END()
} // namespace caf::detail
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#pragma once
#include <functional>
#include <mutex>
#include <set>
#include <unordered_map>
#include "caf/abstract_group.hpp"
#include "caf/group_module.hpp"
namespace caf::detail {
/// 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 CAF_CORE_EXPORT local_group_module : public group_module {
public:
using super = group_module;
struct intermediary_actor_state {
static inline const char* name = "caf.detail.group-intermediary";
intermediary_actor_state(event_based_actor* self, abstract_group_ptr gptr);
behavior make_behavior();
event_based_actor* self;
abstract_group_ptr gptr;
};
/// A group intermediary enables remote actors to join and leave groups on
/// this endpoint as well as sending message to it.
using intermediary_actor = stateful_actor<intermediary_actor_state>;
/// Implementation of the group interface for instances of this module.
class CAF_CORE_EXPORT 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, node_id origin);
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;
actor intermediary() const noexcept override;
void stop() override;
protected:
template <class F>
auto critical_section(F&& fun) const {
std::unique_lock<std::mutex> guard{mtx_};
return fun();
}
/// @pre `mtx_` is locked
std::pair<bool, size_t> subscribe_impl(strong_actor_ptr who);
/// @pre `mtx_` is locked
std::pair<bool, size_t> unsubscribe_impl(const actor_control_block* who);
mutable std::mutex mtx_;
actor intermediary_;
bool stopped_ = false;
subscriber_set subscribers_;
};
using instances_map = std::unordered_map<std::string, intrusive_ptr<impl>>;
explicit local_group_module(actor_system& sys);
~local_group_module() override;
expected<group> get(const std::string& group_name) override;
void stop() override;
private:
std::mutex mtx_;
bool stopped_ = false;
instances_map instances_;
};
} // 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;
...@@ -52,17 +53,15 @@ public: ...@@ -52,17 +53,15 @@ public:
group(const group&) = default; group(const group&) = default;
group(const invalid_group_t&); group(invalid_group_t);
explicit group(intrusive_ptr<abstract_group> gptr);
group& operator=(group&&) = default; group& operator=(group&&) = default;
group& operator=(const group&) = default; group& operator=(const group&) = default;
group& operator=(const invalid_group_t&); group& operator=(invalid_group_t);
group(abstract_group*);
group(intrusive_ptr<abstract_group> gptr);
explicit operator bool() const noexcept { explicit operator bool() const noexcept {
return static_cast<bool>(ptr_); return static_cast<bool>(ptr_);
...@@ -76,7 +75,11 @@ public: ...@@ -76,7 +75,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 +119,20 @@ public: ...@@ -116,22 +119,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 {
...@@ -145,14 +146,18 @@ public: ...@@ -145,14 +146,18 @@ public:
}; };
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
...@@ -335,6 +335,9 @@ public: ...@@ -335,6 +335,9 @@ public:
/// Removes a monitor from `whom`. /// Removes a monitor from `whom`.
void demonitor(const actor_addr& whom); void demonitor(const actor_addr& whom);
/// Removes a monitor from `whom`.
void demonitor(const strong_actor_ptr& whom);
/// Removes a monitor from `node`. /// Removes a monitor from `node`.
void demonitor(const node_id& node); void demonitor(const node_id& node);
......
...@@ -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;
......
...@@ -309,7 +309,7 @@ public: ...@@ -309,7 +309,7 @@ public:
/// Sets a custom handler for unexpected messages. /// Sets a custom handler for unexpected messages.
template <class F> template <class F>
typename std::enable_if<std::is_convertible< typename std::enable_if<std::is_convertible<
F, std::function<result<message>(message&)>>::value>::type F, std::function<skippable_result(message&)>>::value>::type
set_default_handler(F fun) { set_default_handler(F fun) {
default_handler_ = [=](scheduled_actor*, message& xs) { return fun(xs); }; default_handler_ = [=](scheduled_actor*, message& xs) { return fun(xs); };
} }
......
...@@ -349,6 +349,7 @@ CAF_BEGIN_TYPE_ID_BLOCK(core_module, 0) ...@@ -349,6 +349,7 @@ CAF_BEGIN_TYPE_ID_BLOCK(core_module, 0)
CAF_ADD_ATOM(core_module, caf, flush_atom) CAF_ADD_ATOM(core_module, caf, flush_atom)
CAF_ADD_ATOM(core_module, caf, forward_atom) CAF_ADD_ATOM(core_module, caf, forward_atom)
CAF_ADD_ATOM(core_module, caf, get_atom) CAF_ADD_ATOM(core_module, caf, get_atom)
CAF_ADD_ATOM(core_module, caf, group_atom)
CAF_ADD_ATOM(core_module, caf, idle_atom) CAF_ADD_ATOM(core_module, caf, idle_atom)
CAF_ADD_ATOM(core_module, caf, join_atom) CAF_ADD_ATOM(core_module, caf, join_atom)
CAF_ADD_ATOM(core_module, caf, leave_atom) CAF_ADD_ATOM(core_module, caf, leave_atom)
......
...@@ -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,
node_id nid)
: abstract_channel(abstract_channel::is_abstract_group_flag), : abstract_channel(abstract_channel::is_abstract_group_flag),
system_(mod.system()), parent_(std::move(mod)),
parent_(mod), origin_(std::move(nid)),
identifier_(std::move(id)), identifier_(std::move(id)) {
origin_(std::move(nid)) {
// 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::stringify() const {
auto result = module().name();
result += ':';
result += identifier();
return result;
}
actor abstract_group::intermediary() const noexcept {
return {};
}
} // namespace caf } // namespace caf
...@@ -135,6 +135,10 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) { ...@@ -135,6 +135,10 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) {
[=](registry_lookup_atom, const std::string& name) { [=](registry_lookup_atom, const std::string& name) {
return self->home_system().registry().get(name); return self->home_system().registry().get(name);
}, },
// get the intermediary of a local group
[=](get_atom, group_atom, const std::string& id) {
return self->home_system().groups().get_local(id).get()->intermediary();
},
}; };
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/group_tunnel.hpp"
namespace caf::detail {
namespace {
struct group_worker_actor_state {
static inline const char* name = "caf.detail.group-tunnel";
group_worker_actor_state(event_based_actor* self, group_tunnel_ptr gptr,
actor intermediary)
: self(self), gptr(std::move(gptr)), intermediary(std::move(intermediary)) {
// nop
}
behavior make_behavior() {
self->set_down_handler([this](const down_msg& dm) {
if (dm.source == intermediary) {
gptr->stop();
}
});
self->set_default_handler([this](message& msg) -> skippable_result {
gptr->upstream_enqueue(std::move(self->current_sender()),
self->take_current_message_id(), std::move(msg),
self->context());
return message{};
});
self->monitor(intermediary);
return {
[this](sys_atom, join_atom) {
self->send(intermediary, join_atom_v, self->ctrl());
},
[this](sys_atom, leave_atom) {
self->send(intermediary, leave_atom_v, self->ctrl());
},
[this](sys_atom, forward_atom, message& msg) {
self->delegate(intermediary, forward_atom_v, std::move(msg));
},
};
}
event_based_actor* self;
group_tunnel_ptr gptr;
actor intermediary;
};
// A group tunnel enables remote actors to join and leave groups on this
// endpoint as well as sending message to it.
using group_worker_actor = stateful_actor<group_worker_actor_state>;
} // namespace
group_tunnel::group_tunnel(group_module_ptr mod, std::string id,
actor upstream_intermediary)
: super(std::move(mod), std::move(id), upstream_intermediary.node()) {
intermediary_ = std::move(upstream_intermediary);
worker_ = system().spawn<group_worker_actor, hidden>(this, intermediary_);
}
group_tunnel::group_tunnel(group_module_ptr mod, std::string id, node_id nid)
: super(std::move(mod), std::move(id), std::move(nid)) {
// nop
}
group_tunnel::~group_tunnel() {
// nop
}
bool group_tunnel::subscribe(strong_actor_ptr who) {
return critical_section([this, &who] {
auto [added, new_size] = subscribe_impl(std::move(who));
if (added && new_size == 1)
anon_send(worker_, sys_atom_v, join_atom_v);
return added;
});
}
void group_tunnel::unsubscribe(const actor_control_block* who) {
return critical_section([this, who] {
auto [removed, new_size] = unsubscribe_impl(who);
if (removed && new_size == 0)
anon_send(worker_, sys_atom_v, leave_atom_v);
});
}
void group_tunnel::enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(content));
std::unique_lock<std::mutex> guard{mtx_};
if (worker_ != nullptr) {
auto wrapped = make_message(sys_atom_v, forward_atom_v, std::move(content));
worker_->enqueue(std::move(sender), mid, std::move(wrapped), host);
} else if (!stopped_) {
auto wrapped = make_message(sys_atom_v, forward_atom_v, std::move(content));
cached_messages_.emplace_back(std::move(sender), mid, std::move(wrapped));
}
}
void group_tunnel::stop() {
CAF_LOG_TRACE("");
CAF_LOG_DEBUG("stop group tunnel:" << CAF_ARG2("module", module().name())
<< CAF_ARG2("identifier", identifier_));
auto hdl = actor{};
auto subs = subscriber_set{};
auto cache = cached_message_list{};
auto stopped = critical_section([this, &hdl, &subs, &cache] {
using std::swap;
if (!stopped_) {
stopped_ = true;
swap(subs, subscribers_);
swap(hdl, worker_);
swap(cache, cached_messages_);
return true;
} else {
return false;
}
});
if (stopped) {
anon_send_exit(hdl, exit_reason::user_shutdown);
if (!subs.empty()) {
auto bye = make_message(group_down_msg{group{this}});
for (auto& sub : subs)
sub->enqueue(nullptr, make_message_id(), bye, nullptr);
}
}
}
std::string group_tunnel::stringify() const {
std::string result;
result = "remote:";
result += identifier();
result += '@';
result += to_string(origin());
return result;
}
void group_tunnel::upstream_enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(content));
super::enqueue(std::move(sender), mid, std::move(content), host);
}
bool group_tunnel::connect(actor upstream_intermediary) {
CAF_LOG_TRACE(CAF_ARG(upstream_intermediary));
return critical_section([this, &upstream_intermediary] {
if (stopped_ || worker_ != nullptr) {
return false;
} else {
intermediary_ = upstream_intermediary;
worker_ = system().spawn<group_worker_actor, hidden>(
this, upstream_intermediary);
if (!subscribers_.empty())
anon_send(worker_, sys_atom_v, join_atom_v);
for (auto& [sender, mid, content] : cached_messages_)
worker_->enqueue(std::move(sender), mid, std::move(content), nullptr);
cached_messages_.clear();
return true;
}
});
}
bool group_tunnel::connected() const noexcept {
return critical_section([this] { return worker_ != nullptr; });
}
actor group_tunnel::worker() const noexcept {
return critical_section([this] { return worker_; });
}
} // namespace caf::detail
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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,
node_id origin)
: super(mod, std::move(id), origin) {
// nop
}
local_group_module::impl::impl(group_module_ptr mod, std::string id)
: impl(mod, std::move(id), mod->system().node()) {
CAF_LOG_DEBUG("created new local group:" << identifier_);
}
local_group_module::impl::~impl() {
// nop
}
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 subscribe_impl(who).first;
}
void local_group_module::impl::unsubscribe(const actor_control_block* who) {
std::unique_lock<std::mutex> guard{mtx_};
unsubscribe_impl(who);
}
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{};
auto stopped = critical_section([this, &hdl, &subs] {
using std::swap;
if (!stopped_) {
stopped_ = true;
swap(subs, subscribers_);
swap(hdl, intermediary_);
return true;
} else {
return false;
}
});
if (stopped)
anon_send_exit(hdl, exit_reason::user_shutdown);
}
std::pair<bool, size_t>
local_group_module::impl::subscribe_impl(strong_actor_ptr who) {
if (!stopped_) {
auto added = subscribers_.emplace(who).second;
return {added, subscribers_.size()};
} else {
return {false, subscribers_.size()};
}
}
std::pair<bool, size_t>
local_group_module::impl::unsubscribe_impl(const actor_control_block* who) {
if (auto i = subscribers_.find(who); i != subscribers_.end()) {
subscribers_.erase(i);
return {true, subscribers_.size()};
} else {
return {false, subscribers_.size()};
}
}
// -- 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};
} 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
...@@ -29,15 +29,7 @@ ...@@ -29,15 +29,7 @@
namespace caf { namespace caf {
group::group(abstract_group* ptr) : ptr_(ptr) { group::group(invalid_group_t) : ptr_(nullptr) {
// nop
}
group::group(abstract_group* ptr, bool add_ref) : ptr_(ptr, add_ref) {
// nop
}
group::group(const invalid_group_t&) : ptr_(nullptr) {
// nop // nop
} }
...@@ -45,7 +37,7 @@ group::group(abstract_group_ptr gptr) : ptr_(std::move(gptr)) { ...@@ -45,7 +37,7 @@ group::group(abstract_group_ptr gptr) : ptr_(std::move(gptr)) {
// nop // nop
} }
group& group::operator=(const invalid_group_t&) { group& group::operator=(invalid_group_t) {
ptr_.reset(); ptr_.reset();
return *this; return *this;
} }
...@@ -58,13 +50,31 @@ intptr_t group::compare(const group& other) const noexcept { ...@@ -58,13 +50,31 @@ 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()) {
if (mod == "remote") {
// Local groups on this node appear as remote groups on other nodes.
// Hence, serializing back and forth results in receiving a "remote"
// representation for a group that actually runs locally.
return sys.groups().get_local(id);
} else {
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)
return x.get()->stringify();
else
return "<invalid-group>"; return "<invalid-group>";
std::string result = x.get()->module().name();
result += ":";
result += x.get()->identifier();
return result;
} }
} // namespace caf } // namespace caf
...@@ -16,17 +16,12 @@ ...@@ -16,17 +16,12 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include <condition_variable> #include "caf/group_manager.hpp"
#include <mutex>
#include <set>
#include <sstream>
#include <stdexcept>
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/deserializer.hpp" #include "caf/deserializer.hpp"
#include "caf/detail/local_group_module.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/group.hpp" #include "caf/group.hpp"
#include "caf/group_manager.hpp"
#include "caf/locks.hpp" #include "caf/locks.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
...@@ -34,369 +29,12 @@ ...@@ -34,369 +29,12 @@
namespace caf { namespace caf {
namespace {
using exclusive_guard = unique_lock<detail::shared_spinlock>;
using shared_guard = shared_lock<detail::shared_spinlock>;
using upgrade_guard = upgrade_lock<detail::shared_spinlock>;
using upgrade_to_unique_guard = upgrade_to_unique_lock<detail::shared_spinlock>;
class local_dispatcher;
class local_group_module;
void await_all_locals_down(actor_system& sys, std::initializer_list<actor> xs) {
CAF_LOG_TRACE("");
scoped_actor self{sys, true};
std::vector<actor> ys;
for (auto& x : xs)
if (x.node() == sys.node()) {
self->send_exit(x, exit_reason::kill);
ys.push_back(x);
}
// Don't block when using the test coordinator.
if (auto policy = get_if<std::string>(&sys.config(), "caf.scheduler.policy");
policy && *policy != "testing")
self->wait_for(ys);
}
class local_group : public abstract_group {
public:
void send_all_subscribers(const strong_actor_ptr& sender, const message& msg,
execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(msg));
shared_guard guard(mtx_);
for (auto& s : subscribers_)
s->enqueue(sender, make_message_id(), msg, host);
}
void enqueue(strong_actor_ptr sender, message_id, message msg,
execution_unit* host) override {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(msg));
send_all_subscribers(sender, msg, host);
dispatcher_->enqueue(sender, make_message_id(), msg, host);
}
std::pair<bool, size_t> add_subscriber(strong_actor_ptr who) {
CAF_LOG_TRACE(CAF_ARG(who));
if (!who)
return {false, subscribers_.size()};
exclusive_guard guard(mtx_);
auto res = subscribers_.emplace(std::move(who)).second;
return {res, subscribers_.size()};
}
std::pair<bool, size_t> erase_subscriber(const actor_control_block* who) {
CAF_LOG_TRACE(""); // serializing who would cause a deadlock
exclusive_guard guard(mtx_);
auto e = subscribers_.end();
auto cmp = [&](const strong_actor_ptr& lhs) { return lhs.get() == who; };
auto i = std::find_if(subscribers_.begin(), e, cmp);
if (i == e)
return {false, subscribers_.size()};
subscribers_.erase(i);
return {true, subscribers_.size()};
}
bool subscribe(strong_actor_ptr who) override {
CAF_LOG_TRACE(CAF_ARG(who));
return add_subscriber(std::move(who)).first;
}
void unsubscribe(const actor_control_block* who) override {
CAF_LOG_TRACE(CAF_ARG(who));
erase_subscriber(who);
}
void stop() override {
CAF_LOG_TRACE("");
await_all_locals_down(system(), {dispatcher_});
}
local_group(local_group_module& mod, std::string id, node_id nid,
optional<actor> lb);
~local_group() override;
protected:
detail::shared_spinlock mtx_;
#if __cplusplus > 201103L
std::set<strong_actor_ptr, std::less<>> subscribers_;
#else
std::set<strong_actor_ptr> subscribers_;
#endif
actor dispatcher_;
};
using local_group_ptr = intrusive_ptr<local_group>;
class local_dispatcher : public event_based_actor {
public:
explicit local_dispatcher(actor_config& cfg, local_group_ptr g)
: event_based_actor(cfg), group_(std::move(g)) {
// nop
}
void on_exit() override {
acquaintances_.clear();
group_.reset();
}
const char* name() const override {
return "local_dispatcher";
}
behavior make_behavior() override {
CAF_LOG_TRACE("");
// instead of dropping "unexpected" messages,
// we simply forward them to our acquaintances
auto fwd = [=](scheduled_actor*, message& msg) -> skippable_result {
send_to_acquaintances(std::move(msg));
return delegated<message>{};
};
set_default_handler(fwd);
set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
auto first = acquaintances_.begin();
auto last = acquaintances_.end();
auto i = std::find_if(first, last,
[&](const actor& a) { return a == dm.source; });
if (i != last)
acquaintances_.erase(i);
});
// return behavior
return {[=](join_atom, const actor& other) {
CAF_LOG_TRACE(CAF_ARG(other));
if (acquaintances_.insert(other).second) {
monitor(other);
}
},
[=](leave_atom, const actor& other) {
CAF_LOG_TRACE(CAF_ARG(other));
acquaintances_.erase(other);
if (acquaintances_.erase(other) > 0)
demonitor(other);
},
[=](forward_atom, const message& what) {
CAF_LOG_TRACE(CAF_ARG(what));
// local forwarding
group_->send_all_subscribers(current_element_->sender, what,
context());
// forward to all acquaintances
send_to_acquaintances(what);
}};
}
private:
void send_to_acquaintances(const message& what) {
// send to all remote subscribers
auto src = current_element_->sender;
CAF_LOG_DEBUG(CAF_ARG(acquaintances_.size())
<< CAF_ARG(src) << CAF_ARG(what));
for (auto& acquaintance : acquaintances_)
acquaintance->enqueue(src, make_message_id(), what, context());
}
local_group_ptr group_;
std::set<actor> acquaintances_;
};
// Send a join message to the original group if a proxy
// has local subscriptions and a "LEAVE" message to the original group
// if there's no subscription left.
class local_group_proxy;
using local_group_proxy_ptr = intrusive_ptr<local_group_proxy>;
class proxy_dispatcher : public event_based_actor {
public:
proxy_dispatcher(actor_config& cfg, local_group_proxy_ptr grp)
: event_based_actor(cfg), group_(std::move(grp)) {
CAF_LOG_TRACE("");
}
behavior make_behavior() override;
void on_exit() override {
group_.reset();
}
private:
local_group_proxy_ptr group_;
};
class local_group_proxy : public local_group {
public:
local_group_proxy(actor_system& sys, actor remote_dispatcher,
local_group_module& mod, std::string id, node_id nid)
: local_group(mod, std::move(id), std::move(nid),
std::move(remote_dispatcher)),
proxy_dispatcher_{sys.spawn<proxy_dispatcher, hidden>(this)},
monitor_{sys.spawn<hidden>(dispatcher_monitor_actor, this)} {
// nop
}
bool subscribe(strong_actor_ptr who) override {
CAF_LOG_TRACE(CAF_ARG(who));
auto res = add_subscriber(std::move(who));
if (res.first) {
// join remote source
if (res.second == 1)
anon_send(dispatcher_, join_atom_v, proxy_dispatcher_);
return true;
}
CAF_LOG_WARNING("actor already joined group");
return false;
}
void unsubscribe(const actor_control_block* who) override {
CAF_LOG_TRACE(CAF_ARG(who));
auto res = erase_subscriber(who);
if (res.first && res.second == 0) {
// leave the remote source,
// because there's no more subscriber on this node
anon_send(dispatcher_, leave_atom_v, proxy_dispatcher_);
}
}
void enqueue(strong_actor_ptr sender, message_id mid, message msg,
execution_unit* eu) override {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(mid) << CAF_ARG(msg));
// forward message to the dispatcher
dispatcher_->enqueue(std::move(sender), mid,
make_message(forward_atom_v, std::move(msg)), eu);
}
void stop() override {
CAF_LOG_TRACE("");
await_all_locals_down(system_, {monitor_, proxy_dispatcher_, dispatcher_});
}
private:
static behavior dispatcher_monitor_actor(event_based_actor* self,
local_group_proxy* grp) {
CAF_LOG_TRACE("");
self->monitor(grp->dispatcher_);
self->set_down_handler([=](down_msg& down) {
CAF_LOG_TRACE(CAF_ARG(down));
auto msg = make_message(group_down_msg{group(grp)});
grp->send_all_subscribers(self->ctrl(), std::move(msg), self->context());
self->quit(down.reason);
});
return {[] {
// nop
}};
}
actor proxy_dispatcher_;
actor monitor_;
};
behavior proxy_dispatcher::make_behavior() {
CAF_LOG_TRACE("");
// instead of dropping "unexpected" messages,
// we simply forward them to our acquaintances
auto fwd = [=](local_actor*, message& msg) -> skippable_result {
group_->send_all_subscribers(current_element_->sender, std::move(msg),
context());
return message{};
};
set_default_handler(fwd);
// return dummy behavior
return {[] {
// nop
}};
}
class local_group_module : public group_module {
public:
local_group_module(actor_system& sys) : group_module(sys, "local") {
CAF_LOG_TRACE("");
}
expected<group> get(const std::string& identifier) override {
CAF_LOG_TRACE(CAF_ARG(identifier));
upgrade_guard guard(instances_mtx_);
auto i = instances_.find(identifier);
if (i != instances_.end())
return group{i->second};
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;
uguard.unlock();
// someone might preempt us
if (result != tmp)
tmp->stop();
return group{result};
}
expected<group> get(const std::string& identifier,
const caf::actor& dispatcher) override {
CAF_LOG_DEBUG(CAF_ARG(identifier) << CAF_ARG(dispatcher));
if (!dispatcher)
return make_error(sec::invalid_argument, "dispatcher == nullptr");
if (dispatcher->node() == system().node())
return get(identifier);
upgrade_guard guard(proxies_mtx_);
auto i = proxies_.find(dispatcher);
if (i != proxies_.end())
return group{i->second};
local_group_ptr tmp = make_counted<local_group_proxy>(system(), dispatcher,
*this, identifier,
dispatcher->node());
upgrade_to_unique_guard uguard(guard);
auto p = proxies_.emplace(dispatcher, tmp);
// someone might preempt us
return group{p.first->second};
}
void stop() override {
CAF_LOG_TRACE("");
std::map<std::string, local_group_ptr> imap;
std::map<actor, local_group_ptr> pmap;
{ // critical section
exclusive_guard guard1{instances_mtx_};
exclusive_guard guard2{proxies_mtx_};
imap.swap(instances_);
pmap.swap(proxies_);
}
for (auto& kvp : imap)
kvp.second->stop();
for (auto& kvp : pmap)
kvp.second->stop();
}
private:
detail::shared_spinlock instances_mtx_;
std::map<std::string, local_group_ptr> instances_;
detail::shared_spinlock proxies_mtx_;
std::map<actor, local_group_ptr> proxies_;
};
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)) {
CAF_LOG_TRACE(CAF_ARG(id) << CAF_ARG(nid));
dispatcher_ = lb ? *lb : mod.system().spawn<local_dispatcher, hidden>(this);
}
local_group::~local_group() {
// nop
}
std::atomic<size_t> s_ad_hoc_id;
} // namespace
void group_manager::init(actor_system_config& cfg) { void group_manager::init(actor_system_config& cfg) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
using ptr_type = std::unique_ptr<group_module>; mmap_.emplace("local", make_counted<detail::local_group_module>(*system_));
mmap_.emplace("local", ptr_type{new local_group_module(system_)});
for (auto& fac : cfg.group_module_factories) { for (auto& fac : cfg.group_module_factories) {
ptr_type ptr{fac()}; auto ptr = group_module_ptr{fac(), false};
std::string name = ptr->name(); auto name = ptr->name();
mmap_.emplace(std::move(name), std::move(ptr)); mmap_.emplace(std::move(name), std::move(ptr));
} }
} }
...@@ -415,19 +53,18 @@ group_manager::~group_manager() { ...@@ -415,19 +53,18 @@ group_manager::~group_manager() {
// nop // nop
} }
group_manager::group_manager(actor_system& sys) : system_(sys) { group_manager::group_manager(actor_system& sys) : system_(&sys) {
// nop // nop
} }
group group_manager::anonymous() const { group group_manager::anonymous() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
std::string id = "__#"; std::string id = "__#";
id += std::to_string(++s_ad_hoc_id); id += std::to_string(++ad_hoc_id_);
// local module is guaranteed to not return an error return get_local(id);
return *get_module("local")->get(id);
} }
expected<group> group_manager::get(std::string group_uri) const { expected<group> group_manager::get(std::string group_uri) {
CAF_LOG_TRACE(CAF_ARG(group_uri)); CAF_LOG_TRACE(CAF_ARG(group_uri));
// URI parsing is pretty much a brute-force approach, no actual validation yet // URI parsing is pretty much a brute-force approach, no actual validation yet
auto p = group_uri.find(':'); auto p = group_uri.find(':');
...@@ -440,38 +77,27 @@ expected<group> group_manager::get(std::string group_uri) const { ...@@ -440,38 +77,27 @@ expected<group> group_manager::get(std::string group_uri) const {
} }
expected<group> group_manager::get(const std::string& module_name, expected<group> group_manager::get(const std::string& module_name,
const std::string& group_identifier) const { const std::string& group_identifier) {
CAF_LOG_TRACE(CAF_ARG(module_name) << CAF_ARG(group_identifier)); CAF_LOG_TRACE(CAF_ARG(module_name) << CAF_ARG(group_identifier));
if (auto mod = get_module(module_name)) if (auto mod = get_module(module_name))
return mod->get(group_identifier); return mod->get(group_identifier);
std::string error_msg = R"(no module named ")"; std::string error_msg = R"__(no module named ")__";
error_msg += module_name;
error_msg += R"(" found)";
return make_error(sec::no_such_group_module, std::move(error_msg));
}
expected<group> group_manager::get(const std::string& module_name,
const std::string& group_identifier,
const actor& dispatcher) const {
CAF_LOG_TRACE(CAF_ARG(module_name) << CAF_ARG(group_identifier));
if (auto mod = get_module(module_name))
return mod->get(group_identifier, dispatcher);
std::string error_msg = R"(no module named ")";
error_msg += module_name; error_msg += module_name;
error_msg += R"(" found)"; error_msg += R"__(" found)__";
return make_error(sec::no_such_group_module, std::move(error_msg)); return make_error(sec::no_such_group_module, std::move(error_msg));
} }
optional<group_module&> group_manager::get_module(const std::string& x) const { group group_manager::get_local(const std::string& group_identifier) {
auto i = mmap_.find(x); auto result = get("local", group_identifier);
if (i != mmap_.end()) CAF_ASSERT(result);
return *(i->second); return std::move(*result);
return none;
} }
group group_manager::get_local(const std::string& group_identifier) const { group_module_ptr group_manager::get_module(const std::string& x) const {
// guaranteed to never return an error if (auto i = mmap_.find(x); i != mmap_.end())
return *get("local", group_identifier); return i->second;
else
return nullptr;
} }
} // namespace caf } // namespace caf
...@@ -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
} }
......
...@@ -119,11 +119,15 @@ void local_actor::monitor(const node_id& node) { ...@@ -119,11 +119,15 @@ void local_actor::monitor(const node_id& node) {
void local_actor::demonitor(const actor_addr& whom) { void local_actor::demonitor(const actor_addr& whom) {
CAF_LOG_TRACE(CAF_ARG(whom)); CAF_LOG_TRACE(CAF_ARG(whom));
auto ptr = actor_cast<strong_actor_ptr>(whom); demonitor(actor_cast<strong_actor_ptr>(whom));
if (ptr) { }
void local_actor::demonitor(const strong_actor_ptr& whom) {
CAF_LOG_TRACE(CAF_ARG(whom));
if (whom) {
default_attachable::observe_token tk{address(), default_attachable::observe_token tk{address(),
default_attachable::monitor}; default_attachable::monitor};
ptr->get()->detach(tk); whom->get()->detach(tk);
} }
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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.group_tunnel
#include "caf/detail/group_tunnel.hpp"
#include "caf/test/dsl.hpp"
using namespace caf;
namespace {
class mock_module : public group_module {
public:
using super = group_module;
explicit mock_module(actor_system& sys) : super(sys, "mock") {
// nop
}
void stop() override {
drop_instances();
}
expected<group> get(const std::string& group_name) override {
auto result = get_impl(group_name);
return group{result.get()};
}
detail::group_tunnel_ptr get_unconnected(const std::string& group_name,
const node_id& origin) {
if (auto i = instances.find(group_name); i != instances.end()) {
return i->second;
} else {
auto result = make_counted<detail::group_tunnel>(this, group_name,
origin);
instances.emplace(group_name, result);
return result;
}
}
void drop_instances() {
for (auto& kvp : instances)
kvp.second->stop();
instances.clear();
}
detail::group_tunnel_ptr get_impl(const std::string& group_name) {
if (auto i = instances.find(group_name); i != instances.end()) {
return i->second;
} else {
auto wrapped = system().groups().get_local(group_name);
auto result = make_counted<detail::group_tunnel>(
this, group_name, wrapped.get()->intermediary());
instances.emplace(group_name, result);
return result;
}
}
std::map<std::string, detail::group_tunnel_ptr> instances;
};
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() {
uut = make_counted<mock_module>(sys);
origin = sys.groups().get_local("test");
intermediary = origin.get()->intermediary();
tunnel = uut->get_impl("test");
proxy = group{tunnel.get()};
worker = tunnel->worker();
run();
}
~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 modules to
// stop here.
for (auto& kvp : uut->instances)
kvp.second->stop();
sys.groups().get_module("local")->stop();
}
void make_unconnected() {
uut->drop_instances();
tunnel = uut->get_unconnected("test", intermediary.node());
proxy = group{tunnel.get()};
worker = actor{};
run();
}
void connect_proxy() {
tunnel->connect(intermediary);
worker = tunnel->worker();
}
intrusive_ptr<mock_module> uut;
group origin;
actor intermediary;
detail::group_tunnel_ptr tunnel;
group proxy;
actor worker;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(group_tunnel_tests, fixture)
CAF_TEST(tunnels automatically subscribe to their origin on first subscribe) {
CAF_MESSAGE("Given a group with two subscribers and a tunnel.");
sys.spawn_in_group<lazy_init>(origin, testee_impl);
sys.spawn_in_group<lazy_init>(origin, testee_impl);
{ // Subtest.
CAF_MESSAGE("When an actors joins the tunnel.");
sys.spawn_in_group<lazy_init>(proxy, testee_impl);
CAF_MESSAGE("Then the tunnel worker joins the origin group.");
expect((sys_atom, join_atom), to(worker));
expect((join_atom, strong_actor_ptr),
from(worker).to(intermediary).with(_, worker));
CAF_CHECK(!sched.has_job());
}
{ // Subtest.
CAF_MESSAGE("When a second actor joins the tunnel.");
sys.spawn_in_group<lazy_init>(proxy, testee_impl);
CAF_MESSAGE("Then no messaging occurs.");
CAF_CHECK(!sched.has_job());
}
}
CAF_TEST("tunnels dispatch published messages") {
CAF_MESSAGE("Given a group with two local subscribers locally and tunneled.");
auto t1 = sys.spawn_in_group<lazy_init>(origin, testee_impl);
auto t2 = sys.spawn_in_group<lazy_init>(origin, testee_impl);
auto t3 = sys.spawn_in_group<lazy_init>(proxy, testee_impl);
auto t4 = sys.spawn_in_group<lazy_init>(proxy, testee_impl);
run();
{ // Subtest.
CAF_MESSAGE("When an actors sends to the group.");
self->send(origin, put_atom_v, 42);
CAF_MESSAGE("Then tunnel subscribers receive the forwarded message.");
expect((put_atom, int), from(self).to(t1).with(_, 42));
expect((put_atom, int), from(self).to(t2).with(_, 42));
expect((put_atom, int), from(self).to(worker).with(_, 42));
expect((put_atom, int), from(self).to(t3).with(_, 42));
expect((put_atom, int), from(self).to(t4).with(_, 42));
CAF_CHECK(!sched.has_job());
}
{ // Subtest.
CAF_MESSAGE("When an actors sends to the tunnel.");
self->send(proxy, put_atom_v, 42);
CAF_MESSAGE("Then the message travels to the origin.");
CAF_MESSAGE("And tunnel subscribers get the forwarded message eventually.");
expect((sys_atom, forward_atom, message), from(self).to(worker));
expect((forward_atom, message), from(self).to(intermediary));
expect((put_atom, int), from(self).to(t1).with(_, 42));
expect((put_atom, int), from(self).to(t2).with(_, 42));
expect((put_atom, int), from(self).to(worker).with(_, 42));
expect((put_atom, int), from(self).to(t3).with(_, 42));
expect((put_atom, int), from(self).to(t4).with(_, 42));
CAF_CHECK(!sched.has_job());
}
}
CAF_TEST(tunnels automatically unsubscribe from their origin) {
CAF_MESSAGE("Given a group with two local subscribers locally and tunneled.");
auto t1 = sys.spawn_in_group<lazy_init>(origin, testee_impl);
auto t2 = sys.spawn_in_group<lazy_init>(origin, testee_impl);
auto t3 = sys.spawn_in_group<lazy_init>(proxy, testee_impl);
auto t4 = sys.spawn_in_group<lazy_init>(proxy, testee_impl);
run();
{ // Subtest.
CAF_MESSAGE("When the first actor leaves the tunnel.");
proxy.unsubscribe(actor_cast<actor_control_block*>(t3));
CAF_MESSAGE("Then no messaging occurs.");
CAF_CHECK(!sched.has_job());
}
{ // Subtest.
CAF_MESSAGE("When an actors sends to the group after the unsubscribe.");
self->send(origin, put_atom_v, 42);
CAF_MESSAGE("Then the unsubscribed actor no longer receives the message.");
expect((put_atom, int), from(self).to(t1).with(_, 42));
expect((put_atom, int), from(self).to(t2).with(_, 42));
expect((put_atom, int), from(self).to(worker).with(_, 42));
disallow((put_atom, int), from(self).to(t3).with(_, 42));
expect((put_atom, int), from(self).to(t4).with(_, 42));
CAF_CHECK(!sched.has_job());
}
{ // Subtest.
CAF_MESSAGE("When the second actor also unsubscribes from the tunnel.");
proxy.unsubscribe(actor_cast<actor_control_block*>(t4));
CAF_MESSAGE("Then the tunnel unsubscribes from its origin.");
expect((sys_atom, leave_atom), to(worker));
expect((leave_atom, strong_actor_ptr),
from(worker).to(intermediary).with(_, worker));
}
{ // Subtest.
CAF_MESSAGE("When an actors sends to the group after the tunnel left.");
self->send(origin, put_atom_v, 42);
CAF_MESSAGE("Then no message arrives at the tunnel.");
expect((put_atom, int), from(self).to(t1).with(_, 42));
expect((put_atom, int), from(self).to(t2).with(_, 42));
disallow((put_atom, int), from(self).to(worker).with(_, 42));
CAF_CHECK(!sched.has_job());
}
}
CAF_TEST(tunnels cache messages until connected) {
CAF_MESSAGE("Given an unconnected tunnel with two subscribers.");
make_unconnected();
auto t1 = sys.spawn_in_group<lazy_init>(proxy, testee_impl);
auto t2 = sys.spawn_in_group<lazy_init>(proxy, testee_impl);
{ // Subtest.
CAF_MESSAGE("When an actors sends to the group.");
self->send(proxy, put_atom_v, 1);
self->send(proxy, put_atom_v, 2);
self->send(proxy, put_atom_v, 3);
CAF_MESSAGE("Then unconnected tunnel caches the messages.");
CAF_CHECK(!sched.has_job());
}
{ // Subtest.
CAF_MESSAGE("When the tunnel becomes connected.");
connect_proxy();
CAF_MESSAGE("Then tunnel subscribes upstream and flushes its cache.");
expect((sys_atom, join_atom), to(worker));
expect((sys_atom, forward_atom, message), from(self).to(worker));
expect((sys_atom, forward_atom, message), from(self).to(worker));
expect((sys_atom, forward_atom, message), from(self).to(worker));
expect((join_atom, strong_actor_ptr),
from(worker).to(intermediary).with(_, worker));
expect((forward_atom, message), from(self).to(intermediary));
expect((forward_atom, message), from(self).to(intermediary));
expect((forward_atom, message), from(self).to(intermediary));
expect((put_atom, int), from(self).to(worker).with(_, 1));
expect((put_atom, int), from(self).to(worker).with(_, 2));
expect((put_atom, int), from(self).to(worker).with(_, 3));
expect((put_atom, int), from(self).to(t1).with(_, 1));
expect((put_atom, int), from(self).to(t1).with(_, 2));
expect((put_atom, int), from(self).to(t1).with(_, 3));
expect((put_atom, int), from(self).to(t2).with(_, 1));
expect((put_atom, int), from(self).to(t2).with(_, 2));
expect((put_atom, int), from(self).to(t2).with(_, 3));
CAF_CHECK(!sched.has_job());
}
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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()
...@@ -29,6 +29,7 @@ endfunction() ...@@ -29,6 +29,7 @@ endfunction()
add_library(libcaf_io_obj OBJECT ${CAF_IO_HEADERS} add_library(libcaf_io_obj OBJECT ${CAF_IO_HEADERS}
src/detail/prometheus_broker.cpp src/detail/prometheus_broker.cpp
src/detail/remote_group_module.cpp
src/detail/socket_guard.cpp src/detail/socket_guard.cpp
src/io/abstract_broker.cpp src/io/abstract_broker.cpp
src/io/basp/header.cpp src/io/basp/header.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#pragma once
#include "caf/detail/group_tunnel.hpp"
#include "caf/detail/io_export.hpp"
#include "caf/fwd.hpp"
#include "caf/group.hpp"
#include "caf/group_module.hpp"
#include "caf/io/fwd.hpp"
#include <mutex>
#include <string>
#include <unordered_map>
namespace caf::detail {
class CAF_IO_EXPORT remote_group_module : public group_module {
public:
using super = group_module;
using instances_map = std::unordered_map<std::string, group_tunnel_ptr>;
using nodes_map = std::unordered_map<node_id, instances_map>;
explicit remote_group_module(io::middleman* mm);
void stop() override;
expected<group> get(const std::string& group_name) override;
// Get instance if it exists or create an unconnected tunnel and ask the
// middleman to connect it lazily.
group_tunnel_ptr get_impl(const node_id& origin,
const std::string& group_name);
// Get instance if it exists or create a connected tunnel.
group_tunnel_ptr get_impl(actor intermediary, const std::string& group_name);
// Get instance if it exists or return `nullptr`.
group_tunnel_ptr lookup(const node_id& origin, const std::string& group_name);
private:
template <class F>
auto critical_section(F&& fun) {
std::unique_lock<std::mutex> guard{mtx_};
return fun();
}
// Stops an instance and removes it from this module.
void drop(const group_tunnel_ptr& instance);
// Connects an instance when it is still associated to this module.
void connect(const group_tunnel_ptr& instance, actor intermediary);
std::function<void(actor)> make_callback(const group_tunnel_ptr& instance);
// Note: the actor system stops the group module before shutting down the
// middleman. Hence, it's safe to hold onto a raw pointer here.
io::middleman* mm_;
std::mutex mtx_;
bool stopped_ = false;
nodes_map nodes_;
};
using remote_group_module_ptr = intrusive_ptr<remote_group_module>;
} // namespace caf::detail
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/config_value.hpp" #include "caf/config_value.hpp"
#include "caf/detail/io_export.hpp" #include "caf/detail/io_export.hpp"
#include "caf/detail/remote_group_module.hpp"
#include "caf/detail/unique_function.hpp" #include "caf/detail/unique_function.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
...@@ -111,12 +112,22 @@ public: ...@@ -111,12 +112,22 @@ public:
return actor_cast<ActorHandle>(std::move(*x)); return actor_cast<ActorHandle>(std::move(*x));
} }
/// <group-name>@<host>:<port> /// Tries to connect to a group that runs on a different node in the network.
expected<group> remote_group(const std::string& group_uri); /// @param group_locator Locator in the format `<group-name>@<host>:<port>`.
expected<group> remote_group(const std::string& group_locator);
/// Tries to connect to a group that runs on a different node in the network.
/// @param group_identifier Unique identifier of the group.
/// @param host Hostname or IP address of the remote CAF node.
/// @param port TCP port for connecting to the group name server of the node.
expected<group> remote_group(const std::string& group_identifier, expected<group> remote_group(const std::string& group_identifier,
const std::string& host, uint16_t port); const std::string& host, uint16_t port);
/// @private
void resolve_remote_group_intermediary(const node_id& origin,
const std::string& group_identifier,
std::function<void(actor)> callback);
/// Returns the enclosing actor system. /// Returns the enclosing actor system.
actor_system& system() { actor_system& system() {
return system_; return system_;
...@@ -371,6 +382,9 @@ private: ...@@ -371,6 +382,9 @@ private:
/// Stores hidden background actors that get killed automatically when the /// Stores hidden background actors that get killed automatically when the
/// actor systems shuts down. /// actor systems shuts down.
std::list<actor> background_brokers_; std::list<actor> background_brokers_;
/// Manages groups that run on a different node in the network.
detail::remote_group_module_ptr remote_groups_;
}; };
} // namespace caf::io } // namespace caf::io
...@@ -101,6 +101,8 @@ using middleman_actor = typed_actor< ...@@ -101,6 +101,8 @@ using middleman_actor = typed_actor<
replies_to<spawn_atom, node_id, std::string, message, replies_to<spawn_atom, node_id, std::string, message,
std::set<std::string>>::with<strong_actor_ptr>, std::set<std::string>>::with<strong_actor_ptr>,
replies_to<get_atom, group_atom, node_id, std::string>::with<actor>,
replies_to<get_atom, node_id>::with<node_id, std::string, uint16_t>>; replies_to<get_atom, node_id>::with<node_id, std::string, uint16_t>>;
/// Spawns the default implementation for the `middleman_actor` interface. /// Spawns the default implementation for the `middleman_actor` interface.
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/remote_group_module.hpp"
#include "caf/detail/group_tunnel.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/io/middleman.hpp"
#include "caf/make_counted.hpp"
#include "caf/sec.hpp"
namespace caf::detail {
remote_group_module::remote_group_module(io::middleman* mm)
: super(mm->system(), "remote"), mm_(mm) {
// nop
}
void remote_group_module::stop() {
nodes_map tmp;
critical_section([this, &tmp] {
using std::swap;
if (!stopped_) {
stopped_ = true;
swap(nodes_, tmp);
}
});
for (auto& nodes_kvp : tmp)
for (auto& instances_kvp : nodes_kvp.second)
instances_kvp.second.get()->stop();
}
expected<group> remote_group_module::get(const std::string& group_locator) {
return mm_->remote_group(group_locator);
}
group_tunnel_ptr remote_group_module::get_impl(const node_id& origin,
const std::string& group_name) {
CAF_ASSERT(origin != none);
bool lazy_connect = false;
auto instance = critical_section([&, this] {
if (stopped_) {
return group_tunnel_ptr{nullptr};
} else {
auto& instances = nodes_[origin];
if (auto i = instances.find(group_name); i != instances.end()) {
return i->second;
} else {
lazy_connect = true;
auto instance = make_counted<detail::group_tunnel>(this, group_name,
origin);
instances.emplace(group_name, instance);
return instance;
}
}
});
if (lazy_connect)
mm_->resolve_remote_group_intermediary(origin, group_name,
make_callback(instance));
return instance;
}
group_tunnel_ptr remote_group_module::get_impl(actor intermediary,
const std::string& group_name) {
CAF_ASSERT(intermediary != nullptr);
return critical_section([&, this]() {
if (stopped_) {
return group_tunnel_ptr{nullptr};
} else {
auto& instances = nodes_[intermediary.node()];
if (auto i = instances.find(group_name); i != instances.end()) {
auto instance = i->second;
instance->connect(std::move(intermediary));
return instance;
} else {
auto instance = make_counted<detail::group_tunnel>(
this, group_name, std::move(intermediary));
instances.emplace(group_name, instance);
return instance;
}
}
});
}
group_tunnel_ptr remote_group_module::lookup(const node_id& origin,
const std::string& group_name) {
return critical_section([&, this] {
if (auto i = nodes_.find(origin); i != nodes_.end())
if (auto j = i->second.find(group_name); j != i->second.end())
return j->second;
return group_tunnel_ptr{nullptr};
});
}
void remote_group_module::drop(const group_tunnel_ptr& instance) {
CAF_ASSERT(instance != nullptr);
critical_section([&, this] {
if (auto i = nodes_.find(instance->origin()); i != nodes_.end()) {
if (auto j = i->second.find(instance->identifier());
j != i->second.end()) {
i->second.erase(j);
if (i->second.empty())
nodes_.erase(i);
}
}
});
instance->stop();
}
void remote_group_module::connect(const group_tunnel_ptr& instance,
actor intermediary) {
CAF_ASSERT(instance != nullptr);
bool stop_instance = critical_section([&, this] {
if (auto i = nodes_.find(instance->origin()); i != nodes_.end()) {
if (auto j = i->second.find(instance->identifier());
j != i->second.end() && j->second == instance) {
instance->connect(intermediary);
return false;
}
}
return true;
});
if (stop_instance)
instance->stop();
}
std::function<void(actor)>
remote_group_module::make_callback(const group_tunnel_ptr& instance) {
return [instance, strong_this{remote_group_module_ptr{this}}](actor hdl) {
if (hdl == nullptr)
strong_this->drop(instance);
else
strong_this->connect(instance, std::move(hdl));
};
}
} // namespace caf::detail
...@@ -40,6 +40,7 @@ ...@@ -40,6 +40,7 @@
#include "caf/detail/set_thread_name.hpp" #include "caf/detail/set_thread_name.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/function_view.hpp" #include "caf/function_view.hpp"
#include "caf/group_module.hpp"
#include "caf/init_global_meta_objects.hpp" #include "caf/init_global_meta_objects.hpp"
#include "caf/io/basp/header.hpp" #include "caf/io/basp/header.hpp"
#include "caf/io/basp_broker.hpp" #include "caf/io/basp_broker.hpp"
...@@ -118,7 +119,7 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) { ...@@ -118,7 +119,7 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
} }
middleman::middleman(actor_system& sys) : system_(sys) { middleman::middleman(actor_system& sys) : system_(sys) {
// nop remote_groups_ = make_counted<detail::remote_group_module>(this);
} }
expected<strong_actor_ptr> expected<strong_actor_ptr>
...@@ -168,19 +169,20 @@ expected<uint16_t> middleman::publish_local_groups(uint16_t port, ...@@ -168,19 +169,20 @@ expected<uint16_t> middleman::publish_local_groups(uint16_t port,
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in)); CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in));
auto group_nameserver = [](event_based_actor* self) -> behavior { auto group_nameserver = [](event_based_actor* self) -> behavior {
return { return {
[self](get_atom, const std::string& name) { [self](get_atom, const std::string& id) {
return self->system().groups().get_local(name); auto grp = self->system().groups().get_local(id);
return grp.get()->intermediary();
}, },
}; };
}; };
auto gn = system().spawn<hidden + lazy_init>(group_nameserver); auto ns = system().spawn<hidden + lazy_init>(group_nameserver);
auto result = publish(gn, port, in, reuse); if (auto result = publish(ns, port, in, reuse)) {
// link gn to our manager manager_->add_link(actor_cast<abstract_actor*>(ns));
if (result) return *result;
manager_->add_link(actor_cast<abstract_actor*>(gn)); } else {
else anon_send_exit(ns, exit_reason::user_shutdown);
anon_send_exit(gn, exit_reason::user_shutdown); return std::move(result.error());
return result; }
} }
expected<void> middleman::unpublish(const actor_addr& whom, uint16_t port) { expected<void> middleman::unpublish(const actor_addr& whom, uint16_t port) {
...@@ -228,8 +230,8 @@ expected<group> middleman::remote_group(const std::string& group_identifier, ...@@ -228,8 +230,8 @@ expected<group> middleman::remote_group(const std::string& group_identifier,
CAF_LOG_TRACE(CAF_ARG(group_identifier) << CAF_ARG(host) << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(group_identifier) << CAF_ARG(host) << CAF_ARG(port));
// Helper actor that first connects to the remote actor at `host:port` and // Helper actor that first connects to the remote actor at `host:port` and
// then tries to get a valid group from that actor. // then tries to get a valid group from that actor.
auto two_step_lookup = [=](event_based_actor* self, auto two_step_lookup = [=](event_based_actor* self, middleman_actor mm,
middleman_actor mm) -> behavior { detail::remote_group_module_ptr rgm) -> behavior {
return { return {
[=](get_atom) { [=](get_atom) {
/// We won't receive a second message, so we drop our behavior here to /// We won't receive a second message, so we drop our behavior here to
...@@ -242,9 +244,12 @@ expected<group> middleman::remote_group(const std::string& group_identifier, ...@@ -242,9 +244,12 @@ expected<group> middleman::remote_group(const std::string& group_identifier,
CAF_LOG_DEBUG("received handle to target node:" << CAF_ARG(ptr)); CAF_LOG_DEBUG("received handle to target node:" << CAF_ARG(ptr));
auto hdl = actor_cast<actor>(ptr); auto hdl = actor_cast<actor>(ptr);
self->request(hdl, infinite, get_atom_v, group_identifier) self->request(hdl, infinite, get_atom_v, group_identifier)
.then([=](group& grp) mutable { .then([=](actor intermediary) mutable {
CAF_LOG_DEBUG("received remote group handle:" << CAF_ARG(grp)); CAF_LOG_DEBUG("lookup successful:" << CAF_ARG(group_identifier)
rp.deliver(std::move(grp)); << CAF_ARG(intermediary));
auto ptr = rgm->get_impl(intermediary, group_identifier);
auto result = group{ptr.get()};
rp.deliver(std::move(result));
}); });
}); });
}, },
...@@ -254,7 +259,8 @@ expected<group> middleman::remote_group(const std::string& group_identifier, ...@@ -254,7 +259,8 @@ expected<group> middleman::remote_group(const std::string& group_identifier,
expected<group> result{sec::cannot_connect_to_node}; expected<group> result{sec::cannot_connect_to_node};
scoped_actor self{system(), true}; scoped_actor self{system(), true};
auto worker = self->spawn<lazy_init + monitored>(two_step_lookup, auto worker = self->spawn<lazy_init + monitored>(two_step_lookup,
actor_handle()); actor_handle(),
remote_groups_);
self->send(worker, get_atom_v); self->send(worker, get_atom_v);
self->receive( self->receive(
[&](group& grp) { [&](group& grp) {
...@@ -272,6 +278,20 @@ expected<group> middleman::remote_group(const std::string& group_identifier, ...@@ -272,6 +278,20 @@ expected<group> middleman::remote_group(const std::string& group_identifier,
return result; return result;
} }
void middleman::resolve_remote_group_intermediary(
const node_id& origin, const std::string& group_identifier,
std::function<void(actor)> callback) {
auto lookup = [=, cb{std::move(callback)}](event_based_actor* self,
middleman_actor mm) {
self
->request(mm, std::chrono::minutes(1), get_atom_v, group_atom_v, origin,
group_identifier)
.then([cb](actor& hdl) { cb(std::move(hdl)); },
[cb](const error&) { cb(actor{}); });
};
system().spawn(lookup, actor_handle());
}
strong_actor_ptr middleman::remote_lookup(std::string name, strong_actor_ptr middleman::remote_lookup(std::string name,
const node_id& nid) { const node_id& nid) {
CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(nid)); CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(nid));
...@@ -334,6 +354,19 @@ void middleman::start() { ...@@ -334,6 +354,19 @@ void middleman::start() {
if (auto prom = get_if<dict>(&system().config(), if (auto prom = get_if<dict>(&system().config(),
"caf.middleman.prometheus-http")) "caf.middleman.prometheus-http"))
expose_prometheus_metrics(*prom); expose_prometheus_metrics(*prom);
// Enable deserialization of groups.
system().groups().get_remote
= [this](const node_id& origin, const std::string& module_name,
const std::string& group_identifier) -> expected<group> {
if (module_name == "local" || module_name == "remote") {
auto ptr = remote_groups_->get_impl(origin, group_identifier);
return group{ptr.get()};
} else {
return make_error(
sec::runtime_error,
"currently, only 'local' groups are accessible remotely");
}
};
} }
void middleman::stop() { void middleman::stop() {
...@@ -388,6 +421,13 @@ void middleman::init(actor_system_config& cfg) { ...@@ -388,6 +421,13 @@ void middleman::init(actor_system_config& cfg) {
system().node_.swap(this_node); system().node_.swap(this_node);
// Give config access to slave mode implementation. // Give config access to slave mode implementation.
cfg.slave_mode_fun = &middleman::exec_slave_mode; cfg.slave_mode_fun = &middleman::exec_slave_mode;
// Enable users to use 'remote:foo@bar' notation for remote groups.
auto dummy_fac = [ptr{remote_groups_}]() -> group_module* {
auto raw = ptr.get();
raw->ref();
return raw;
};
cfg.group_module_factories.emplace_back(dummy_fac);
} }
expected<uint16_t> middleman::expose_prometheus_metrics(uint16_t port, expected<uint16_t> middleman::expose_prometheus_metrics(uint16_t port,
......
...@@ -148,7 +148,7 @@ auto middleman_actor_impl::make_behavior() -> behavior_type { ...@@ -148,7 +148,7 @@ auto middleman_actor_impl::make_behavior() -> behavior_type {
delegate(broker_, atm, p); delegate(broker_, atm, p);
return {}; return {};
}, },
[=](spawn_atom atm, node_id& nid, std::string& name, message& args, [=](spawn_atom, node_id& nid, std::string& name, message& args,
std::set<std::string>& ifs) -> result<strong_actor_ptr> { std::set<std::string>& ifs) -> result<strong_actor_ptr> {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (!nid) if (!nid)
...@@ -168,14 +168,25 @@ auto middleman_actor_impl::make_behavior() -> behavior_type { ...@@ -168,14 +168,25 @@ auto middleman_actor_impl::make_behavior() -> behavior_type {
// reference but spawn_server_id is constexpr). // reference but spawn_server_id is constexpr).
auto id = basp::header::spawn_server_id; auto id = basp::header::spawn_server_id;
delegate(broker_, forward_atom_v, nid, id, delegate(broker_, forward_atom_v, nid, id,
make_message(atm, std::move(name), std::move(args), make_message(spawn_atom_v, std::move(name), std::move(args),
std::move(ifs))); std::move(ifs)));
return delegated<strong_actor_ptr>{}; return delegated<strong_actor_ptr>{};
}, },
[=](get_atom atm, [=](get_atom, group_atom, node_id& nid,
node_id nid) -> delegated<node_id, std::string, uint16_t> { std::string& group_id) -> result<actor> {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
delegate(broker_, atm, std::move(nid)); if (!nid)
return make_error(sec::invalid_argument,
"cannot get group intermediaries from invalid nodes");
auto id = basp::header::config_server_id;
delegate(broker_, forward_atom_v, nid, id,
make_message(get_atom_v, group_atom_v, std::move(nid),
std::move(group_id)));
return delegated<actor>{};
},
[=](get_atom, node_id& nid) -> delegated<node_id, std::string, uint16_t> {
CAF_LOG_TRACE("");
delegate(broker_, get_atom_v, std::move(nid));
return {}; return {};
}, },
}; };
......
...@@ -16,10 +16,11 @@ ...@@ -16,10 +16,11 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#define CAF_SUITE io.remote_group
#include "caf/config.hpp" #include "caf/config.hpp"
#define CAF_SUITE io_dynamic_remote_group #include "io-test.hpp"
#include "caf/test/io_dsl.hpp"
#include <algorithm> #include <algorithm>
#include <vector> #include <vector>
...@@ -48,7 +49,9 @@ size_t received_messages = 0u; ...@@ -48,7 +49,9 @@ size_t received_messages = 0u;
behavior group_receiver(event_based_actor* self) { behavior group_receiver(event_based_actor* self) {
self->set_default_handler(reflect_and_quit); self->set_default_handler(reflect_and_quit);
return {[](ok_atom) { ++received_messages; }}; return {
[](ok_atom) { ++received_messages; },
};
} }
// Our server is `mars` and our client is `earth`. // Our server is `mars` and our client is `earth`.
...@@ -90,7 +93,7 @@ CAF_TEST(connecting to remote group) { ...@@ -90,7 +93,7 @@ CAF_TEST(connecting to remote group) {
CAF_CHECK_EQUAL(grp->get()->identifier(), group_name); CAF_CHECK_EQUAL(grp->get()->identifier(), group_name);
} }
CAF_TEST_DISABLED(message transmission) { CAF_TEST(message transmission) {
CAF_MESSAGE("spawn 5 receivers on mars"); CAF_MESSAGE("spawn 5 receivers on mars");
auto mars_grp = mars.sys.groups().get_local(group_name); auto mars_grp = mars.sys.groups().get_local(group_name);
spawn_receivers(mars, mars_grp, 5u); spawn_receivers(mars, mars_grp, 5u);
...@@ -102,19 +105,18 @@ CAF_TEST_DISABLED(message transmission) { ...@@ -102,19 +105,18 @@ CAF_TEST_DISABLED(message transmission) {
auto earth_grp = unbox(earth.mm.remote_group(group_name, server, port)); auto earth_grp = unbox(earth.mm.remote_group(group_name, server, port));
CAF_MESSAGE("spawn 5 more receivers on earth"); CAF_MESSAGE("spawn 5 more receivers on earth");
spawn_receivers(earth, earth_grp, 5u); spawn_receivers(earth, earth_grp, 5u);
exec_all();
CAF_MESSAGE("send message on mars and expect 10 handled messages total"); CAF_MESSAGE("send message on mars and expect 10 handled messages total");
{ {
received_messages = 0u; received_messages = 0u;
scoped_actor self{mars.sys}; mars.self->send(mars_grp, ok_atom_v);
self->send(mars_grp, ok_atom_v);
exec_all(); exec_all();
CAF_CHECK_EQUAL(received_messages, 10u); CAF_CHECK_EQUAL(received_messages, 10u);
} }
CAF_MESSAGE("send message on earth and again expect 10 handled messages"); CAF_MESSAGE("send message on earth and again expect 10 handled messages");
{ {
received_messages = 0u; received_messages = 0u;
scoped_actor self{earth.sys}; earth.self->send(earth_grp, ok_atom_v);
self->send(earth_grp, ok_atom_v);
exec_all(); exec_all();
CAF_CHECK_EQUAL(received_messages, 10u); CAF_CHECK_EQUAL(received_messages, 10u);
} }
......
...@@ -458,24 +458,24 @@ public: ...@@ -458,24 +458,24 @@ public:
void with(Ts... xs) { void with(Ts... xs) {
if (dest_ == nullptr) if (dest_ == nullptr)
CAF_FAIL("missing .to() in inject() statement"); CAF_FAIL("missing .to() in inject() statement");
auto msg = caf::make_message(std::move(xs)...);
if (src_ == nullptr) if (src_ == nullptr)
caf::anon_send(caf::actor_cast<caf::actor>(dest_), xs...); caf::anon_send(caf::actor_cast<caf::actor>(dest_), msg);
else else
caf::send_as(caf::actor_cast<caf::actor>(src_), caf::send_as(caf::actor_cast<caf::actor>(src_),
caf::actor_cast<caf::actor>(dest_), xs...); caf::actor_cast<caf::actor>(dest_), msg);
CAF_REQUIRE(sched_.prioritize(dest_)); if (!sched_.prioritize(dest_))
CAF_FAIL("inject: failed to schedule destination actor");
auto dest_ptr = &sched_.next_job<caf::abstract_actor>(); auto dest_ptr = &sched_.next_job<caf::abstract_actor>();
auto ptr = dest_ptr->peek_at_next_mailbox_element(); auto ptr = dest_ptr->peek_at_next_mailbox_element();
CAF_REQUIRE(ptr != nullptr); if (ptr == nullptr)
CAF_REQUIRE_EQUAL(ptr->sender, src_); CAF_FAIL("inject: failed to get next message from destination actor");
// TODO: replace this workaround with the make_tuple() line when dropping if (ptr->sender != src_)
// support for GCC 4.8. CAF_FAIL("inject: found unexpected sender for the next message");
std::tuple<Ts...> tmp{std::move(xs)...}; if (ptr->payload.cptr() != msg.cptr())
using namespace caf::detail; CAF_FAIL("inject: found unexpected message => " << ptr->payload << " !! "
elementwise_compare_inspector<decltype(tmp)> inspector{tmp}; << msg);
auto ys = extract<Ts...>(dest_); msg.reset(); // drop local reference before running the actor
auto ys_indices = get_indices(ys);
CAF_REQUIRE(apply_args(inspector, ys_indices, ys));
run_once(); run_once();
} }
......
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