Commit 713eb9b8 authored by Dominik Charousset's avatar Dominik Charousset

Implement lazy connections for group tunnels

parent 8ceb5019
...@@ -79,7 +79,7 @@ public: ...@@ -79,7 +79,7 @@ public:
} }
/// Returns a human-readable string representation of the group ID. /// Returns a human-readable string representation of the group ID.
virtual std::string to_string() const; virtual std::string stringify() const;
/// Returns the intermediary actor for the group if applicable. /// Returns the intermediary actor for the group if applicable.
virtual actor intermediary() const noexcept; virtual actor intermediary() const noexcept;
......
...@@ -20,21 +20,33 @@ ...@@ -20,21 +20,33 @@
#include "caf/abstract_group.hpp" #include "caf/abstract_group.hpp"
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/local_group_module.hpp" #include "caf/detail/local_group_module.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/stateful_actor.hpp" #include "caf/stateful_actor.hpp"
#include <tuple>
namespace caf::detail { namespace caf::detail {
/// Represents a group that runs on a different CAF node. /// Represents a group that runs on a different CAF node.
class group_tunnel : public local_group_module::impl { class CAF_CORE_EXPORT group_tunnel : public local_group_module::impl {
public: public:
using super = local_group_module::impl; using super = local_group_module::impl;
using cached_message = std::tuple<strong_actor_ptr, message_id, message>;
using cached_message_list = std::vector<cached_message>;
// Creates a connected tunnel.
group_tunnel(group_module_ptr mod, std::string id, group_tunnel(group_module_ptr mod, std::string id,
actor upstream_intermediary); 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; ~group_tunnel() override;
bool subscribe(strong_actor_ptr who) override; bool subscribe(strong_actor_ptr who) override;
...@@ -47,16 +59,21 @@ public: ...@@ -47,16 +59,21 @@ public:
void stop() override; void stop() override;
std::string stringify() const override;
// Messages received from the upstream group, forwarded to local subscribers. // Messages received from the upstream group, forwarded to local subscribers.
void upstream_enqueue(strong_actor_ptr sender, message_id mid, void upstream_enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* host); message content, execution_unit* host);
auto worker() const noexcept { bool connect(actor upstream_intermediary);
return worker_;
} bool connected() const noexcept;
actor worker() const noexcept;
private: private:
actor worker_; actor worker_;
cached_message_list cached_messages_;
}; };
using group_tunnel_ptr = intrusive_ptr<group_tunnel>; using group_tunnel_ptr = intrusive_ptr<group_tunnel>;
......
...@@ -33,7 +33,7 @@ namespace caf::detail { ...@@ -33,7 +33,7 @@ namespace caf::detail {
/// ID as prefix. Each group instance spins up an intermediary actor to enable /// 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 /// remote access to the group. The default module is used internally by the
/// "local" as well as the "remote" module. /// "local" as well as the "remote" module.
class local_group_module : public group_module { class CAF_CORE_EXPORT local_group_module : public group_module {
public: public:
using super = group_module; using super = group_module;
...@@ -54,7 +54,7 @@ public: ...@@ -54,7 +54,7 @@ public:
using intermediary_actor = stateful_actor<intermediary_actor_state>; using intermediary_actor = stateful_actor<intermediary_actor_state>;
/// Implementation of the group interface for instances of this module. /// Implementation of the group interface for instances of this module.
class impl : public abstract_group { class CAF_CORE_EXPORT impl : public abstract_group {
public: public:
friend local_group_module; friend local_group_module;
...@@ -62,6 +62,8 @@ public: ...@@ -62,6 +62,8 @@ public:
using subscriber_set = std::set<strong_actor_ptr, std::less<>>; 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(group_module_ptr mod, std::string id);
~impl() override; ~impl() override;
...@@ -79,7 +81,7 @@ public: ...@@ -79,7 +81,7 @@ public:
protected: protected:
template <class F> template <class F>
auto critical_section(F&& fun) { auto critical_section(F&& fun) const {
std::unique_lock<std::mutex> guard{mtx_}; std::unique_lock<std::mutex> guard{mtx_};
return fun(); return fun();
} }
......
...@@ -53,17 +53,15 @@ public: ...@@ -53,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);
explicit 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_);
......
...@@ -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)
......
...@@ -44,7 +44,7 @@ actor_system& abstract_group::system() const noexcept { ...@@ -44,7 +44,7 @@ actor_system& abstract_group::system() const noexcept {
return module().system(); return module().system();
} }
std::string abstract_group::to_string() const { std::string abstract_group::stringify() const {
auto result = module().name(); auto result = module().name();
result += ':'; result += ':';
result += identifier(); result += identifier();
......
...@@ -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();
},
}; };
} }
......
...@@ -34,9 +34,6 @@ struct group_worker_actor_state { ...@@ -34,9 +34,6 @@ struct group_worker_actor_state {
behavior make_behavior() { behavior make_behavior() {
self->set_down_handler([this](const down_msg& dm) { self->set_down_handler([this](const down_msg& dm) {
if (dm.source == intermediary) { if (dm.source == intermediary) {
gptr->upstream_enqueue(nullptr, make_message_id(),
make_message(group_down_msg{group{gptr.get()}}),
self->context());
gptr->stop(); gptr->stop();
} }
}); });
...@@ -74,9 +71,14 @@ using group_worker_actor = stateful_actor<group_worker_actor_state>; ...@@ -74,9 +71,14 @@ using group_worker_actor = stateful_actor<group_worker_actor_state>;
group_tunnel::group_tunnel(group_module_ptr mod, std::string id, group_tunnel::group_tunnel(group_module_ptr mod, std::string id,
actor upstream_intermediary) actor upstream_intermediary)
: super(std::move(mod), std::move(id)) { : super(std::move(mod), std::move(id), upstream_intermediary.node()) {
intermediary_ = std::move(upstream_intermediary); intermediary_ = std::move(upstream_intermediary);
worker_ = system().spawn<group_worker_actor>(this, 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() { group_tunnel::~group_tunnel() {
...@@ -102,36 +104,86 @@ void group_tunnel::unsubscribe(const actor_control_block* who) { ...@@ -102,36 +104,86 @@ void group_tunnel::unsubscribe(const actor_control_block* who) {
void group_tunnel::enqueue(strong_actor_ptr sender, message_id mid, void group_tunnel::enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* host) { message content, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(content));
std::unique_lock<std::mutex> guard{mtx_}; std::unique_lock<std::mutex> guard{mtx_};
if (!stopped_) { if (worker_ != nullptr) {
auto wrapped = make_message(sys_atom_v, forward_atom_v, std::move(content)); auto wrapped = make_message(sys_atom_v, forward_atom_v, std::move(content));
worker_->enqueue(std::move(sender), mid, std::move(wrapped), host); 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() { void group_tunnel::stop() {
CAF_LOG_TRACE("");
CAF_LOG_DEBUG("stop group tunnel:" << CAF_ARG2("module", module().name()) CAF_LOG_DEBUG("stop group tunnel:" << CAF_ARG2("module", module().name())
<< CAF_ARG2("identifier", identifier_)); << CAF_ARG2("identifier", identifier_));
auto hdl = actor{}; auto hdl = actor{};
auto subs = subscriber_set{}; auto subs = subscriber_set{};
auto stopped = critical_section([this, &hdl, &subs] { auto cache = cached_message_list{};
auto stopped = critical_section([this, &hdl, &subs, &cache] {
using std::swap; using std::swap;
if (!stopped_) { if (!stopped_) {
stopped_ = true; stopped_ = true;
swap(subs, subscribers_); swap(subs, subscribers_);
swap(hdl, worker_); swap(hdl, worker_);
swap(cache, cached_messages_);
return true; return true;
} else { } else {
return false; return false;
} }
}); });
if (stopped) if (stopped) {
anon_send_exit(hdl, exit_reason::user_shutdown); 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, void group_tunnel::upstream_enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* host) { message content, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(content));
super::enqueue(std::move(sender), mid, std::move(content), host); 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 } // namespace caf::detail
...@@ -65,13 +65,19 @@ behavior local_group_module::intermediary_actor_state::make_behavior() { ...@@ -65,13 +65,19 @@ behavior local_group_module::intermediary_actor_state::make_behavior() {
// -- local group impl --------------------------------------------------------- // -- 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) local_group_module::impl::impl(group_module_ptr mod, std::string id)
: super(mod, std::move(id), mod->system().node()) { : impl(mod, std::move(id), mod->system().node()) {
CAF_LOG_DEBUG("created new local group:" << identifier_); CAF_LOG_DEBUG("created new local group:" << identifier_);
} }
local_group_module::impl::~impl() { local_group_module::impl::~impl() {
CAF_LOG_DEBUG("destroyed local group:" << identifier_); // nop
} }
void local_group_module::impl::enqueue(strong_actor_ptr sender, message_id mid, void local_group_module::impl::enqueue(strong_actor_ptr sender, message_id mid,
...@@ -152,7 +158,7 @@ expected<group> local_group_module::get(const std::string& group_name) { ...@@ -152,7 +158,7 @@ expected<group> local_group_module::get(const std::string& group_name) {
return make_error(sec::runtime_error, return make_error(sec::runtime_error,
"cannot get a group from on a stopped module"); "cannot get a group from on a stopped module");
} else if (auto i = instances_.find(group_name); i != instances_.end()) { } else if (auto i = instances_.find(group_name); i != instances_.end()) {
return group{i->second.get()}; return group{i->second};
} else { } else {
auto ptr = make_counted<impl>(this, group_name); auto ptr = make_counted<impl>(this, group_name);
ptr->intermediary_ = system().spawn<intermediary_actor, hidden>(ptr); ptr->intermediary_ = system().spawn<intermediary_actor, hidden>(ptr);
......
...@@ -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;
} }
...@@ -61,20 +53,28 @@ intptr_t group::compare(const group& other) const noexcept { ...@@ -61,20 +53,28 @@ intptr_t group::compare(const group& other) const noexcept {
expected<group> group::load_impl(actor_system& sys, const node_id& origin, expected<group> group::load_impl(actor_system& sys, const node_id& origin,
const std::string& mod, const std::string& mod,
const std::string& id) { const std::string& id) {
if (!origin || origin == sys.node()) 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); return sys.groups().get(mod, id);
else if (auto& get_remote = sys.groups().get_remote) }
} else if (auto& get_remote = sys.groups().get_remote) {
return get_remote(origin, mod, id); return get_remote(origin, mod, id);
else } else {
return make_error(sec::feature_disabled, return make_error(sec::feature_disabled,
"cannot access remote group: middleman not loaded"); "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 "<invalid-group>"; return x.get()->stringify();
else else
return x.get()->to_string(); return "<invalid-group>";
} }
} // namespace caf } // namespace caf
...@@ -34,7 +34,28 @@ public: ...@@ -34,7 +34,28 @@ public:
// nop // nop
} }
void stop() { 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) for (auto& kvp : instances)
kvp.second->stop(); kvp.second->stop();
instances.clear(); instances.clear();
...@@ -52,11 +73,6 @@ public: ...@@ -52,11 +73,6 @@ public:
} }
} }
expected<group> get(const std::string& group_name) {
auto result = get_impl(group_name);
return group{result.get()};
}
std::map<std::string, detail::group_tunnel_ptr> instances; std::map<std::string, detail::group_tunnel_ptr> instances;
}; };
...@@ -92,6 +108,19 @@ struct fixture : test_coordinator_fixture<> { ...@@ -92,6 +108,19 @@ struct fixture : test_coordinator_fixture<> {
sys.groups().get_module("local")->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; intrusive_ptr<mock_module> uut;
group origin; group origin;
actor intermediary; actor intermediary;
...@@ -202,4 +231,43 @@ CAF_TEST(tunnels automatically unsubscribe from their origin) { ...@@ -202,4 +231,43 @@ CAF_TEST(tunnels automatically unsubscribe from their origin) {
} }
} }
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() CAF_TEST_FIXTURE_SCOPE_END()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment