Commit 5c7af235 authored by Dominik Charousset's avatar Dominik Charousset

Implement `fast_send`, reducing memory allocations

Whenever an actor calls `send(dest, v1, v2, ...)`, we previously allocated a
`mailbox_element` and a `message` object. The `fast_send` optimization joins
both objects into a single one using `pair_storage`:

```
pair_storage<mailbox_element, tuple_vals<Ts...>>:

    /-----------------------------------------------\
    |                                               |
    |     /------------\                            |
    |     |            | intrusive_ptr              | intrusive_ptr
    v     v            |                            |
+------------+-------------------+---------------------+
|  refcount  |  mailbox_element  |  tuple_vals<Ts...>  |
+------------+-------------------+---------------------+
                      ^                   ^
                      |                   |
           unique_ptr<mailbox_element,    |
                      detail::disposer>   |
                                          |
                                          |
                             intrusive_ptr<message_data>
```

Since each element in `pair_storage` lives inside a `union`, it is safe to
simply call the destructor for each element in `request_deletion`. This is done
by extending both types and override `request_deletion` accordingly (see
`detail::embedded`).  The only additional safety measure we need to take is to
make sure to swap the `intrusive_ptr` out of each element *before* calling the
destructor. This way, the destruct is guaranteed to be done before we attempt
to release the storage.
parent ade02937
......@@ -45,6 +45,15 @@ class abstract_channel : public ref_counted {
virtual void enqueue(const actor_addr& sender, message_id mid,
message content, execution_unit* host) = 0;
/**
* Enqueues a new message wrapped in a `mailbox_element` to the channel.
* This variant is used by actors whenever it is possible to allocate
* mailbox element and message on the same memory block and is thus
* more efficient. Non-actors use the default implementation which simply
* calls the pure virtual version.
*/
virtual void enqueue(mailbox_element_ptr what, execution_unit* host);
/**
* Returns the ID of the node this actor is running on.
*/
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_DETAIL_EMBEDDED_HPP
#define CAF_DETAIL_EMBEDDED_HPP
#include "caf/ref_counted.hpp"
#include "caf/intrusive_ptr.hpp"
namespace caf {
namespace detail {
template <class Base>
class embedded : public Base {
public:
template <class... Vs>
embedded(intrusive_ptr<ref_counted> storage, Vs&&... vs)
: Base(std::forward<Vs>(vs)...),
m_storage(std::move(storage)) {
// nop
}
~embedded() {
// nop
}
void request_deletion() override {
intrusive_ptr<ref_counted> guard;
guard.swap(m_storage);
// this code assumes that embedded is part of pair_storage<>,
// i.e., this object lives inside a union!
this->~embedded();
}
protected:
intrusive_ptr<ref_counted> m_storage;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_EMBEDDED_HPP
......@@ -69,29 +69,44 @@ class message_data : public ref_counted {
bool equals(const message_data& other) const;
class ptr {
public:
ptr() = default;
ptr(ptr&&) = default;
ptr(const ptr&) = default;
ptr& operator=(ptr&&) = default;
ptr& operator=(const ptr&) = default;
inline ptr(const std::nullptr_t&) {
inline explicit ptr(message_data* p) : m_ptr(p) {
// nop
}
inline explicit ptr(message_data* p) : m_ptr(p) {}
inline void detach() {
static_cast<void>(get_detached());
}
inline void detach() { static_cast<void>(get_detached()); }
inline message_data* operator->() {
return get_detached();
}
inline message_data* operator->() { return get_detached(); }
inline message_data& operator*() { return *get_detached(); }
inline const message_data* operator->() const { return m_ptr.get(); }
inline const message_data& operator*() const { return *m_ptr.get(); }
inline void swap(ptr& other) { m_ptr.swap(other.m_ptr); }
inline void reset(message_data* p = nullptr) { m_ptr.reset(p); }
inline message_data& operator*() {
return *get_detached();
}
inline const message_data* operator->() const {
return m_ptr.get();
}
inline const message_data& operator*() const {
return *m_ptr.get();
}
inline void swap(ptr& other) {
m_ptr.swap(other.m_ptr);
}
inline void reset(message_data* p = nullptr) {
m_ptr.reset(p);
}
inline explicit operator bool() const {
return static_cast<bool>(m_ptr);
......@@ -102,11 +117,8 @@ class message_data : public ref_counted {
}
private:
message_data* get_detached();
intrusive_ptr<message_data> m_ptr;
};
};
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_DETAIL_PAIR_STORAGE_HPP
#define CAF_DETAIL_PAIR_STORAGE_HPP
#include "caf/extend.hpp"
#include "caf/ref_counted.hpp"
#include "caf/detail/embedded.hpp"
#include "caf/mixin/memory_cached.hpp"
namespace caf {
namespace detail {
template <class FirstType, class SecondType>
class pair_storage : public extend<ref_counted>::
with<mixin::memory_cached> {
public:
union { embedded<FirstType> first; };
union { embedded<SecondType> second; };
template <class... Vs>
pair_storage(std::integral_constant<size_t, 0>, Vs&&... vs)
: first(this),
second(this, std::forward<Vs>(vs)...) {
// nop
}
template <class V0, class... Vs>
pair_storage(std::integral_constant<size_t, 1>, V0&& v0, Vs&&... vs)
: first(this, std::forward<V0>(v0)),
second(this, std::forward<Vs>(vs)...) {
// nop
}
template <class V0, class V1, class... Vs>
pair_storage(std::integral_constant<size_t, 2>, V0&& v0, V1&& v1, Vs&&... vs)
: first(this, std::forward<V0>(v0), std::forward<V1>(v1)),
second(this, std::forward<Vs>(vs)...) {
// nop
}
~pair_storage() {
// nop
}
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_PAIR_STORAGE_HPP
......@@ -54,7 +54,14 @@ class proper_actor_base : public Policies::resume_policy::template
void enqueue(const actor_addr& sender, message_id mid,
message msg, execution_unit* eu) override {
scheduling_policy().enqueue(dptr(), sender, mid, msg, eu);
auto d = dptr();
scheduling_policy().enqueue(d, d->new_mailbox_element(sender, mid,
std::move(msg)),
eu);
}
void enqueue(mailbox_element_ptr what, execution_unit* eu) override {
scheduling_policy().enqueue(dptr(), std::move(what), eu);
}
void launch(bool hide, bool lazy, execution_unit* eu) {
......
......@@ -143,35 +143,33 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
**************************************************************************/
/**
* Sends `{what...} to `whom` using the priority `prio`.
* Sends `{vs...} to `whom` using the priority `prio`.
*/
template <class... Ts>
inline void send(message_priority prio, const channel& whom, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "sizeof...(Ts) == 0");
send_impl(prio, whom, make_message(std::forward<Ts>(what)...));
template <class... Vs>
inline void send(message_priority prio, const channel& whom, Vs&&... vs) {
static_assert(sizeof...(Vs) > 0, "sizeof...(Vs) == 0");
fast_send(prio, whom, std::forward<Vs>(vs)...);
}
/**
* Sends `{what...} to `whom`.
* Sends `{vs...} to `whom` using normal priority.
*/
template <class... Ts>
inline void send(const channel& whom, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "sizeof...(Ts) == 0");
send_impl(message_priority::normal, whom,
make_message(std::forward<Ts>(what)...));
template <class... Vs>
inline void send(const channel& whom, Vs&&... vs) {
static_assert(sizeof...(Vs) > 0, "sizeof...(Vs) == 0");
fast_send(message_priority::normal, whom, std::forward<Vs>(vs)...);
}
/**
* Sends `{what...} to `whom`.
*/
template <class... Rs, class... Ts>
void send(const typed_actor<Rs...>& whom, Ts... what) {
template <class... Rs, class... Vs>
void send(const typed_actor<Rs...>& whom, Vs&&... vs) {
check_typed_input(whom,
detail::type_list<typename detail::implicit_conversions<
typename std::decay<Ts>::type
typename std::decay<Vs>::type
>::type...>{});
send_impl(message_priority::normal, actor{whom.m_ptr.get()},
make_message(std::forward<Ts>(what)...));
fast_send(message_priority::normal, whom, std::forward<Vs>(vs)...);
}
/**
......@@ -530,7 +528,7 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
template <class... Ts>
inline mailbox_element_ptr new_mailbox_element(Ts&&... args) {
return mailbox_element::create(std::forward<Ts>(args)...);
return mailbox_element::make(std::forward<Ts>(args)...);
}
protected:
......@@ -553,7 +551,37 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
/** @endcond */
private:
void send_impl(message_priority prio, const channel& dest, message&& what);
template <class... Vs>
void fast_send_impl(message_priority mp, abstract_channel* dest, Vs&&... vs) {
if (!dest) {
return;
}
dest->enqueue(mailbox_element::make_joint(address(), message_id::make(mp),
std::forward<Vs>(vs)...),
host());
}
template <class T, class V0, class... Vs>
typename std::enable_if<
!std::is_same<typename std::decay<V0>::type, message>::value
>::type
fast_send(message_priority mp, const T& dest, V0&& v0, Vs&&... vs) {
fast_send_impl(mp, actor_cast<abstract_channel*>(dest),
std::forward<V0>(v0), std::forward<Vs>(vs)...);
}
template <class T>
void fast_send(message_priority mp, const T& dest, message what) {
send_impl(mp, actor_cast<abstract_channel*>(dest), std::move(what));
}
void send_impl(message_priority prio, abstract_channel* dest, message&& what);
template <class T>
void send_impl(message_priority prio, const T& dest, message&& what) {
send_impl(prio, actor_cast<abstract_channel*>(dest), std::move(what));
}
void delayed_send_impl(message_priority prio, const channel& whom,
const duration& rtime, message data);
using super = combined_type;
......
......@@ -17,10 +17,10 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_RECURSIVE_QUEUE_NODE_HPP
#define CAF_RECURSIVE_QUEUE_NODE_HPP
#ifndef CAF_MAILBOX_ELEMENT_HPP
#define CAF_MAILBOX_ELEMENT_HPP
#include <cstdint>
#include <cstddef>
#include "caf/extend.hpp"
#include "caf/message.hpp"
......@@ -30,7 +30,11 @@
#include "caf/mixin/memory_cached.hpp"
#include "caf/detail/embedded.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/tuple_vals.hpp"
#include "caf/detail/pair_storage.hpp"
#include "caf/detail/message_data.hpp"
namespace caf {
......@@ -57,7 +61,23 @@ class mailbox_element : public extend<memory_managed>::
using unique_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
static unique_ptr create(actor_addr sender, message_id id, message msg);
static unique_ptr make(actor_addr sender, message_id id, message msg);
template <class... Vs>
static unique_ptr make_joint(actor_addr sender, message_id id, Vs&&... vs) {
using value_storage =
detail::tuple_vals<
typename unbox_message_element<
typename detail::strip_and_convert<Vs>::type
>::type...
>;
std::integral_constant<size_t, 2> tk;
using storage = detail::pair_storage<mailbox_element, value_storage>;
auto ptr = detail::memory::create<storage>(tk, std::move(sender), id,
std::forward<Vs>(vs)...);
ptr->first.msg.reset(&(ptr->second));
return unique_ptr{&(ptr->first)};
}
inline bool is_high_priority() const {
return mid.is_high_priority();
......@@ -68,4 +88,4 @@ using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
} // namespace caf
#endif // CAF_RECURSIVE_QUEUE_NODE_HPP
#endif // CAF_MAILBOX_ELEMENT_HPP
......@@ -298,7 +298,7 @@ class message {
m_vals.detach();
}
void reset(data_ptr new_ptr = nullptr);
void reset(raw_ptr new_ptr = nullptr);
explicit message(raw_ptr);
......
......@@ -23,6 +23,7 @@
#include <cstdint>
#include "caf/config.hpp"
#include "caf/message_priority.hpp"
#include "caf/detail/comparable.hpp"
namespace caf {
......@@ -114,6 +115,11 @@ class message_id : detail::comparable<message_id> {
return result;
}
static inline message_id make(message_priority prio) {
message_id res;
return prio == message_priority::high ? res.with_high_priority() : res;
}
long compare(const message_id& other) const {
return (m_value == other.m_value) ? 0
: (m_value < other.m_value ? -1 : 1);
......
......@@ -27,7 +27,6 @@ namespace caf {
enum class message_priority : uint32_t {
normal,
high
};
} // namespace caf
......
......@@ -53,9 +53,9 @@ class cooperative_scheduling {
}
template <class Actor>
void enqueue(Actor* self, const actor_addr& sender, message_id mid,
message& msg, execution_unit* eu) {
auto ptr = self->new_mailbox_element(sender, mid, std::move(msg));
void enqueue(Actor* self, mailbox_element_ptr ptr, execution_unit* eu) {
auto mid = ptr->mid;
auto sender = ptr->sender;
switch (self->mailbox().enqueue(ptr.release())) {
case detail::enqueue_result::unblocked_reader: {
// re-schedule actor
......
......@@ -52,9 +52,9 @@ class no_scheduling {
using timeout_type = std::chrono::high_resolution_clock::time_point;
template <class Actor>
void enqueue(Actor* self, const actor_addr& sender, message_id mid,
message& msg, execution_unit*) {
auto ptr = self->new_mailbox_element(sender, mid, std::move(msg));
void enqueue(Actor* self, mailbox_element_ptr ptr, execution_unit*) {
auto mid = ptr->mid;
auto sender = ptr->sender;
// returns false if mailbox has been closed
if (!self->mailbox().synchronized_enqueue(m_mtx, m_cv, ptr.release())) {
if (mid.is_request()) {
......
......@@ -86,12 +86,11 @@ class scheduling_policy {
timed_fetch_result fetch_messages(Actor* self, F cb, timeout_type abs_time);
/**
* Enqueues given message to the actor's mailbox and take any
* steps to resume the actor if it's currently blocked.
* Enqueues given message to the actor's mailbox and takes any
* steps necessary to resume the actor if it's currently blocked.
*/
template <class Actor>
void enqueue(Actor* self, const actor_addr& sender, message_id mid,
message& msg, execution_unit* host);
void enqueue(Actor* self, mailbox_element_ptr ptr, execution_unit* host);
/**
* Starts the given actor either by launching a thread or enqueuing
......
......@@ -18,6 +18,8 @@
******************************************************************************/
#include "caf/abstract_channel.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/detail/singletons.hpp"
namespace caf {
......@@ -44,6 +46,10 @@ abstract_channel::~abstract_channel() {
// nop
}
void abstract_channel::enqueue(mailbox_element_ptr what, execution_unit* host) {
enqueue(what->sender, what->mid, what->msg, host);
}
bool abstract_channel::is_remote() const {
return m_node != singletons::get_node_id();
}
......
......@@ -126,16 +126,12 @@ void local_actor::forward_message(const actor& dest, message_priority prio) {
m_current_node->mid = invalid_message_id;
}
void local_actor::send_impl(message_priority prio, const channel& dest,
void local_actor::send_impl(message_priority prio, abstract_channel* dest,
message&& what) {
if (!dest) {
return;
}
message_id mid;
if (prio == message_priority::high) {
mid = mid.with_high_priority();
}
dest->enqueue(address(), mid, std::move(what), host());
dest->enqueue(address(), message_id::make(prio), std::move(what), host());
}
void local_actor::send_exit(const actor_addr& whom, uint32_t reason) {
......
......@@ -51,11 +51,24 @@ mailbox_element::~mailbox_element() {
// nop
}
mailbox_element_ptr mailbox_element::create(actor_addr sender, message_id id,
mailbox_element_ptr mailbox_element::make(actor_addr sender, message_id id,
message msg) {
auto ptr = detail::memory::create<mailbox_element>(std::move(sender), id,
std::move(msg));
return mailbox_element_ptr{ptr};
}
/*
mailbox_element::joint::joint(ref_counted* v0, actor_addr&& v1, message_id v2)
: embedded<mailbox_element>(v0, std::move(v1), v2) {
// nop
}
void mailbox_element::joint::request_deletion() {
sender = invalid_actor_addr;
msg.reset();
m_storage->deref();
}
*/
} // namespace caf
......@@ -46,8 +46,8 @@ message& message::operator=(message&& other) {
return *this;
}
void message::reset(data_ptr new_ptr) {
m_vals.swap(new_ptr);
void message::reset(raw_ptr new_ptr) {
m_vals.reset(new_ptr);
}
void* message::mutable_at(size_t p) {
......
......@@ -187,7 +187,7 @@ void broker::invoke_message(const actor_addr& sender, message_id mid,
break;
case policy::im_skipped: {
CAF_LOG_DEBUG("handle_message returned hm_skip_msg or hm_cache_msg");
auto ptr = mailbox_element::create(sender, bid,
auto ptr = mailbox_element::make(sender, bid,
std::move(m_dummy_node.msg));
m_cache.push_back(std::move(ptr));
break;
......
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