Commit 59992443 authored by Dominik Charousset's avatar Dominik Charousset

Replace forwarding_actor_proxy with basp::proxy

parent 940e37a3
...@@ -53,7 +53,6 @@ set(LIBCAF_CORE_SRCS ...@@ -53,7 +53,6 @@ set(LIBCAF_CORE_SRCS
src/event_based_actor.cpp src/event_based_actor.cpp
src/execution_unit.cpp src/execution_unit.cpp
src/exit_reason.cpp src/exit_reason.cpp
src/forwarding_actor_proxy.cpp
src/get_mac_addresses.cpp src/get_mac_addresses.cpp
src/get_process_id.cpp src/get_process_id.cpp
src/get_root_uuid.cpp src/get_root_uuid.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/actor.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/detail/shared_spinlock.hpp"
namespace caf {
/// Implements a simple proxy forwarding all operations to a manager.
class forwarding_actor_proxy : public actor_proxy {
public:
using forwarding_stack = std::vector<strong_actor_ptr>;
forwarding_actor_proxy(actor_config& cfg, actor dest);
~forwarding_actor_proxy() override;
void enqueue(mailbox_element_ptr what, execution_unit* context) override;
bool add_backlink(abstract_actor* x) override;
bool remove_backlink(abstract_actor* x) override;
void kill_proxy(execution_unit* ctx, error rsn) override;
private:
void forward_msg(strong_actor_ptr sender, message_id mid, message msg,
const forwarding_stack* fwd = nullptr);
mutable detail::shared_spinlock broker_mtx_;
actor broker_;
};
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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 <utility>
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/send.hpp"
#include "caf/locks.hpp"
#include "caf/logger.hpp"
#include "caf/mailbox_element.hpp"
namespace caf {
forwarding_actor_proxy::forwarding_actor_proxy(actor_config& cfg, actor dest)
: actor_proxy(cfg),
broker_(std::move(dest)) {
anon_send(broker_, monitor_atom::value, ctrl());
}
forwarding_actor_proxy::~forwarding_actor_proxy() {
anon_send(broker_, make_message(delete_atom::value, node(), id()));
}
void forwarding_actor_proxy::forward_msg(strong_actor_ptr sender,
message_id mid, message msg,
const forwarding_stack* fwd) {
CAF_LOG_TRACE(CAF_ARG(id()) << CAF_ARG(sender)
<< CAF_ARG(mid) << CAF_ARG(msg));
if (msg.match_elements<exit_msg>())
unlink_from(msg.get_as<exit_msg>(0).source);
forwarding_stack tmp;
shared_lock<detail::shared_spinlock> guard(broker_mtx_);
if (broker_)
broker_->enqueue(nullptr, make_message_id(),
make_message(forward_atom::value, std::move(sender),
fwd != nullptr ? *fwd : tmp,
strong_actor_ptr{ctrl()}, mid,
std::move(msg)),
nullptr);
}
void forwarding_actor_proxy::enqueue(mailbox_element_ptr what,
execution_unit*) {
CAF_PUSH_AID(0);
CAF_ASSERT(what);
forward_msg(std::move(what->sender), what->mid,
what->move_content_to_message(), &what->stages);
}
bool forwarding_actor_proxy::add_backlink(abstract_actor* x) {
if (monitorable_actor::add_backlink(x)) {
forward_msg(ctrl(), make_message_id(),
make_message(link_atom::value, x->ctrl()));
return true;
}
return false;
}
bool forwarding_actor_proxy::remove_backlink(abstract_actor* x) {
if (monitorable_actor::remove_backlink(x)) {
forward_msg(ctrl(), make_message_id(),
make_message(unlink_atom::value, x->ctrl()));
return true;
}
return false;
}
void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, error rsn) {
actor tmp;
{ // lifetime scope of guard
std::unique_lock<detail::shared_spinlock> guard(broker_mtx_);
broker_.swap(tmp); // manually break cycle
}
cleanup(std::move(rsn), ctx);
}
} // namespace caf
...@@ -31,7 +31,6 @@ ...@@ -31,7 +31,6 @@
#include "caf/proxy_registry.hpp" #include "caf/proxy_registry.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/io/basp/all.hpp" #include "caf/io/basp/all.hpp"
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
......
...@@ -27,8 +27,8 @@ ...@@ -27,8 +27,8 @@
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/io/basp/all.hpp" #include "caf/io/basp/all.hpp"
#include "caf/io/basp/proxy.hpp"
#include "caf/io/connection_helper.hpp" #include "caf/io/connection_helper.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/network/interfaces.hpp" #include "caf/io/network/interfaces.hpp"
...@@ -131,10 +131,63 @@ behavior basp_broker::make_behavior() { ...@@ -131,10 +131,63 @@ behavior basp_broker::make_behavior() {
ctx.cstate = next; ctx.cstate = next;
} }
}, },
// received from proxy instances // received from BASP proxy instances
[=](const strong_actor_ptr& dest, basp::header& hdr,
const std::vector<char>& payload) {
CAF_LOG_TRACE(CAF_ARG(dest) << CAF_ARG2("payload.size", payload.size()));
// Sanity checking.
if (dest == nullptr) {
CAF_LOG_WARNING("drop message due to invalid receiver");
return;
}
auto& dest_node = dest->node();
if (dest_node == system().node()) {
CAF_LOG_WARNING("drop message due to local receiver");
return;
}
// Get route to destination node.
auto route = instance.tbl().lookup(dest_node);
if (!route) {
CAF_LOG_DEBUG("drop message to unknown node:" << dest_node);
return;
}
// Get the output buffer.
auto& out_buf = get_buffer(route->hdl);
auto before = out_buf.size();
// Check whether we need to adjust the header. The BASP proxy only
// considers the sender when deciding on whether a message is direct or
// routed. However, we also need to look at the receiver and in case the
// next hop is not the receiver than we need to set the operation field
// accordingly and prefix the payload with the node IDs.
if (route->next_hop != dest->node()
&& hdr.operation != basp::message_type::routed_message) {
hdr.operation = basp::message_type::routed_message;
hdr.payload_len += 2 * node_id::serialized_size;
binary_serializer sink(system(), out_buf);
if (auto err = sink(hdr, system().node(), dest_node)) {
CAF_LOG_ERROR("serialization error: " << system().render(err));
out_buf.resize(before);
return;
}
CAF_ASSERT(out_buf.size()
== before + basp::header_size
+ 2 * node_id::serialized_size);
} else {
binary_serializer sink(system(), out_buf);
if (auto err = sink(hdr)) {
CAF_LOG_ERROR("serialization error: " << system().render(err));
out_buf.resize(before);
return;
}
CAF_ASSERT(out_buf.size() == before + basp::header_size);
}
out_buf.insert(out_buf.end(), payload.begin(), payload.end());
flush(route->hdl);
},
// received from forwarding_proxy instances
[=](forward_atom, strong_actor_ptr& src, [=](forward_atom, strong_actor_ptr& src,
const std::vector<strong_actor_ptr>& fwd_stack, const std::vector<strong_actor_ptr>& fwd_stack, strong_actor_ptr& dest,
strong_actor_ptr& dest, message_id mid, const message& msg) { message_id mid, const message& msg) {
CAF_LOG_TRACE(CAF_ARG(src) << CAF_ARG(dest) CAF_LOG_TRACE(CAF_ARG(src) << CAF_ARG(dest)
<< CAF_ARG(mid) << CAF_ARG(msg)); << CAF_ARG(mid) << CAF_ARG(msg));
if (!dest || system().node() == dest->node()) { if (!dest || system().node() == dest->node()) {
...@@ -266,8 +319,8 @@ behavior basp_broker::make_behavior() { ...@@ -266,8 +319,8 @@ behavior basp_broker::make_behavior() {
return unit; return unit;
return sec::cannot_close_invalid_port; return sec::cannot_close_invalid_port;
}, },
[=](get_atom, const node_id& x) [=](get_atom,
-> std::tuple<node_id, std::string, uint16_t> { const node_id& x) -> std::tuple<node_id, std::string, uint16_t> {
std::string addr; std::string addr;
uint16_t port = 0; uint16_t port = 0;
auto hdl = instance.tbl().lookup_direct(x); auto hdl = instance.tbl().lookup_direct(x);
...@@ -281,8 +334,7 @@ behavior basp_broker::make_behavior() { ...@@ -281,8 +334,7 @@ behavior basp_broker::make_behavior() {
instance.handle_heartbeat(context()); instance.handle_heartbeat(context());
delayed_send(this, std::chrono::milliseconds{interval}, delayed_send(this, std::chrono::milliseconds{interval},
tick_atom::value, interval); tick_atom::value, interval);
} }};
};
} }
proxy_registry* basp_broker::proxy_registry_ptr() { proxy_registry* basp_broker::proxy_registry_ptr() {
...@@ -314,8 +366,8 @@ strong_actor_ptr basp_broker::make_proxy(node_id nid, actor_id aid) { ...@@ -314,8 +366,8 @@ strong_actor_ptr basp_broker::make_proxy(node_id nid, actor_id aid) {
// create proxy and add functor that will be called if we // create proxy and add functor that will be called if we
// receive a basp::down_message // receive a basp::down_message
actor_config cfg; actor_config cfg;
auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>( auto res = make_actor<basp::proxy, strong_actor_ptr>(aid, nid, &(system()),
aid, nid, &(system()), cfg, this); cfg, this);
strong_actor_ptr selfptr{ctrl()}; strong_actor_ptr selfptr{ctrl()};
res->get()->attach_functor([=](const error& rsn) { res->get()->attach_functor([=](const error& rsn) {
mm->backend().post([=] { mm->backend().post([=] {
......
...@@ -138,7 +138,19 @@ public: ...@@ -138,7 +138,19 @@ public:
binary_serializer sink{self_.home_system(), buf}; binary_serializer sink{self_.home_system(), buf};
// Differntiate between routed and direct messages based on the sender. // Differntiate between routed and direct messages based on the sender.
message_type msg_type; message_type msg_type;
if (x.sender == nullptr || x.sender->node() == self_.home_system().node()) { bool local_sender;
if (x.sender == nullptr) {
local_sender = true;
} else if (x.sender->node() == self_.home_system().node()) {
// Actors that communicate to remotes are kept alive by putting them into
// the registry, which also allows to get their handle for response
// messages.
self_.home_system().registry().put(x.sender->id(), x.sender);
local_sender = true;
} else {
local_sender = false;
}
if (local_sender) {
msg_type = message_type::direct_message; msg_type = message_type::direct_message;
if (auto err = sink(x.stages)) { if (auto err = sink(x.stages)) {
CAF_LOG_ERROR("cannot serialize stages:" << x); CAF_LOG_ERROR("cannot serialize stages:" << x);
......
...@@ -34,6 +34,7 @@ ...@@ -34,6 +34,7 @@
#include "caf/deep_to_string.hpp" #include "caf/deep_to_string.hpp"
#include "caf/io/basp/proxy.hpp"
#include "caf/io/network/interfaces.hpp" #include "caf/io/network/interfaces.hpp"
#include "caf/io/network/test_multiplexer.hpp" #include "caf/io/network/test_multiplexer.hpp"
...@@ -69,7 +70,6 @@ bool operator==(const maybe<T>& x, const T& y) { ...@@ -69,7 +70,6 @@ bool operator==(const maybe<T>& x, const T& y) {
} }
constexpr uint8_t no_flags = 0; constexpr uint8_t no_flags = 0;
constexpr uint32_t no_payload = 0;
constexpr uint64_t no_operation_data = 0; constexpr uint64_t no_operation_data = 0;
constexpr uint64_t default_operation_data = make_message_id().integer_value(); constexpr uint64_t default_operation_data = make_message_id().integer_value();
...@@ -108,15 +108,24 @@ struct node { ...@@ -108,15 +108,24 @@ struct node {
} }
}; };
struct config : actor_system_config {
config(bool autoconn) {
load<io::middleman, network::test_multiplexer>();
set("middleman.enable-automatic-connections", autoconn);
set("middleman.workers", size_t{0});
set("scheduler.policy", caf::atom("testing"));
set("middleman.attach-utility-actors", autoconn);
}
};
class fixture { class fixture {
public: public:
using scheduler_type = caf::scheduler::test_coordinator;
fixture(bool autoconn = false) fixture(bool autoconn = false)
: sys(cfg.load<io::middleman, network::test_multiplexer>() : cfg(autoconn),
.set("middleman.enable-automatic-connections", autoconn) sys(cfg),
.set("middleman.workers", size_t{0}) sched(dynamic_cast<scheduler_type&>(sys.scheduler())) {
.set("scheduler.policy",
autoconn ? caf::atom("testing") : caf::atom("stealing"))
.set("middleman.attach-utility-actors", autoconn)) {
auto& mm = sys.middleman(); auto& mm = sys.middleman();
mpx_ = dynamic_cast<network::test_multiplexer*>(&mm.backend()); mpx_ = dynamic_cast<network::test_multiplexer*>(&mm.backend());
CAF_REQUIRE(mpx_ != nullptr); CAF_REQUIRE(mpx_ != nullptr);
...@@ -362,8 +371,8 @@ public: ...@@ -362,8 +371,8 @@ public:
basp::header hdr; basp::header hdr;
{ // lifetime scope of source { // lifetime scope of source
binary_deserializer source{this_->mpx(), ob}; binary_deserializer source{this_->mpx(), ob};
auto e = source(hdr); if (auto err = source(hdr))
CAF_REQUIRE_EQUAL(e, none); CAF_FAIL("serializing the header failed: " << this_->sys.render(err));
} }
buffer payload; buffer payload;
if (hdr.payload_len > 0) { if (hdr.payload_len > 0) {
...@@ -409,8 +418,15 @@ public: ...@@ -409,8 +418,15 @@ public:
return {this}; return {this};
} }
actor_system_config cfg; void run() {
while (mpx_->try_exec_runnable() || sched.try_run_once()) {
// repeat
}
}
config cfg;
actor_system sys; actor_system sys;
scheduler_type& sched;
private: private:
basp_broker* aut_; basp_broker* aut_;
...@@ -419,26 +435,15 @@ private: ...@@ -419,26 +435,15 @@ private:
node_id this_node_; node_id this_node_;
unique_ptr<scoped_actor> self_; unique_ptr<scoped_actor> self_;
array<node, num_remote_nodes> nodes_; array<node, num_remote_nodes> nodes_;
/*
array<node_id, num_remote_nodes> remote_node_;
array<connection_handle, num_remote_nodes> remote_hdl_;
array<unique_ptr<scoped_actor>, num_remote_nodes> pseudo_remote_;
*/
actor_registry* registry_; actor_registry* registry_;
}; };
class autoconn_enabled_fixture : public fixture { class autoconn_enabled_fixture : public fixture {
public: public:
using scheduler_type = caf::scheduler::test_coordinator;
scheduler_type& sched;
middleman_actor mma; middleman_actor mma;
autoconn_enabled_fixture() autoconn_enabled_fixture()
: fixture(true), : fixture(true), mma(sys.middleman().actor_handle()) {
sched(dynamic_cast<scheduler_type&>(sys.scheduler())),
mma(sys.middleman().actor_handle()) {
// nop // nop
} }
...@@ -540,7 +545,8 @@ CAF_TEST(client_handshake_and_dispatch) { ...@@ -540,7 +545,8 @@ CAF_TEST(client_handshake_and_dispatch) {
return a + b + c; return a + b + c;
} }
); );
CAF_MESSAGE("exec message of forwarding proxy"); CAF_MESSAGE("exec message of BASP proxy");
sched.run();
mpx()->exec_runnable(); mpx()->exec_runnable();
// deserialize and send message from out buf // deserialize and send message from out buf
dispatch_out_buf(jupiter().connection); dispatch_out_buf(jupiter().connection);
...@@ -614,7 +620,7 @@ CAF_TEST(remote_actor_and_send) { ...@@ -614,7 +620,7 @@ CAF_TEST(remote_actor_and_send) {
[&](node_id nid, strong_actor_ptr res, std::set<std::string> ifs) { [&](node_id nid, strong_actor_ptr res, std::set<std::string> ifs) {
CAF_REQUIRE(res); CAF_REQUIRE(res);
auto aptr = actor_cast<abstract_actor*>(res); auto aptr = actor_cast<abstract_actor*>(res);
CAF_REQUIRE(dynamic_cast<forwarding_actor_proxy*>(aptr) != nullptr); CAF_REQUIRE(dynamic_cast<io::basp::proxy*>(aptr) != nullptr);
CAF_CHECK_EQUAL(proxies().count_proxies(jupiter().id), 1u); CAF_CHECK_EQUAL(proxies().count_proxies(jupiter().id), 1u);
CAF_CHECK_EQUAL(nid, jupiter().id); CAF_CHECK_EQUAL(nid, jupiter().id);
CAF_CHECK_EQUAL(res->node(), jupiter().id); CAF_CHECK_EQUAL(res->node(), jupiter().id);
...@@ -632,6 +638,7 @@ CAF_TEST(remote_actor_and_send) { ...@@ -632,6 +638,7 @@ CAF_TEST(remote_actor_and_send) {
CAF_MESSAGE("send message to proxy"); CAF_MESSAGE("send message to proxy");
anon_send(actor_cast<actor>(result), 42); anon_send(actor_cast<actor>(result), 42);
mpx()->flush_runnables(); mpx()->flush_runnables();
sched.run();
// mpx()->exec_runnable(); // process forwarded message in basp_broker // mpx()->exec_runnable(); // process forwarded message in basp_broker
mock().receive(jupiter().connection, basp::message_type::direct_message, mock().receive(jupiter().connection, basp::message_type::direct_message,
no_flags, any_vals, default_operation_data, invalid_actor_id, no_flags, any_vals, default_operation_data, invalid_actor_id,
...@@ -677,8 +684,8 @@ CAF_TEST(actor_serialize_and_deserialize) { ...@@ -677,8 +684,8 @@ CAF_TEST(actor_serialize_and_deserialize) {
std::vector<strong_actor_ptr>{}, msg); std::vector<strong_actor_ptr>{}, msg);
// testee must've responded (process forwarded message in BASP broker) // testee must've responded (process forwarded message in BASP broker)
CAF_MESSAGE("wait until BASP broker writes to its output buffer"); CAF_MESSAGE("wait until BASP broker writes to its output buffer");
while (mpx()->output_buffer(jupiter().connection).empty()) if (mpx()->output_buffer(jupiter().connection).empty())
mpx()->exec_runnable(); // process forwarded message in basp_broker run();
// output buffer must contain the reflected message // output buffer must contain the reflected message
mock().receive(jupiter().connection, basp::message_type::direct_message, mock().receive(jupiter().connection, basp::message_type::direct_message,
no_flags, any_vals, default_operation_data, testee->id(), no_flags, any_vals, default_operation_data, testee->id(),
...@@ -721,7 +728,8 @@ CAF_TEST(indirect_connections) { ...@@ -721,7 +728,8 @@ CAF_TEST(indirect_connections) {
return "hello from earth!"; return "hello from earth!";
} }
); );
mpx()->exec_runnable(); // process forwarded message in basp_broker run();
CAF_MESSAGE("send response to jupiter (via mars)");
mock().receive(mars().connection, basp::message_type::routed_message, mock().receive(mars().connection, basp::message_type::routed_message,
no_flags, any_vals, default_operation_data, self()->id(), no_flags, any_vals, default_operation_data, self()->id(),
jupiter().dummy_actor->id(), this_node(), jupiter().id, jupiter().dummy_actor->id(), this_node(), jupiter().id,
...@@ -817,7 +825,7 @@ CAF_TEST(automatic_connection) { ...@@ -817,7 +825,7 @@ CAF_TEST(automatic_connection) {
return "hello from earth!"; return "hello from earth!";
} }
); );
mpx()->exec_runnable(); // process forwarded message in basp_broker run();
CAF_MESSAGE("response message must take direct route now"); CAF_MESSAGE("response message must take direct route now");
mock().receive(jupiter().connection, basp::message_type::direct_message, mock().receive(jupiter().connection, basp::message_type::direct_message,
no_flags, any_vals, make_message_id().integer_value(), no_flags, any_vals, make_message_id().integer_value(),
......
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