Commit e4ff83c9 authored by Dominik Charousset's avatar Dominik Charousset

Re-implement remote groups using group tunnels

parent 713eb9b8
......@@ -29,6 +29,7 @@ endfunction()
add_library(libcaf_io_obj OBJECT ${CAF_IO_HEADERS}
src/detail/prometheus_broker.cpp
src/detail/remote_group_module.cpp
src/detail/socket_guard.cpp
src/io/abstract_broker.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();
}
// Removes an instance when connecting it failed.
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 @@
#include "caf/actor_system.hpp"
#include "caf/config_value.hpp"
#include "caf/detail/io_export.hpp"
#include "caf/detail/remote_group_module.hpp"
#include "caf/detail/unique_function.hpp"
#include "caf/expected.hpp"
#include "caf/fwd.hpp"
......@@ -111,12 +112,22 @@ public:
return actor_cast<ActorHandle>(std::move(*x));
}
/// <group-name>@<host>:<port>
expected<group> remote_group(const std::string& group_uri);
/// Tries to connect to a group that runs on a different node in the network.
/// @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,
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.
actor_system& system() {
return system_;
......@@ -371,6 +382,9 @@ private:
/// Stores hidden background actors that get killed automatically when the
/// actor systems shuts down.
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
......@@ -101,6 +101,8 @@ using middleman_actor = typed_actor<
replies_to<spawn_atom, node_id, std::string, message,
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>>;
/// 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 result = i->second;
result->connect(std::move(intermediary));
return result;
} else {
auto result = make_counted<detail::group_tunnel>(
this, group_name, std::move(intermediary));
instances.emplace(group_name, result);
return result;
}
}
});
}
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 @@
#include "caf/detail/set_thread_name.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/function_view.hpp"
#include "caf/group_module.hpp"
#include "caf/init_global_meta_objects.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/io/basp_broker.hpp"
......@@ -118,7 +119,7 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
}
middleman::middleman(actor_system& sys) : system_(sys) {
// nop
remote_groups_ = make_counted<detail::remote_group_module>(this);
}
expected<strong_actor_ptr>
......@@ -168,19 +169,20 @@ expected<uint16_t> middleman::publish_local_groups(uint16_t port,
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in));
auto group_nameserver = [](event_based_actor* self) -> behavior {
return {
[self](get_atom, const std::string& name) {
return self->system().groups().get_local(name);
[self](get_atom, const std::string& id) {
auto grp = self->system().groups().get_local(id);
return grp.get()->intermediary();
},
};
};
auto gn = system().spawn<hidden + lazy_init>(group_nameserver);
auto result = publish(gn, port, in, reuse);
// link gn to our manager
if (result)
manager_->add_link(actor_cast<abstract_actor*>(gn));
else
anon_send_exit(gn, exit_reason::user_shutdown);
return result;
auto ns = system().spawn<hidden + lazy_init>(group_nameserver);
if (auto result = publish(ns, port, in, reuse)) {
manager_->add_link(actor_cast<abstract_actor*>(ns));
return *result;
} else {
anon_send_exit(ns, exit_reason::user_shutdown);
return std::move(result.error());
}
}
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,
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
// then tries to get a valid group from that actor.
auto two_step_lookup = [=](event_based_actor* self,
middleman_actor mm) -> behavior {
auto two_step_lookup = [=](event_based_actor* self, middleman_actor mm,
detail::remote_group_module_ptr rgm) -> behavior {
return {
[=](get_atom) {
/// 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,
CAF_LOG_DEBUG("received handle to target node:" << CAF_ARG(ptr));
auto hdl = actor_cast<actor>(ptr);
self->request(hdl, infinite, get_atom_v, group_identifier)
.then([=](group& grp) mutable {
CAF_LOG_DEBUG("received remote group handle:" << CAF_ARG(grp));
rp.deliver(std::move(grp));
.then([=](actor intermediary) mutable {
CAF_LOG_DEBUG("lookup successful:" << CAF_ARG(group_identifier)
<< 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,
expected<group> result{sec::cannot_connect_to_node};
scoped_actor self{system(), true};
auto worker = self->spawn<lazy_init + monitored>(two_step_lookup,
actor_handle());
actor_handle(),
remote_groups_);
self->send(worker, get_atom_v);
self->receive(
[&](group& grp) {
......@@ -272,6 +278,20 @@ expected<group> middleman::remote_group(const std::string& group_identifier,
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,
const node_id& nid) {
CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(nid));
......@@ -334,6 +354,19 @@ void middleman::start() {
if (auto prom = get_if<dict>(&system().config(),
"caf.middleman.prometheus-http"))
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() {
......@@ -388,6 +421,13 @@ void middleman::init(actor_system_config& cfg) {
system().node_.swap(this_node);
// Give config access to slave mode implementation.
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,
......
......@@ -148,7 +148,7 @@ auto middleman_actor_impl::make_behavior() -> behavior_type {
delegate(broker_, atm, p);
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> {
CAF_LOG_TRACE("");
if (!nid)
......@@ -168,14 +168,25 @@ auto middleman_actor_impl::make_behavior() -> behavior_type {
// reference but spawn_server_id is constexpr).
auto id = basp::header::spawn_server_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)));
return delegated<strong_actor_ptr>{};
},
[=](get_atom atm,
node_id nid) -> delegated<node_id, std::string, uint16_t> {
[=](get_atom, group_atom, node_id& nid,
std::string& group_id) -> result<actor> {
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 {};
},
};
......
......@@ -16,10 +16,11 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE io.remote_group
#include "caf/config.hpp"
#define CAF_SUITE io_dynamic_remote_group
#include "caf/test/io_dsl.hpp"
#include "io-test.hpp"
#include <algorithm>
#include <vector>
......@@ -48,7 +49,9 @@ size_t received_messages = 0u;
behavior group_receiver(event_based_actor* self) {
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`.
......@@ -90,7 +93,7 @@ CAF_TEST(connecting to remote group) {
CAF_CHECK_EQUAL(grp->get()->identifier(), group_name);
}
CAF_TEST_DISABLED(message transmission) {
CAF_TEST(message transmission) {
CAF_MESSAGE("spawn 5 receivers on mars");
auto mars_grp = mars.sys.groups().get_local(group_name);
spawn_receivers(mars, mars_grp, 5u);
......@@ -102,19 +105,18 @@ CAF_TEST_DISABLED(message transmission) {
auto earth_grp = unbox(earth.mm.remote_group(group_name, server, port));
CAF_MESSAGE("spawn 5 more receivers on earth");
spawn_receivers(earth, earth_grp, 5u);
exec_all();
CAF_MESSAGE("send message on mars and expect 10 handled messages total");
{
received_messages = 0u;
scoped_actor self{mars.sys};
self->send(mars_grp, ok_atom_v);
mars.self->send(mars_grp, ok_atom_v);
exec_all();
CAF_CHECK_EQUAL(received_messages, 10u);
}
CAF_MESSAGE("send message on earth and again expect 10 handled messages");
{
received_messages = 0u;
scoped_actor self{earth.sys};
self->send(earth_grp, ok_atom_v);
earth.self->send(earth_grp, ok_atom_v);
exec_all();
CAF_CHECK_EQUAL(received_messages, 10u);
}
......
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