Commit 8ceb5019 authored by Dominik Charousset's avatar Dominik Charousset

Implement new group tunnel

parent cf9a69e3
......@@ -86,6 +86,7 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS}
src/detail/get_process_id.cpp
src/detail/get_root_uuid.cpp
src/detail/glob_match.cpp
src/detail/group_tunnel.cpp
src/detail/invoke_result_visitor.cpp
src/detail/local_group_module.cpp
src/detail/message_builder_element.cpp
......@@ -247,6 +248,7 @@ caf_add_test_suites(caf-core-test
detail.bounds_checker
detail.config_consumer
detail.encode_base64
detail.group_tunnel
detail.ieee_754
detail.limited_vector
detail.local_group_module
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/abstract_group.hpp"
#include "caf/actor.hpp"
#include "caf/detail/local_group_module.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/fwd.hpp"
#include "caf/stateful_actor.hpp"
namespace caf::detail {
/// Represents a group that runs on a different CAF node.
class group_tunnel : public local_group_module::impl {
public:
using super = local_group_module::impl;
group_tunnel(group_module_ptr mod, std::string id,
actor upstream_intermediary);
~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;
// 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);
auto worker() const noexcept {
return worker_;
}
private:
actor worker_;
};
using group_tunnel_ptr = intrusive_ptr<group_tunnel>;
} // namespace caf::detail
......@@ -78,6 +78,18 @@ public:
void stop() override;
protected:
template <class F>
auto critical_section(F&& fun) {
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;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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->upstream_enqueue(nullptr, make_message_id(),
make_message(group_down_msg{group{gptr.get()}}),
self->context());
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{};
});
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)) {
intermediary_ = std::move(upstream_intermediary);
worker_ = system().spawn<group_worker_actor>(this, intermediary_);
}
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) {
std::unique_lock<std::mutex> guard{mtx_};
if (!stopped_) {
auto wrapped = make_message(sys_atom_v, forward_atom_v, std::move(content));
worker_->enqueue(std::move(sender), mid, std::move(wrapped), host);
}
}
void group_tunnel::stop() {
CAF_LOG_DEBUG("stop group tunnel:" << CAF_ARG2("module", module().name())
<< CAF_ARG2("identifier", 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, worker_);
return true;
} else {
return false;
}
});
if (stopped)
anon_send_exit(hdl, exit_reason::user_shutdown);
}
void group_tunnel::upstream_enqueue(strong_actor_ptr sender, message_id mid,
message content, execution_unit* host) {
super::enqueue(std::move(sender), mid, std::move(content), host);
}
} // namespace caf::detail
......@@ -83,15 +83,12 @@ void local_group_module::impl::enqueue(strong_actor_ptr sender, message_id mid,
bool local_group_module::impl::subscribe(strong_actor_ptr who) {
std::unique_lock<std::mutex> guard{mtx_};
return !stopped_ && subscribers_.emplace(who).second;
return subscribe_impl(who).first;
}
void local_group_module::impl::unsubscribe(const actor_control_block* who) {
std::unique_lock<std::mutex> guard{mtx_};
// Note: can't call erase with `who` directly, because only set::find has an
// overload for any K that is less-comparable to Key.
if (auto i = subscribers_.find(who); i != subscribers_.end())
subscribers_.erase(i);
unsubscribe_impl(who);
}
actor local_group_module::impl::intermediary() const noexcept {
......@@ -103,18 +100,41 @@ 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;
std::unique_lock<std::mutex> guard{mtx_};
if (stopped_)
return;
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)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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() {
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;
}
}
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;
};
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();
}
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_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