Commit bbfa7659 authored by Dominik Charousset's avatar Dominik Charousset

{io::remote => forwarding}_actor_proxy

parent 4283b08c
...@@ -44,6 +44,7 @@ set (LIBCAF_CORE_SRCS ...@@ -44,6 +44,7 @@ set (LIBCAF_CORE_SRCS
src/exception.cpp src/exception.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_root_uuid.cpp src/get_root_uuid.cpp
src/group.cpp src/group.cpp
......
...@@ -38,8 +38,8 @@ class actor_proxy; ...@@ -38,8 +38,8 @@ class actor_proxy;
using actor_proxy_ptr = intrusive_ptr<actor_proxy>; using actor_proxy_ptr = intrusive_ptr<actor_proxy>;
/** /**
* Represents a remote actor. * Represents an actor running on a remote machine,
* @extends abstract_actor * or different hardware, or in a separate process.
*/ */
class actor_proxy : public abstract_actor { class actor_proxy : public abstract_actor {
public: public:
......
...@@ -17,27 +17,24 @@ ...@@ -17,27 +17,24 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_IO_REMOTE_ACTOR_PROXY_HPP #ifndef CAF_FORWARDING_ACTOR_PROXY_HPP
#define CAF_IO_REMOTE_ACTOR_PROXY_HPP #define CAF_FORWARDING_ACTOR_PROXY_HPP
#include "caf/extend.hpp" #include "caf/actor.hpp"
#include "caf/actor_proxy.hpp" #include "caf/actor_proxy.hpp"
#include "caf/mixin/memory_cached.hpp" #include "caf/detail/shared_spinlock.hpp"
#include "caf/detail/single_reader_queue.hpp"
namespace caf { namespace caf {
namespace io {
/** /**
* Implements a simple proxy forwarding all operations to a manager. * Implements a simple proxy forwarding all operations to a manager.
*/ */
class remote_actor_proxy : public actor_proxy { class forwarding_actor_proxy : public actor_proxy {
public: public:
remote_actor_proxy(actor_id mid, node_id pinfo, actor parent); forwarding_actor_proxy(actor_id mid, node_id pinfo, actor parent);
~remote_actor_proxy(); ~forwarding_actor_proxy();
void enqueue(const actor_addr&, message_id, void enqueue(const actor_addr&, message_id,
message, execution_unit*) override; message, execution_unit*) override;
...@@ -50,12 +47,17 @@ class remote_actor_proxy : public actor_proxy { ...@@ -50,12 +47,17 @@ class remote_actor_proxy : public actor_proxy {
void kill_proxy(uint32_t reason) override; void kill_proxy(uint32_t reason) override;
actor manager() const;
void manager(actor new_manager);
private: private:
void forward_msg(const actor_addr& sender, message_id mid, message msg); void forward_msg(const actor_addr& sender, message_id mid, message msg);
mutable detail::shared_spinlock m_manager_mtx;
actor m_manager; actor m_manager;
}; };
} // namespace io
} // namespace caf } // namespace caf
#endif // CAF_IO_REMOTE_ACTOR_PROXY_HPP #endif // CAF_FORWARDING_ACTOR_PROXY_HPP
...@@ -47,6 +47,7 @@ class blocking_actor; ...@@ -47,6 +47,7 @@ class blocking_actor;
class message_handler; class message_handler;
class uniform_type_info; class uniform_type_info;
class event_based_actor; class event_based_actor;
class forwarding_actor_proxy;
// structs // structs
struct anything; struct anything;
......
...@@ -17,48 +17,62 @@ ...@@ -17,48 +17,62 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/locks.hpp"
#include "caf/to_string.hpp" #include "caf/to_string.hpp"
#include "caf/io/remote_actor_proxy.hpp"
#include "caf/detail/memory.hpp"
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
using namespace std; using namespace std;
namespace caf { namespace caf {
namespace io {
remote_actor_proxy::remote_actor_proxy(actor_id aid, node_id nid, actor parent) forwarding_actor_proxy::forwarding_actor_proxy(actor_id aid, node_id nid,
actor manager)
: actor_proxy(aid, nid), : actor_proxy(aid, nid),
m_manager(parent) { m_manager(manager) {
CAF_REQUIRE(parent != invalid_actor); CAF_REQUIRE(manager != invalid_actor);
CAF_LOG_INFO(CAF_ARG(aid) << ", " << CAF_TARG(nid, to_string)); CAF_LOG_INFO(CAF_ARG(aid) << ", " << CAF_TARG(nid, to_string));
} }
remote_actor_proxy::~remote_actor_proxy() { forwarding_actor_proxy::~forwarding_actor_proxy() {
anon_send(m_manager, make_message(atom("_DelProxy"), node(), id())); anon_send(m_manager, make_message(atom("_DelProxy"), node(), id()));
} }
void remote_actor_proxy::forward_msg(const actor_addr& sender, message_id mid, actor forwarding_actor_proxy::manager() const {
message msg) { actor result;
{
shared_lock<detail::shared_spinlock> m_guard(m_manager_mtx);
result = m_manager;
}
return result;
}
void forwarding_actor_proxy::manager(actor new_manager) {
std::unique_lock<detail::shared_spinlock> m_guard(m_manager_mtx);
m_manager.swap(new_manager);
}
void forwarding_actor_proxy::forward_msg(const actor_addr& sender,
message_id mid, message msg) {
CAF_LOG_TRACE(CAF_ARG(id()) << ", " << CAF_TSARG(sender) << ", " CAF_LOG_TRACE(CAF_ARG(id()) << ", " << CAF_TSARG(sender) << ", "
<< CAF_MARG(mid, integer_value) << ", " << CAF_TSARG(msg)); << CAF_MARG(mid, integer_value) << ", "
<< CAF_TSARG(msg));
shared_lock<detail::shared_spinlock> m_guard(m_manager_mtx);
m_manager->enqueue(invalid_actor_addr, invalid_message_id, m_manager->enqueue(invalid_actor_addr, invalid_message_id,
make_message(atom("_Dispatch"), sender, make_message(atom("_Dispatch"), sender,
address(), mid, std::move(msg)), address(), mid, std::move(msg)),
nullptr); nullptr);
} }
void remote_actor_proxy::enqueue(const actor_addr& sender, message_id mid, void forwarding_actor_proxy::enqueue(const actor_addr& sender, message_id mid,
message m, execution_unit*) { message m, execution_unit*) {
forward_msg(sender, mid, std::move(m)); forward_msg(sender, mid, std::move(m));
} }
bool remote_actor_proxy::link_impl(linking_operation op, bool forwarding_actor_proxy::link_impl(linking_operation op,
const actor_addr& other) { const actor_addr& other) {
switch (op) { switch (op) {
case establish_link_op: case establish_link_op:
...@@ -98,17 +112,16 @@ bool remote_actor_proxy::link_impl(linking_operation op, ...@@ -98,17 +112,16 @@ bool remote_actor_proxy::link_impl(linking_operation op,
return false; return false;
} }
void remote_actor_proxy::local_link_to(const actor_addr& other) { void forwarding_actor_proxy::local_link_to(const actor_addr& other) {
establish_link_impl(other); establish_link_impl(other);
} }
void remote_actor_proxy::local_unlink_from(const actor_addr& other) { void forwarding_actor_proxy::local_unlink_from(const actor_addr& other) {
remove_link_impl(other); remove_link_impl(other);
} }
void remote_actor_proxy::kill_proxy(uint32_t reason) { void forwarding_actor_proxy::kill_proxy(uint32_t reason) {
cleanup(reason); cleanup(reason);
} }
} // namespace io
} // namespace caf } // namespace caf
...@@ -16,7 +16,6 @@ set (LIBCAF_IO_SRCS ...@@ -16,7 +16,6 @@ set (LIBCAF_IO_SRCS
src/publish.cpp src/publish.cpp
src/publish_local_groups.cpp src/publish_local_groups.cpp
src/remote_actor.cpp src/remote_actor.cpp
src/remote_actor_proxy.cpp
src/remote_group.cpp src/remote_group.cpp
src/manager.cpp src/manager.cpp
src/stream_manager.cpp src/stream_manager.cpp
......
...@@ -29,10 +29,10 @@ ...@@ -29,10 +29,10 @@
#include "caf/actor_namespace.hpp" #include "caf/actor_namespace.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.hpp" #include "caf/io/basp.hpp"
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/io/remote_actor_proxy.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
......
...@@ -27,7 +27,6 @@ class basp_broker; ...@@ -27,7 +27,6 @@ class basp_broker;
class broker; class broker;
class middleman; class middleman;
class receive_policy; class receive_policy;
class remote_actor_proxy;
namespace network { namespace network {
......
...@@ -17,9 +17,12 @@ ...@@ -17,9 +17,12 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/io/basp_broker.hpp"
#include "caf/exception.hpp" #include "caf/exception.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/detail/singletons.hpp" #include "caf/detail/singletons.hpp"
#include "caf/detail/make_counted.hpp" #include "caf/detail/make_counted.hpp"
...@@ -28,8 +31,6 @@ ...@@ -28,8 +31,6 @@
#include "caf/io/basp.hpp" #include "caf/io/basp.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/unpublish.hpp" #include "caf/io/unpublish.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/remote_actor_proxy.hpp"
using std::string; using std::string;
...@@ -570,7 +571,7 @@ actor_proxy_ptr basp_broker::make_proxy(const id_type& nid, actor_id aid) { ...@@ -570,7 +571,7 @@ actor_proxy_ptr basp_broker::make_proxy(const id_type& nid, actor_id aid) {
// receive a kill_proxy_instance message // receive a kill_proxy_instance message
intrusive_ptr<basp_broker> self = this; intrusive_ptr<basp_broker> self = this;
auto mm = middleman::instance(); auto mm = middleman::instance();
auto res = make_counted<remote_actor_proxy>(aid, nid, self); auto res = make_counted<forwarding_actor_proxy>(aid, nid, self);
res->attach_functor([=](uint32_t) { res->attach_functor([=](uint32_t) {
mm->backend().dispatch([=] { mm->backend().dispatch([=] {
// using res->id() instead of aid keeps this actor instance alive // using res->id() instead of aid keeps this actor instance alive
......
...@@ -35,7 +35,6 @@ ...@@ -35,7 +35,6 @@
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/system_messages.hpp" #include "caf/io/system_messages.hpp"
#include "caf/io/remote_actor_proxy.hpp"
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
#include "caf/detail/ripemd_160.hpp" #include "caf/detail/ripemd_160.hpp"
......
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