Commit 394c4f74 authored by Dominik Charousset's avatar Dominik Charousset

actors no longer hold process_information pointers

process_information is now an implementation detail of the default
binary protocol and is thus removed from the base class actor
parent 67b5f6ea
...@@ -39,7 +39,6 @@ ...@@ -39,7 +39,6 @@
#include "cppa/channel.hpp" #include "cppa/channel.hpp"
#include "cppa/attachable.hpp" #include "cppa/attachable.hpp"
#include "cppa/message_id.hpp" #include "cppa/message_id.hpp"
#include "cppa/process_information.hpp"
#include "cppa/util/rm_ref.hpp" #include "cppa/util/rm_ref.hpp"
...@@ -152,20 +151,6 @@ class actor : public channel { ...@@ -152,20 +151,6 @@ class actor : public channel {
*/ */
virtual bool remove_backlink(const intrusive_ptr<actor>& other) = 0; virtual bool remove_backlink(const intrusive_ptr<actor>& other) = 0;
/**
* @brief Gets the {@link process_information} of the parent process.
* @returns The {@link process_information} of the parent process.
*/
inline const process_information& parent_process() const;
/**
* @brief Gets the {@link process_information} pointer
* of the parent process.
* @returns A pointer to the {@link process_information}
* of the parent process.
*/
inline process_information_ptr parent_process_ptr() const;
/** /**
* @brief Gets an integer value that uniquely identifies this Actor in * @brief Gets an integer value that uniquely identifies this Actor in
* the process it's executed in. * the process it's executed in.
...@@ -183,16 +168,14 @@ class actor : public channel { ...@@ -183,16 +168,14 @@ class actor : public channel {
protected: protected:
actor(const process_information_ptr& parent = process_information::get()); actor();
actor(actor_id aid, actor(actor_id aid);
const process_information_ptr& parent = process_information::get());
private: private:
actor_id m_id; actor_id m_id;
bool m_is_proxy; bool m_is_proxy;
process_information_ptr m_parent_process;
}; };
...@@ -211,14 +194,6 @@ bool operator!=(const self_type& lhs, const actor_ptr& rhs); ...@@ -211,14 +194,6 @@ bool operator!=(const self_type& lhs, const actor_ptr& rhs);
* inline and template member function implementations * * inline and template member function implementations *
******************************************************************************/ ******************************************************************************/
inline const process_information& actor::parent_process() const {
return *m_parent_process;
}
inline process_information_ptr actor::parent_process_ptr() const {
return m_parent_process;
}
inline std::uint32_t actor::id() const { inline std::uint32_t actor::id() const {
return m_id; return m_id;
} }
......
...@@ -61,8 +61,7 @@ class actor_proxy : public enable_weak_ptr_mixin<actor> { ...@@ -61,8 +61,7 @@ class actor_proxy : public enable_weak_ptr_mixin<actor> {
protected: protected:
actor_proxy(actor_id mid, actor_proxy(actor_id mid);
const process_information_ptr& pinfo);
}; };
......
...@@ -76,6 +76,7 @@ class default_actor_addressing : public actor_addressing { ...@@ -76,6 +76,7 @@ class default_actor_addressing : public actor_addressing {
private: private:
default_protocol* m_parent; default_protocol* m_parent;
process_information_ptr m_pinf;
std::map<process_information,proxy_map> m_proxies; std::map<process_information,proxy_map> m_proxies;
}; };
......
...@@ -65,6 +65,10 @@ class default_actor_proxy : public detail::abstract_actor<actor_proxy> { ...@@ -65,6 +65,10 @@ class default_actor_proxy : public detail::abstract_actor<actor_proxy> {
void local_unlink_from(const actor_ptr& other); void local_unlink_from(const actor_ptr& other);
inline const process_information_ptr& process_info() const {
return m_pinf;
}
protected: protected:
~default_actor_proxy(); ~default_actor_proxy();
...@@ -75,7 +79,8 @@ class default_actor_proxy : public detail::abstract_actor<actor_proxy> { ...@@ -75,7 +79,8 @@ class default_actor_proxy : public detail::abstract_actor<actor_proxy> {
any_tuple msg, any_tuple msg,
message_id_t mid = message_id_t()); message_id_t mid = message_id_t());
default_protocol_ptr m_proto; default_protocol_ptr m_proto;
process_information_ptr m_pinf;
}; };
......
...@@ -44,22 +44,19 @@ ...@@ -44,22 +44,19 @@
#include "cppa/detail/actor_registry.hpp" #include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp" #include "cppa/detail/singleton_manager.hpp"
namespace cppa {
namespace { namespace {
inline cppa::detail::actor_registry& registry() { inline detail::actor_registry& registry() {
return *(cppa::detail::singleton_manager::get_actor_registry()); return *(detail::singleton_manager::get_actor_registry());
} }
} // namespace <anonymous> } // namespace <anonymous>
namespace cppa { actor::actor(actor_id aid) : m_id(aid), m_is_proxy(true) { }
actor::actor(actor_id aid, const process_information_ptr& pptr) actor::actor() : m_id(registry().next_id()), m_is_proxy(false) { }
: m_id(aid), m_is_proxy(true), m_parent_process(pptr) {
if (!pptr) {
throw std::logic_error("parent == nullptr");
}
}
bool actor::chained_enqueue(actor* sender, any_tuple msg) { bool actor::chained_enqueue(actor* sender, any_tuple msg) {
enqueue(sender, std::move(msg)); enqueue(sender, std::move(msg));
...@@ -71,11 +68,4 @@ bool actor::chained_sync_enqueue(actor* ptr, message_id_t id, any_tuple msg) { ...@@ -71,11 +68,4 @@ bool actor::chained_sync_enqueue(actor* ptr, message_id_t id, any_tuple msg) {
return false; return false;
} }
actor::actor(const process_information_ptr& pptr)
: m_id(registry().next_id()), m_is_proxy(false), m_parent_process(pptr) {
if (!pptr) {
throw std::logic_error("parent == nullptr");
}
}
} // namespace cppa } // namespace cppa
...@@ -45,7 +45,6 @@ using namespace std; ...@@ -45,7 +45,6 @@ using namespace std;
namespace cppa { namespace cppa {
actor_proxy::actor_proxy(actor_id mid, const process_information_ptr& pptr) actor_proxy::actor_proxy(actor_id mid) : super(mid) { }
: super(mid, pptr) { }
} // namespace cppa } // namespace cppa
...@@ -47,7 +47,7 @@ using namespace std; ...@@ -47,7 +47,7 @@ using namespace std;
namespace cppa { namespace network { namespace cppa { namespace network {
default_actor_addressing::default_actor_addressing(default_protocol* parent) default_actor_addressing::default_actor_addressing(default_protocol* parent)
: m_parent(parent) { } : m_parent(parent), m_pinf(process_information::get()) { }
atom_value default_actor_addressing::technology_id() const { atom_value default_actor_addressing::technology_id() const {
return atom("DEFAULT"); return atom("DEFAULT");
...@@ -62,16 +62,24 @@ void default_actor_addressing::write(serializer* sink, const actor_ptr& ptr) { ...@@ -62,16 +62,24 @@ void default_actor_addressing::write(serializer* sink, const actor_ptr& ptr) {
} }
else { else {
// local actor? // local actor?
if (*ptr->parent_process_ptr() == *process_information::get()) { if (!ptr->is_proxy()) {
detail::singleton_manager::get_actor_registry()->put(ptr->id(), ptr); detail::singleton_manager::get_actor_registry()->put(ptr->id(), ptr);
} }
auto pinf = m_pinf;
if (ptr->is_proxy()) {
auto dptr = ptr.downcast<default_actor_proxy>();
if (dptr) pinf = dptr->process_info();
else {
CPPA_LOG_ERROR("ptr is not a default_actor_proxy instance");
}
}
primitive_variant ptup[2]; primitive_variant ptup[2];
ptup[0] = ptr->id(); ptup[0] = ptr->id();
ptup[1] = ptr->parent_process().process_id(); ptup[1] = pinf->process_id();
sink->begin_object("@actor"); sink->begin_object("@actor");
sink->write_tuple(2, ptup); sink->write_tuple(2, ptup);
sink->write_raw(process_information::node_id_size, sink->write_raw(process_information::node_id_size,
ptr->parent_process().node_id().data()); pinf->node_id().data());
sink->end_object(); sink->end_object();
} }
} }
......
...@@ -43,13 +43,12 @@ namespace cppa { namespace network { ...@@ -43,13 +43,12 @@ namespace cppa { namespace network {
default_actor_proxy::default_actor_proxy(actor_id mid, default_actor_proxy::default_actor_proxy(actor_id mid,
const process_information_ptr& pinfo, const process_information_ptr& pinfo,
const default_protocol_ptr& parent) const default_protocol_ptr& parent)
: super(mid, pinfo), m_proto(parent) { } : super(mid), m_proto(parent), m_pinf(pinfo) { }
default_actor_proxy::~default_actor_proxy() { default_actor_proxy::~default_actor_proxy() {
CPPA_LOG_TRACE("node = " << to_string(*parent_process_ptr()) CPPA_LOG_TRACE("node = " << to_string(*m_pinf) << ", aid = " << id());
<< ", aid = " << id());
auto aid = id(); auto aid = id();
auto node = parent_process_ptr(); auto node = m_pinf;
auto proto = m_proto; auto proto = m_proto;
proto->run_later([aid, node, proto] { proto->run_later([aid, node, proto] {
CPPA_LOGF_TRACE("lambda from ~default_actor_proxy" CPPA_LOGF_TRACE("lambda from ~default_actor_proxy"
...@@ -67,7 +66,7 @@ default_actor_proxy::~default_actor_proxy() { ...@@ -67,7 +66,7 @@ default_actor_proxy::~default_actor_proxy() {
void default_actor_proxy::forward_msg(const actor_ptr& sender, any_tuple msg, message_id_t mid) { void default_actor_proxy::forward_msg(const actor_ptr& sender, any_tuple msg, message_id_t mid) {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
auto node = parent_process_ptr(); auto node = m_pinf;
actor_ptr receiver = this; actor_ptr receiver = this;
auto proto = m_proto; auto proto = m_proto;
m_proto->run_later([proto, node, sender, receiver, msg, mid] { m_proto->run_later([proto, node, sender, receiver, msg, mid] {
......
...@@ -67,7 +67,7 @@ class local_group : public group { ...@@ -67,7 +67,7 @@ class local_group : public group {
public: public:
void send_all_subscribers(actor* sender, const any_tuple& msg) { void send_all_subscribers(actor* sender, const any_tuple& msg) {
shared_guard guard(m_shared_mtx); shared_guard guard(m_mtx);
for (auto& s : m_subscribers) { for (auto& s : m_subscribers) {
s->enqueue(sender, msg); s->enqueue(sender, msg);
} }
...@@ -79,7 +79,7 @@ class local_group : public group { ...@@ -79,7 +79,7 @@ class local_group : public group {
} }
pair<bool, size_t> add_subscriber(const channel_ptr& who) { pair<bool, size_t> add_subscriber(const channel_ptr& who) {
exclusive_guard guard(m_shared_mtx); exclusive_guard guard(m_mtx);
if (m_subscribers.insert(who).second) { if (m_subscribers.insert(who).second) {
return {true, m_subscribers.size()}; return {true, m_subscribers.size()};
} }
...@@ -87,7 +87,7 @@ class local_group : public group { ...@@ -87,7 +87,7 @@ class local_group : public group {
} }
pair<bool, size_t> erase_subscriber(const channel_ptr& who) { pair<bool, size_t> erase_subscriber(const channel_ptr& who) {
exclusive_guard guard(m_shared_mtx); exclusive_guard guard(m_mtx);
auto erased_one = m_subscribers.erase(who) > 0; auto erased_one = m_subscribers.erase(who) > 0;
return {erased_one, m_subscribers.size()}; return {erased_one, m_subscribers.size()};
} }
...@@ -105,26 +105,15 @@ class local_group : public group { ...@@ -105,26 +105,15 @@ class local_group : public group {
void serialize(serializer* sink); void serialize(serializer* sink);
inline const process_information& process() const {
return *m_process;
}
inline const process_information_ptr& process_ptr() const {
return m_process;
}
const actor_ptr& broker() const { const actor_ptr& broker() const {
return m_broker; return m_broker;
} }
local_group(bool spawn_local_broker, local_group(bool spawn_local_broker, local_group_module* mod, string id);
local_group_module* mod, string id,
process_information_ptr parent = process_information::get());
protected: protected:
process_information_ptr m_process; util::shared_spinlock m_mtx;
util::shared_spinlock m_shared_mtx;
set<channel_ptr> m_subscribers; set<channel_ptr> m_subscribers;
actor_ptr m_broker; actor_ptr m_broker;
...@@ -297,23 +286,20 @@ class local_group_module : public group::module { ...@@ -297,23 +286,20 @@ class local_group_module : public group::module {
m_actor_utype->deserialize(&broker, source); m_actor_utype->deserialize(&broker, source);
CPPA_REQUIRE(broker != nullptr); CPPA_REQUIRE(broker != nullptr);
if (!broker) return nullptr; if (!broker) return nullptr;
if (broker->parent_process() == process()) { if (!broker->is_proxy()) {
return this->get(identifier); return this->get(identifier);
} }
else { else {
auto& pinf = broker->parent_process();
shared_guard guard(m_proxies_mtx); shared_guard guard(m_proxies_mtx);
auto& node_map = m_proxies[pinf]; auto i = m_proxies.find(broker);
auto i = node_map.find(identifier); if (i != m_proxies.end()) {
if (i != node_map.end()) {
return i->second; return i->second;
} }
else { else {
local_group_ptr tmp(new local_group_proxy(broker, this, local_group_ptr tmp(new local_group_proxy(broker, this,
identifier, identifier));
broker->parent_process_ptr()));
upgrade_guard uguard(guard); upgrade_guard uguard(guard);
auto p = node_map.insert(make_pair(identifier, tmp)); auto p = m_proxies.insert(make_pair(broker, tmp));
// someone might preempt us // someone might preempt us
return p.first->second; return p.first->second;
} }
...@@ -333,14 +319,12 @@ class local_group_module : public group::module { ...@@ -333,14 +319,12 @@ class local_group_module : public group::module {
private: private:
typedef map<string, local_group_ptr> local_group_map;
process_information_ptr m_process; process_information_ptr m_process;
const uniform_type_info* m_actor_utype; const uniform_type_info* m_actor_utype;
util::shared_spinlock m_instances_mtx; util::shared_spinlock m_instances_mtx;
local_group_map m_instances; map<string,local_group_ptr> m_instances;
util::shared_spinlock m_proxies_mtx; util::shared_spinlock m_proxies_mtx;
map<process_information, local_group_map> m_proxies; map<actor_ptr,local_group_ptr> m_proxies;
}; };
...@@ -545,12 +529,9 @@ class remote_group_module : public group::module { ...@@ -545,12 +529,9 @@ class remote_group_module : public group::module {
local_group::local_group(bool spawn_local_broker, local_group::local_group(bool spawn_local_broker,
local_group_module* mod, local_group_module* mod,
string id, string id)
process_information_ptr parent) : group(mod, move(id)) {
: group(mod, move(id)), m_process(move(parent)) { if (spawn_local_broker) m_broker = spawn_hidden<local_broker>(this);
if (spawn_local_broker) {
m_broker = spawn_hidden<local_broker>(this);
}
} }
void local_group::serialize(serializer* sink) { void local_group::serialize(serializer* sink) {
......
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