Commit fc579fa3 authored by neverlord's avatar neverlord

fixed link behavior for remote actors

parent 2e4f96ce
......@@ -63,8 +63,16 @@ class actor_proxy : public abstract_actor<actor>
void link_to(intrusive_ptr<actor>& other);
// do not cause to send this actor an ":Unlink" message
// to the "original" remote actor
void local_link_to(intrusive_ptr<actor>& other);
void unlink_from(intrusive_ptr<actor>& other);
// do not cause to send this actor an ":Unlink" message
// to the "original" remote actor
void local_unlink_from(intrusive_ptr<actor>& other);
bool remove_backlink(intrusive_ptr<actor>& to);
bool establish_backlink(intrusive_ptr<actor>& to);
......
......@@ -68,41 +68,6 @@ void actor_proxy::enqueue(actor* sender, any_tuple&& msg)
void actor_proxy::enqueue(actor* sender, const any_tuple& msg)
{
/*
if (msg.size() > 0 && msg.utype_info_at(0) == typeid(atom_value))
{
if (msg.size() == 2 && msg.utype_info_at(1) == typeid(actor_ptr))
{
switch (to_int(msg.get_as<atom_value>(0)))
{
// received via post_office
case to_int(atom(":Link")):
{
auto s = msg.get_as<actor_ptr>(1);
(void) link_to_impl(s);
//link_to(s);
return;
}
// received via post_office
case to_int(atom(":Unlink")):
{
auto s = msg.get_as<actor_ptr>(1);
(void) unlink_from_impl(s);
//unlink_from(s);
return;
}
default: break;
}
}
else if ( msg.size() == 2
&& msg.get_as<atom_value>(0) == atom(":KillProxy")
&& msg.utype_info_at(1) == typeid(std::uint32_t))
{
cleanup(msg.get_as<std::uint32_t>(1));
return;
}
}
*/
if ( msg.size() == 2
&& *(msg.type_at(0)) == typeid(atom_value)
&& msg.get_as<atom_value>(0) == atom(":KillProxy")
......@@ -121,10 +86,15 @@ void actor_proxy::link_to(intrusive_ptr<actor>& other)
// causes remote actor to link to (proxy of) other
forward_message(parent_process_ptr(),
other.get(),
make_tuple(atom(":Link"), actor_ptr(this)));
make_tuple(atom(":Link"), other));
}
}
void actor_proxy::local_link_to(intrusive_ptr<actor>& other)
{
link_to_impl(other);
}
void actor_proxy::unlink_from(intrusive_ptr<actor>& other)
{
if (unlink_from_impl(other))
......@@ -132,18 +102,24 @@ void actor_proxy::unlink_from(intrusive_ptr<actor>& other)
// causes remote actor to unlink from (proxy of) other
forward_message(parent_process_ptr(),
other.get(),
make_tuple(atom(":Unlink"), actor_ptr(this)));
make_tuple(atom(":Unlink"), other));
}
}
void actor_proxy::local_unlink_from(intrusive_ptr<actor>& other)
{
unlink_from_impl(other);
}
bool actor_proxy::establish_backlink(intrusive_ptr<actor>& other)
{
bool result = super::establish_backlink(other);
if (result)
{
// causes remote actor to unlink from (proxy of) other
forward_message(parent_process_ptr(),
other.get(),
make_tuple(atom(":Link"), actor_ptr(this)));
make_tuple(atom(":Link"), other));
}
return result;
}
......
......@@ -59,6 +59,7 @@
#include "cppa/detail/thread.hpp"
#include "cppa/detail/buffer.hpp"
#include "cppa/detail/mailman.hpp"
#include "cppa/detail/types_array.hpp"
#include "cppa/detail/post_office.hpp"
#include "cppa/detail/native_socket.hpp"
#include "cppa/detail/actor_registry.hpp"
......@@ -82,6 +83,8 @@ using std::endl;
namespace {
cppa::detail::types_array<cppa::atom_value, cppa::actor_ptr> t_atom_actor_ptr_types;
// allocate in 1KB chunks (minimize reallocations)
constexpr size_t s_chunk_size = 1024;
......@@ -186,11 +189,8 @@ class po_peer
void add_child(const actor_proxy_ptr& pptr)
{
CPPA_REQUIRE(pptr.get() != nullptr);
if (pptr) m_children.push_back(pptr);
else
{
DEBUG("po_peer::add_child(nullptr) called");
}
}
inline size_t children_count() const
......@@ -322,12 +322,13 @@ class po_peer
}
auto& content = msg.content();
DEBUG("<-- " << to_string(content));
// intercept ":Monitor" messages
if ( content.size() == 1
&& *(content.type_at(0)) == typeid(atom_value)
&& t_atom_actor_ptr_types[0] == content.type_at(0)
&& content.get_as<atom_value>(0) == atom(":Monitor"))
{
auto receiver = msg.receiver().downcast<actor>();
//actor_ptr receiver = dynamic_cast<actor*>(receiver_ch.get());
CPPA_REQUIRE(receiver.get() != nullptr);
if (!receiver)
{
DEBUG("empty receiver");
......@@ -354,6 +355,30 @@ class po_peer
DEBUG(":Monitor received for an remote actor");
}
}
// intercept ":Link" messages
else if ( content.size() == 2
&& t_atom_actor_ptr_types[0] == content.type_at(0)
&& t_atom_actor_ptr_types[1] == content.type_at(1)
&& content.get_as<atom_value>(0) == atom(":Link"))
{
CPPA_REQUIRE(msg.sender()->is_proxy());
auto whom = msg.sender().downcast<actor_proxy>();
auto to = content.get_as<actor_ptr>(1);
if ((whom) && (to))
whom->local_link_to(to);
}
// intercept ":Unlink" messages
else if ( content.size() == 2
&& t_atom_actor_ptr_types[0] == content.type_at(0)
&& t_atom_actor_ptr_types[1] == content.type_at(1)
&& content.get_as<atom_value>(0) == atom(":Link"))
{
CPPA_REQUIRE(msg.sender()->is_proxy());
auto whom = msg.sender().downcast<actor_proxy>();
auto from = content.get_as<actor_ptr>(1);
if ((whom) && (from))
whom->local_unlink_from(from);
}
else
{
if (msg.receiver())
......
#define CPPA_VERBOSE_CHECK
#include <list>
#include <string>
#include <utility>
......@@ -42,10 +40,8 @@ size_t test__tuple()
CPPA_CHECK_EQUAL(t0_0, "1");
CPPA_CHECK_EQUAL(t0_1, 2);
// get a view of t0
/*
any_tuple atup0(t0);
any_tuple_view aview0(atup0);
auto v1opt = tuple_cast<std::string, anything>(aview0);
auto v1opt = tuple_cast<std::string, anything>(any_tuple_view(atup0));
CPPA_CHECK((v1opt));
if (v1opt)
{
......@@ -54,7 +50,6 @@ size_t test__tuple()
CPPA_CHECK_EQUAL(v1.size(), 1);
CPPA_CHECK_EQUAL(v1_0, "1");
}
// */
// use tuple cast to get a subtuple
any_tuple at0(t0);
auto v0opt = tuple_cast<std::string, anything>(at0);
......
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