Commit b5bc1675 authored by Dominik Charousset's avatar Dominik Charousset

actor_registry optimization

parent d2423e00
......@@ -195,13 +195,6 @@ class actor : public channel
*/
inline actor_id id() const;
/**
* @brief Get the actor that has the unique identifier @p needle.
* @returns A pointer to the requestet actor or @c nullptr if no
* running actor with the ID @p actor_id was found in this process.
*/
static intrusive_ptr<actor> by_id(actor_id needle);
inline bool is_proxy() const;
};
......
......@@ -36,6 +36,7 @@
#include <cstdint>
#include "cppa/actor.hpp"
#include "cppa/attachable.hpp"
#include "cppa/detail/thread.hpp"
#include "cppa/util/shared_spinlock.hpp"
......@@ -48,14 +49,12 @@ class actor_registry
actor_registry();
// adds @p whom to the list of known actors
void add(actor* whom);
// return nullptr if the actor wasn't put *or* finished execution
actor_ptr get(actor_id key) const;
// removes @p whom from the list of known actors
void remove(actor* whom);
void put(actor_id key, actor_ptr const& value);
// looks for an actor with id @p whom in the list of known actors
actor_ptr find(actor_id whom);
void erase(actor_id key);
// gets the next free actor id
actor_id next_id();
......@@ -66,7 +65,7 @@ class actor_registry
// decreases running-actors-count by one
void dec_running();
size_t running();
size_t running() const;
// blocks the caller until running-actors-count becomes @p expected
void await_running_count_equal(size_t expected);
......@@ -79,8 +78,8 @@ class actor_registry
mutex m_running_mtx;
condition_variable m_running_cv;
util::shared_spinlock m_instances_mtx;
std::map<std::uint32_t, actor*> m_instances;
mutable util::shared_spinlock m_instances_mtx;
std::map<std::uint32_t, actor_ptr> m_instances;
};
......
......@@ -200,8 +200,34 @@ class intrusive_ptr : util::comparable<intrusive_ptr<T>, T const*>,
return compare(other.get());
}
template<class C>
intrusive_ptr<C> downcast() const
{
if (m_ptr) return dynamic_cast<C*>(const_cast<T*>(m_ptr));
return nullptr;
}
template<class C>
intrusive_ptr<C> upcast() const
{
if (m_ptr) return static_cast<C*>(const_cast<T*>(m_ptr));
return nullptr;
}
};
template<typename X>
inline bool operator==(intrusive_ptr<X> const& lhs, decltype(nullptr))
{
return lhs.get() == nullptr;
}
template<typename X>
inline bool operator==(decltype(nullptr), intrusive_ptr<X> const& rhs)
{
return rhs.get() == nullptr;
}
template<typename X, typename Y>
bool operator==(intrusive_ptr<X> const& lhs, intrusive_ptr<Y> const& rhs)
{
......
......@@ -70,23 +70,10 @@ actor::actor(const process_information_ptr& pptr)
{
throw std::logic_error("parent == nullptr");
}
else
{
registry().add(this);
}
}
actor::~actor()
{
if (!m_is_proxy)
{
registry().remove(this);
}
}
intrusive_ptr<actor> actor::by_id(actor_id whom)
{
return registry().find(whom);
}
void actor::join(group_ptr& what)
......
......@@ -32,13 +32,16 @@
#include <limits>
#include <stdexcept>
#include "cppa/attachable.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/util/shared_lock_guard.hpp"
#include "cppa/util/upgrade_lock_guard.hpp"
namespace {
typedef cppa::detail::lock_guard<cppa::util::shared_spinlock> exclusive_guard;
typedef cppa::util::shared_lock_guard<cppa::util::shared_spinlock> shared_guard;
typedef cppa::util::upgrade_lock_guard<cppa::util::shared_spinlock> upgrade_guard;
} // namespace <anonymous>
......@@ -48,31 +51,57 @@ actor_registry::actor_registry() : m_running(0), m_ids(1)
{
}
void actor_registry::add(actor* whom)
actor_ptr actor_registry::get(actor_id key) const
{
exclusive_guard guard(m_instances_mtx);
m_instances.insert(std::make_pair(whom->id(), whom));
}
void actor_registry::remove(actor* whom)
{
exclusive_guard guard(m_instances_mtx);
m_instances.erase(whom->id());
shared_guard guard(m_instances_mtx);
auto i = m_instances.find(key);
if (i != m_instances.end())
{
return i->second;
}
return nullptr;
}
actor_ptr actor_registry::find(actor_id whom)
void actor_registry::put(actor_id key, actor_ptr const& value)
{
actor_ptr result;
// lifetime scope of guard
bool add_attachable = false;
if (value != nullptr)
{
shared_guard guard(m_instances_mtx);
auto i = m_instances.find(whom);
if (i != m_instances.end())
auto i = m_instances.find(key);
if (i == m_instances.end())
{
result.reset(i->second);
upgrade_guard uguard(guard);
m_instances.insert(std::make_pair(key, value));
add_attachable = true;
}
}
return std::move(result);
if (add_attachable)
{
struct eraser : attachable
{
actor_id m_id;
actor_registry* m_singleton;
eraser(actor_id id, actor_registry* s) : m_id(id), m_singleton(s)
{
}
void actor_exited(std::uint32_t)
{
m_singleton->erase(m_id);
}
bool matches(token const&)
{
return false;
}
};
const_cast<actor_ptr&>(value)->attach(new eraser(value->id(), this));
}
}
void actor_registry::erase(actor_id key)
{
exclusive_guard guard(m_instances_mtx);
m_instances.insert(std::pair<actor_id, actor_ptr>(key, nullptr));
}
std::uint32_t actor_registry::next_id()
......@@ -85,7 +114,7 @@ void actor_registry::inc_running()
++m_running;
}
size_t actor_registry::running()
size_t actor_registry::running() const
{
return m_running.load();
}
......
......@@ -36,7 +36,13 @@
#include "cppa/binary_serializer.hpp"
#include "cppa/detail/post_office.hpp"
//#define DEBUG(arg) std::cout << arg << std::endl
/*
#define DEBUG(arg) \
std::cout << "[process id: " \
<< cppa::process_information::get()->process_id() \
<< "] " << arg << std::endl
*/
#define DEBUG(unused) ((void) 0)
// implementation of mailman.hpp
......
......@@ -326,9 +326,13 @@ class po_peer
&& content.utype_info_at(0) == typeid(atom_value)
&& content.get_as<atom_value>(0) == atom(":Monitor"))
{
auto receiver_ch = msg.receiver();
actor_ptr receiver = dynamic_cast<actor*>(receiver_ch.get());
if (receiver->parent_process() == *process_information::get())
auto receiver = msg.receiver().downcast<actor>();
//actor_ptr receiver = dynamic_cast<actor*>(receiver_ch.get());
if (!receiver)
{
DEBUG("empty receiver");
}
else if (receiver->parent_process() == *process_information::get())
{
//cout << pinfo << " ':Monitor'; actor id = "
// << sender->id() << endl;
......@@ -352,8 +356,15 @@ class po_peer
}
else
{
msg.receiver()->enqueue(msg.sender().get(),
std::move(msg.content()));
if (msg.receiver())
{
msg.receiver()->enqueue(msg.sender().get(),
std::move(msg.content()));
}
else
{
DEBUG("empty receiver");
}
}
m_rdbuf.reset();
m_state = wait_for_msg_size;
......
......@@ -52,7 +52,9 @@
#include "cppa/detail/mailman.hpp"
#include "cppa/detail/post_office.hpp"
#include "cppa/detail/native_socket.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
#include "cppa/detail/singleton_manager.hpp"
using std::cout;
using std::endl;
......@@ -111,6 +113,7 @@ struct socket_guard
void publish(actor_ptr& whom, std::uint16_t port)
{
if (!whom) return;
detail::singleton_manager::get_actor_registry()->put(whom->id(), whom);
detail::native_socket_type sockfd;
struct sockaddr_in serv_addr;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
......
......@@ -58,6 +58,7 @@
#include "cppa/detail/demangle.hpp"
#include "cppa/detail/object_array.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/to_uniform_name.hpp"
#include "cppa/detail/addressed_message.hpp"
#include "cppa/detail/singleton_manager.hpp"
......@@ -181,16 +182,20 @@ class actor_ptr_tinfo : public util::abstract_uniform_type_info<actor_ptr>
public:
static void s_serialize(const actor* ptr,
static void s_serialize(actor_ptr const& ptr,
serializer* sink,
const std::string name)
{
if (!ptr)
if (ptr == nullptr)
{
serialize_nullptr(sink);
}
else
{
if (ptr->is_proxy() == false)
{
singleton_manager::get_actor_registry()->put(ptr->id(), ptr);
}
primitive_variant ptup[3];
ptup[0] = ptr->id();
ptup[1] = ptr->parent_process().process_id();
......@@ -231,7 +236,9 @@ class actor_ptr_tinfo : public util::abstract_uniform_type_info<actor_ptr>
if ( pinf->process_id() == get<std::uint32_t>(ptup[1])
&& cppa::equal(nstr, pinf->node_id()))
{
ptrref = actor::by_id(get<std::uint32_t>(ptup[0]));
auto id = get<std::uint32_t>(ptup[0]);
ptrref = singleton_manager::get_actor_registry()->get(id);
//ptrref = actor::by_id(get<std::uint32_t>(ptup[0]));
}
else
{
......@@ -248,7 +255,7 @@ class actor_ptr_tinfo : public util::abstract_uniform_type_info<actor_ptr>
void serialize(const void* ptr, serializer* sink) const
{
s_serialize(reinterpret_cast<const actor_ptr*>(ptr)->get(),
s_serialize(*reinterpret_cast<const actor_ptr*>(ptr),
sink,
name());
}
......@@ -265,11 +272,11 @@ class group_ptr_tinfo : public util::abstract_uniform_type_info<group_ptr>
public:
static void s_serialize(const group* ptr,
static void s_serialize(group_ptr const& ptr,
serializer* sink,
const std::string& name)
{
if (!ptr)
if (ptr == nullptr)
{
serialize_nullptr(sink);
}
......@@ -314,7 +321,7 @@ class group_ptr_tinfo : public util::abstract_uniform_type_info<group_ptr>
void serialize(const void* ptr, serializer* sink) const
{
s_serialize(reinterpret_cast<const group_ptr*>(ptr)->get(),
s_serialize(*reinterpret_cast<const group_ptr*>(ptr),
sink,
name());
}
......@@ -334,26 +341,26 @@ class channel_ptr_tinfo : public util::abstract_uniform_type_info<channel_ptr>
public:
static void s_serialize(const channel* ptr,
static void s_serialize(channel_ptr const& ptr,
serializer* sink,
const std::string& channel_type_name,
const std::string& actor_ptr_type_name,
const std::string& group_ptr_type_name)
{
sink->begin_object(channel_type_name);
if (!ptr)
if (ptr == nullptr)
{
serialize_nullptr(sink);
}
else
{
const group* gptr;
auto aptr = dynamic_cast<const actor*>(ptr);
group_ptr gptr;
auto aptr = ptr.downcast<actor>();
if (aptr)
{
actor_ptr_tinfo::s_serialize(aptr, sink, actor_ptr_type_name);
}
else if ((gptr = dynamic_cast<const group*>(ptr)) != nullptr)
else if ((gptr = ptr.downcast<group>()))
{
group_ptr_tinfo::s_serialize(gptr, sink, group_ptr_type_name);
}
......@@ -408,7 +415,7 @@ class channel_ptr_tinfo : public util::abstract_uniform_type_info<channel_ptr>
void serialize(const void* instance, serializer* sink) const
{
s_serialize(reinterpret_cast<const channel_ptr*>(instance)->get(),
s_serialize(*reinterpret_cast<const channel_ptr*>(instance),
sink,
name(),
actor_ptr_name,
......@@ -501,8 +508,8 @@ class addr_msg_tinfo : public util::abstract_uniform_type_info<addressed_message
const addressed_message& msg = *reinterpret_cast<const addressed_message*>(instance);
const any_tuple& data = msg.content();
sink->begin_object(name());
actor_ptr_tinfo::s_serialize(msg.sender().get(), sink, actor_ptr_name);
channel_ptr_tinfo::s_serialize(msg.receiver().get(),
actor_ptr_tinfo::s_serialize(msg.sender(), sink, actor_ptr_name);
channel_ptr_tinfo::s_serialize(msg.receiver(),
sink,
channel_ptr_name,
actor_ptr_name,
......
......@@ -24,6 +24,7 @@
#include "test.hpp"
#include "cppa/self.hpp"
#include "cppa/tuple.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/announce.hpp"
......@@ -134,6 +135,16 @@ size_t test__serialization()
CPPA_ERROR("exception: " << e.what());
}
{
any_tuple ttup = make_tuple(1, 2, actor_ptr(self));
binary_serializer bs;
bs << ttup;
binary_deserializer bd(bs.data(), bs.size());
any_tuple ttup2;
uniform_typeid<any_tuple>()->deserialize(&ttup2, &bd);
CPPA_CHECK_EQUAL(ttup, ttup2);
}
{
// serialize b1 to buf
binary_serializer bs;
......
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