Commit 4513ad61 authored by Dominik Charousset's avatar Dominik Charousset

use per-thread caching for actors

this patch extends the caching strategy to actor instances to minimize
access to the heap through new/delete
parent 4c617725
......@@ -117,6 +117,7 @@ set(LIBCPPA_SRC
src/logging.cpp
src/match.cpp
src/memory.cpp
src/memory_managed.cpp
src/message_header.cpp
src/middleman.cpp
src/object.cpp
......@@ -127,6 +128,7 @@ set(LIBCPPA_SRC
src/process_information.cpp
src/protocol.cpp
src/receive.cpp
src/recursive_queue_node.cpp
src/ref_counted.cpp
src/response_handle.cpp
src/ripemd_160.cpp
......
......@@ -277,3 +277,7 @@ unit_testing/test__uniform_type.cpp
unit_testing/test__yield_interface.cpp
cppa/detail/memory.hpp
src/memory.cpp
src/recursive_queue_node.cpp
cppa/intrusive_fwd_ptr.hpp
cppa/memory_managed.hpp
src/memory_managed.cpp
......@@ -64,6 +64,7 @@
#include "cppa/util/rm_ref.hpp"
#include "cppa/network/acceptor.hpp"
#include "cppa/detail/memory.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/get_behavior.hpp"
#include "cppa/detail/receive_loop_helper.hpp"
......@@ -638,7 +639,6 @@ inline const self_type& operator<<(const self_type& s, any_tuple what) {
* @{
*/
/**
* @brief Spawns a new context-switching or thread-mapped {@link actor}
* that executes <tt>fun(args...)</tt>.
......@@ -719,7 +719,8 @@ actor_ptr spawn_in_group(Fun&& fun, Args&&... args) {
*/
template<class ActorImpl, typename... Args>
actor_ptr spawn(Args&&... args) {
return get_scheduler()->spawn(new ActorImpl(std::forward<Args>(args)...));
auto ptr = detail::memory::create<ActorImpl>(std::forward<Args>(args)...);
return get_scheduler()->spawn(ptr);
}
/**
......@@ -735,23 +736,23 @@ actor_ptr spawn(Args&&... args) {
*/
template<class ActorImpl, typename... Args>
actor_ptr spawn_in_group(const group_ptr& grp, Args&&... args) {
return get_scheduler()->spawn(new ActorImpl(std::forward<Args>(args)...),
[&grp](local_actor* ptr) { ptr->join(grp); });
auto ptr = detail::memory::create<ActorImpl>(std::forward<Args>(args)...);
return get_scheduler()->spawn(ptr, [&](local_actor* p) { p->join(grp); });
}
#ifndef CPPA_DOCUMENTATION
template<class ActorImpl, typename... Args>
actor_ptr spawn_hidden_in_group(const group_ptr& grp, Args&&... args) {
return get_scheduler()->spawn(new ActorImpl(std::forward<Args>(args)...),
[&grp](local_actor* ptr) { ptr->join(grp); },
auto ptr = detail::memory::create<ActorImpl>(std::forward<Args>(args)...);
return get_scheduler()->spawn(ptr, [&](local_actor* p) { p->join(grp); },
scheduled_and_hidden);
}
template<class ActorImpl, typename... Args>
actor_ptr spawn_hidden(Args&&... args) {
return get_scheduler()->spawn(new ActorImpl(std::forward<Args>(args)...),
scheduled_and_hidden);
auto ptr = detail::memory::create<ActorImpl>(std::forward<Args>(args)...);
return get_scheduler()->spawn(ptr, scheduled_and_hidden);
}
#endif
......
......@@ -187,9 +187,9 @@ class abstract_actor : public abstract_actor_base<Base, std::is_base_of<local_ac
inline mailbox_element* fetch_node(actor* sender,
any_tuple msg,
message_id_t id = message_id_t()) {
auto result = memory::new_queue_node();
result->reset(sender, std::move(msg), id);
return result;
return memory::create<mailbox_element>(sender, std::move(msg), id);
//result->reset(sender, std::move(msg), id);
//return result;
}
template<typename... Args>
......
......@@ -31,26 +31,197 @@
#ifndef CPPA_MEMORY_HPP
#define CPPA_MEMORY_HPP
#include <new>
#include <vector>
#include <utility>
#include <typeinfo>
#include "cppa/detail/recursive_queue_node.hpp"
namespace cppa { namespace detail {
class memory_cache;
namespace {
constexpr size_t s_alloc_size = 1024; // allocate ~1kb chunks
constexpr size_t s_cache_size = 10240; // cache about 10kb per thread
} // namespace <anonymous>
class recursive_queue_node;
class instance_wrapper {
public:
virtual ~instance_wrapper();
// calls the destructor
virtual void destroy() = 0;
// releases memory
virtual void deallocate() = 0;
};
class memory_cache {
public:
virtual ~memory_cache();
// calls dtor and either releases memory or re-uses it later
virtual void release_instance(void*) = 0;
virtual std::pair<instance_wrapper*,void*> new_instance() = 0;
// casts @p ptr to the derived type and returns it
virtual void* downcast(memory_managed* ptr) = 0;
};
template<typename T>
class basic_memory_cache : public memory_cache {
struct wrapper : instance_wrapper {
ref_counted* parent;
union { T instance; };
wrapper() : parent(nullptr) { }
~wrapper() { }
void destroy() { instance.~T(); }
void deallocate() { parent->deref(); }
};
class storage : public ref_counted {
public:
storage() {
for (auto& elem : data) {
// each instance has a reference to its parent
elem.parent = this;
ref(); // deref() is called in wrapper::deallocate
}
}
typedef wrapper* iterator;
iterator begin() { return std::begin(data); }
iterator end() { return std::end(data); }
private:
wrapper data[s_alloc_size / sizeof(T)];
};
public:
std::vector<wrapper*> cached_elements;
basic_memory_cache() {
cached_elements.reserve(s_cache_size / sizeof(T));
}
~basic_memory_cache() {
for (auto e : cached_elements) e->deallocate();
}
void* downcast(memory_managed* ptr) {
return static_cast<T*>(ptr);
}
virtual void release_instance(void* vptr) {
CPPA_REQUIRE(vptr != nullptr);
auto ptr = reinterpret_cast<T*>(vptr);
CPPA_REQUIRE(ptr->outer_memory != nullptr);
auto wptr = static_cast<wrapper*>(ptr->outer_memory);
wptr->destroy();
if (cached_elements.capacity() > 0) cached_elements.push_back(wptr);
else wptr->deallocate();
}
virtual std::pair<instance_wrapper*,void*> new_instance() {
if (cached_elements.empty()) {
auto elements = new storage;
for (auto i = elements->begin(); i != elements->end(); ++i) {
cached_elements.push_back(i);
}
}
wrapper* wptr = cached_elements.back();
cached_elements.pop_back();
return std::make_pair(wptr, &(wptr->instance));
}
};
class memory {
memory() = delete;
friend class memory_cache;
template<typename>
friend class basic_memory_cache;
public:
static recursive_queue_node* new_queue_node();
/*
* @brief Allocates storage, initializes a new object, and returns
* the new instance.
*/
template<typename T, typename... Args>
static inline T* create(Args&&... args) {
auto mc = get_or_set_cache_map_entry<T>();
auto p = mc->new_instance();
auto result = new (p.second) T (std::forward<Args>(args)...);
result->outer_memory = p.first;
return result;
}
static void dispose(recursive_queue_node* ptr);
/*
* @brief Calls the destructor of @p ptr and allows the memory manager
* to reuse this address.
*/
template<typename T>
static inline void dispose(T* ptr) {
auto mc = get_or_set_cache_map_entry<T>();
mc->release_instance(ptr);
}
/*
* @brief Calls the destructor of @p ptr and allows the memory manager
* to reuse this address. This function must be used when disposing
* through a base pointer.
*/
template<typename T>
static inline void dispose_base(T* ptr) {
auto mc = get_cache_map_entry(&typeid(*ptr));
if (mc) mc->release_instance(mc->downcast(ptr));
else {
auto wptr = ptr->outer_memory;
if (wptr) {
wptr->destroy();
wptr->deallocate();
}
else delete ptr;
}
}
private:
static void destroy(recursive_queue_node* ptr);
static memory_cache* get_cache_map_entry(const std::type_info* tinf);
static void add_cache_map_entry(const std::type_info* tinf, memory_cache* instance);
template<typename T>
static inline memory_cache* get_or_set_cache_map_entry() {
auto mc = get_cache_map_entry(&typeid(T));
if (!mc) {
mc = new basic_memory_cache<T>;
add_cache_map_entry(&typeid(T), mc);
}
return mc;
}
};
......
......@@ -61,6 +61,7 @@ class receive_policy {
public:
typedef recursive_queue_node* pointer;
typedef std::unique_ptr<recursive_queue_node,disposer> smart_pointer;
enum handle_message_result {
hm_timeout_msg,
......@@ -103,22 +104,21 @@ class receive_policy {
template<class Client, class Fun>
bool invoke(Client* client,
pointer node,
pointer node_ptr,
Fun& fun,
message_id_t awaited_response = message_id_t()) {
smart_pointer node(node_ptr);
std::integral_constant<receive_policy_flag,Client::receive_flag> policy;
switch (this->handle_message(client, node, fun,
switch (this->handle_message(client, node.get(), fun,
awaited_response, policy)) {
case hm_msg_handled: {
memory::dispose(node);
return true;
}
case hm_drop_msg: {
memory::dispose(node);
break;
}
case hm_cache_msg: {
m_cache.emplace_back(node);
m_cache.emplace_back(std::move(node));
break;
}
case hm_skip_msg: {
......
......@@ -35,8 +35,8 @@
#include "cppa/actor.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/message_id.hpp"
#include "cppa/ref_counted.hpp"
// needs access to constructor + destructor to initialize m_dummy_node
namespace cppa { class local_actor; }
......@@ -44,13 +44,15 @@ namespace cppa { class local_actor; }
namespace cppa { namespace detail {
class memory;
class recursive_queue_node_storage;
class instance_wrapper;
class recursive_queue_node : public memory_managed {
class recursive_queue_node {
template<typename>
friend class basic_memory_cache;
friend class memory;
friend class local_actor;
friend class recursive_queue_node_storage;
public:
......@@ -62,19 +64,6 @@ class recursive_queue_node {
any_tuple msg; // 'content field'
message_id_t mid;
inline void reset(actor* sptr, message_id_t id, bool reset_msg = true) {
next = nullptr;
marked = false;
sender = sptr;
mid = id;
if (reset_msg) msg.reset();
}
inline void reset(actor* sptr, any_tuple&& data, message_id_t id) {
reset(sptr, id, false);
msg = std::move(data);
}
recursive_queue_node(recursive_queue_node&&) = delete;
recursive_queue_node(const recursive_queue_node&) = delete;
recursive_queue_node& operator=(recursive_queue_node&&) = delete;
......@@ -82,10 +71,14 @@ class recursive_queue_node {
private:
inline recursive_queue_node() { reset(nullptr, message_id_t(), false); }
inline ~recursive_queue_node() { }
recursive_queue_node() = default;
recursive_queue_node(actor_ptr sptr, any_tuple data, message_id_t id);
~recursive_queue_node();
recursive_queue_node_storage* parent;
// a pointer to the outer memory region this instance belongs to
instance_wrapper* outer_memory;
};
......
......@@ -52,6 +52,12 @@ class scheduler;
class message_future;
class local_scheduler;
namespace detail {
class memory;
class instance_wrapper;
template<typename> class basic_memory_cache;
} // namespace detail
template<bool DiscardOld>
struct behavior_policy { static const bool discard_old = DiscardOld; };
......@@ -111,8 +117,11 @@ class message_future;
*/
class local_actor : public actor {
typedef actor super;
friend class scheduler;
//friend inline sync_recv_helper receive_response(message_future);
friend class detail::memory;
template<typename> friend class detail::basic_memory_cache;
public:
......@@ -449,16 +458,11 @@ class local_actor : public actor {
do_become(std::move(copy), discard_old);
}
/*
* @brief Waits for a response to @p request_id.
* Blocks until either a response message arrives or until a
* timeout occurs.
* @param bhvr A behavior with timeout denoting the
* actor's response to the response message.
* @warning You should not call this member function by hand.
* Use always the {@link cppa::receive_response receive_response}
* function.
*/
void request_deletion();
private:
detail::instance_wrapper* outer_memory;
};
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_MEMORY_MANAGED_HPP
#define CPPA_MEMORY_MANAGED_HPP
namespace cppa {
class memory_managed {
protected:
virtual ~memory_managed();
/**
* @brief Default implementations calls <tt>delete this</tt>, but can
* be overriden in case deletion depends on some condition or
* the class doesn't use default new/delete.
*/
virtual void request_deletion();
};
} // namespace cppa
#endif // CPPA_MEMORY_MANAGED_HPP
......@@ -34,6 +34,8 @@
#include <atomic>
#include <cstddef>
#include "cppa/memory_managed.hpp"
namespace cppa {
/**
......@@ -43,11 +45,11 @@ namespace cppa {
* Serves the requirements of {@link intrusive_ptr}.
* @relates intrusive_ptr
*/
class ref_counted {
class ref_counted : public memory_managed {
public:
inline ref_counted() : m_rc(0) { }
ref_counted();
/**
* @brief Increases reference count by one.
......@@ -71,13 +73,6 @@ class ref_counted {
virtual ~ref_counted();
/**
* @brief Default implementations calls <tt>delete this</tt>, but can
* be overriden in case deletion depends on more than the
* reference count alone.
*/
virtual void request_deletion();
private:
std::atomic<size_t> m_rc;
......
......@@ -70,7 +70,8 @@ class down_observer : public attachable {
local_actor::local_actor(bool sflag)
: m_chaining(sflag), m_trap_exit(false)
, m_is_scheduled(sflag), m_dummy_node(), m_current_node(&m_dummy_node) { }
, m_is_scheduled(sflag), m_dummy_node(), m_current_node(&m_dummy_node)
, outer_memory(nullptr) { }
void local_actor::monitor(actor_ptr whom) {
if (whom) whom->attach(new down_observer(this, whom));
......@@ -157,4 +158,9 @@ response_handle local_actor::make_response_handle() {
return std::move(result);
}
void local_actor::request_deletion() {
if (outer_memory) detail::memory::dispose_base(this);
else super::request_deletion();
}
} // namespace cppa
......@@ -29,6 +29,7 @@
#include <vector>
#include <typeinfo>
#include "cppa/detail/memory.hpp"
#include "cppa/detail/recursive_queue_node.hpp"
......@@ -42,90 +43,58 @@ namespace {
pthread_key_t s_key;
pthread_once_t s_key_once = PTHREAD_ONCE_INIT;
constexpr size_t s_queue_node_storage_size = 20;
constexpr size_t s_max_cached_queue_nodes = 100;
} // namespace <anonymous>
class recursive_queue_node_storage : public ref_counted {
public:
recursive_queue_node_storage() {
for (auto& instance : m_instances) {
// each instance has a reference to its parent
instance.parent = this;
ref(); // deref() is called in memory::destroy
}
}
typedef recursive_queue_node* iterator;
iterator begin() { return std::begin(m_instances); }
memory_cache::~memory_cache() { }
iterator end() { return std::end(m_instances); }
typedef map<const type_info*,unique_ptr<memory_cache> > cache_map;
private:
recursive_queue_node m_instances[s_queue_node_storage_size];
};
class memory_cache {
public:
vector<recursive_queue_node*> qnodes;
memory_cache() {
qnodes.reserve(s_max_cached_queue_nodes);
}
void cache_map_destructor(void* ptr) {
if (ptr) delete reinterpret_cast<cache_map*>(ptr);
}
~memory_cache() {
for (auto node : qnodes) memory::destroy(node);
}
void make_cache_map() {
pthread_key_create(&s_key, cache_map_destructor);
}
static void destructor(void* ptr) {
if (ptr) delete reinterpret_cast<memory_cache*>(ptr);
cache_map& get_cache_map() {
pthread_once(&s_key_once, make_cache_map);
auto cache = reinterpret_cast<cache_map*>(pthread_getspecific(s_key));
if (!cache) {
cache = new cache_map;
pthread_setspecific(s_key, cache);
// insert default types
unique_ptr<memory_cache> tmp(new basic_memory_cache<recursive_queue_node>);
cache->insert(make_pair(&typeid(recursive_queue_node), move(tmp)));
}
return *cache;
}
static void make() {
pthread_key_create(&s_key, destructor);
}
memory_cache* memory::get_cache_map_entry(const type_info* tinf) {
auto& cache = get_cache_map();
auto i = cache.find(tinf);
if (i != cache.end()) return i->second.get();
return nullptr;
}
static memory_cache* get() {
pthread_once(&s_key_once, make);
auto result = static_cast<memory_cache*>(pthread_getspecific(s_key));
if (!result) {
result = new memory_cache;
pthread_setspecific(s_key, result);
}
return result;
}
void memory::add_cache_map_entry(const type_info* tinf, memory_cache* instance) {
auto& cache = get_cache_map();
cache[tinf].reset(instance);
}
};
instance_wrapper::~instance_wrapper() { }
recursive_queue_node* memory::new_queue_node() {
auto& vec = memory_cache::get()->qnodes;
if (!vec.empty()) {
recursive_queue_node* result = vec.back();
vec.pop_back();
return result;
}
auto storage = new recursive_queue_node_storage;
for (auto i = storage->begin(); i != storage->end(); ++i) vec.push_back(i);
return new_queue_node();
}
//pair<instance_wrapper*,void*> memory::allocate(const type_info* type) {
// return get_cache_map_entry(type)->allocate();
//}
void memory::dispose(recursive_queue_node* ptr) {
auto& vec = memory_cache::get()->qnodes;
if (vec.size() < s_max_cached_queue_nodes) vec.push_back(ptr);
else destroy(ptr);
}
//recursive_queue_node* memory::new_queue_node() {
// typedef recursive_queue_node type;
// return get_cache_map_entry(&typeid(type))->typed_new<type>();
//}
void memory::destroy(recursive_queue_node* ptr) {
auto parent = ptr->parent;
parent->deref();
}
//void memory::deallocate(const type_info* type, void* ptr) {
// return get_cache_map_entry(type)->deallocate(ptr);
//}
} } // namespace cppa::detail
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include "cppa/memory_managed.hpp"
namespace cppa {
memory_managed::~memory_managed() { }
void memory_managed::request_deletion() { delete this; }
} // namespace cppa
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include "cppa/detail/recursive_queue_node.hpp"
using std::move;
namespace cppa { namespace detail {
recursive_queue_node::recursive_queue_node(actor_ptr sptr,
any_tuple data,
message_id_t id)
: next(nullptr), marked(false), sender(move(sptr))
, msg(move(data)), mid(id), outer_memory(nullptr) { }
recursive_queue_node::~recursive_queue_node() { }
} } // namespace cppa::detail
......@@ -33,8 +33,8 @@
namespace cppa {
ref_counted::~ref_counted() { }
ref_counted::ref_counted() : m_rc(0) { }
void ref_counted::request_deletion() { delete this; }
ref_counted::~ref_counted() { }
} // namespace cppa
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