Commit 1fa62d11 authored by Dominik Charousset's avatar Dominik Charousset

Optimize forwarding of messages

When forwarding a message, an actor transfers ownership of its current
`mailbox_element` to the receiver. Combined with the previous optimization,
this means that there is only a single memory allocation for a message from
source to destination.
parent 5c7af235
......@@ -23,6 +23,8 @@
#include <memory>
#include <iterator>
#include "caf/policy/invoke_policy.hpp"
namespace caf {
namespace detail {
......@@ -173,6 +175,41 @@ class intrusive_partitioned_list {
insert(second_end(), val);
}
template <class Actor, class... Vs>
bool invoke(Actor* self, iterator first, iterator last, Vs&... vs) {
pointer prev = first->prev;
pointer next = first->next;
auto patch = [&] {
prev->next = next;
next->prev = prev;
};
while (first != last) {
std::unique_ptr<value_type, deleter_type> tmp{first.ptr};
switch (self->invoke_message(tmp, vs...)) {
case policy::im_dropped:
patch();
first = next;
next = first->next;
break;
case policy::im_success:
patch();
return true;
default:
if (tmp) {
prev = tmp.release();
first = next;
next = first->next;
} else {
patch();
first = next;
next = first->next;
}
break;
}
}
return false;
}
size_t count(iterator first, iterator last, size_t max_count) {
size_t result = 0;
while (first != last && result < max_count) {
......
......@@ -106,7 +106,7 @@ class proper_actor_base : public Policies::resume_policy::template
// member functions from invoke policy
template <class PartialFunctionOrBehavior>
policy::invoke_message_result invoke_message(mailbox_element& me,
policy::invoke_message_result invoke_message(mailbox_element_ptr& me,
PartialFunctionOrBehavior& fun,
message_id awaited_response) {
return invoke_policy().invoke_message(dptr(), me, fun, awaited_response);
......@@ -157,7 +157,7 @@ class proper_actor
// required by event_based_resume::mixin::resume
policy::invoke_message_result invoke_message(mailbox_element& me) {
policy::invoke_message_result invoke_message(mailbox_element_ptr& me) {
CAF_LOG_TRACE("");
auto bhvr = this->bhvr_stack().back();
auto mid = this->bhvr_stack().back_id();
......@@ -221,11 +221,13 @@ class proper_actor<Base, Policies, true>
for (;;) {
this->await_ready();
auto msg = this->next_message();
switch (this->invoke_message(*msg, bhvr, mid)) {
switch (this->invoke_message(msg, bhvr, mid)) {
case policy::im_success:
return;
case policy::im_skipped:
if (msg) {
this->push_to_cache(std::move(msg));
}
break;
default:
// delete msg
......
......@@ -296,16 +296,12 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
* Returns the last message that was dequeued from the actor's mailbox.
* @warning Only set during callback invocation.
*/
inline message& last_dequeued() {
return m_current_node->msg;
}
message& last_dequeued();
/**
* Returns the address of the last sender of the last dequeued message.
*/
inline actor_addr& last_sender() {
return m_current_node->sender;
}
actor_addr& last_sender();
/**
* Adds a unidirectional `monitor` to `whom`.
......@@ -442,12 +438,8 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
return res;
}
inline void current_node(mailbox_element* ptr) {
this->m_current_node = ptr;
}
inline mailbox_element* current_node() {
return this->m_current_node;
inline mailbox_element_ptr& current_element() {
return m_current_element;
}
inline message_id new_request_id() {
......@@ -494,7 +486,7 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
// returns 0 if last_dequeued() is an asynchronous or sync request message,
// a response id generated from the request id otherwise
inline message_id get_response_id() {
auto mid = m_current_node->mid;
auto mid = m_current_element->mid;
return (mid.is_request()) ? mid.response_id() : message_id();
}
......@@ -522,10 +514,6 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
void cleanup(uint32_t reason);
inline mailbox_element* dummy_node() {
return &m_dummy_node;
}
template <class... Ts>
inline mailbox_element_ptr new_mailbox_element(Ts&&... args) {
return mailbox_element::make(std::forward<Ts>(args)...);
......@@ -538,12 +526,13 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
// identifies all IDs of sync messages waiting for a response
std::forward_list<message_id> m_pending_responses;
// "default value" for m_current_node
mailbox_element m_dummy_node;
// "default value" for m_current_element
actor_addr m_dummy_sender;
message m_dummy_message;
// points to m_dummy_node if no callback is currently invoked,
// points to the node under processing otherwise
mailbox_element* m_current_node;
mailbox_element_ptr m_current_element;
// set by quit
uint32_t m_planned_exit_reason;
......
......@@ -300,6 +300,8 @@ class message {
void reset(raw_ptr new_ptr = nullptr);
void swap(message& other);
explicit message(raw_ptr);
inline std::string tuple_type_names() const {
......
......@@ -76,7 +76,7 @@ class actor_widget : public Base {
auto ptr = dynamic_cast<event_type*>(event);
if (ptr) {
m_invoke.invoke_message(m_companion.get(),
*ptr->mptr.get(),
ptr->mptr,
m_companion->bhvr_stack().back(),
m_companion->bhvr_stack().back_id());
return true;
......
......@@ -125,7 +125,7 @@ class event_based_resume {
for (size_t i = 0; i < max_throughput; ++i) {
auto ptr = d->next_message();
if (ptr) {
switch (d->invoke_message(*ptr)) {
switch (d->invoke_message(ptr)) {
case policy::im_success:
d->bhvr_stack().cleanup();
++handled_msgs;
......@@ -144,6 +144,7 @@ class event_based_resume {
}
break;
case policy::im_skipped:
CAF_REQUIRE(ptr != nullptr);
d->push_to_cache(std::move(ptr));
break;
case policy::im_dropped:
......
......@@ -86,14 +86,14 @@ class invoke_policy {
// - yes: cleanup()
// - no: revert(...) -> set self back to state it had before begin()
template <class Actor, class Fun>
invoke_message_result invoke_message(Actor* self, mailbox_element& node,
invoke_message_result invoke_message(Actor* self, mailbox_element_ptr& node,
Fun& fun, message_id awaited_response) {
CAF_LOG_TRACE("");
bool handle_sync_failure_on_mismatch = true;
if (dptr()->hm_should_skip(node)) {
if (dptr()->hm_should_skip(*node)) {
return im_skipped;
}
switch (this->filter_msg(self, node)) {
switch (this->filter_msg(self, *node)) {
case msg_type::normal_exit:
CAF_LOG_DEBUG("dropped normal exit signal");
return im_dropped;
......@@ -113,7 +113,7 @@ class invoke_policy {
return im_success;
case msg_type::timeout: {
CAF_LOG_DEBUG("handle timeout message");
auto& tm = node.msg.get_as<timeout_msg>(0);
auto& tm = node->msg.get_as<timeout_msg>(0);
self->handle_timeout(fun, tm.timeout_id);
if (awaited_response.valid()) {
self->mark_arrived(awaited_response);
......@@ -126,12 +126,12 @@ class invoke_policy {
CAF_ANNOTATE_FALLTHROUGH;
case msg_type::sync_response:
CAF_LOG_DEBUG("handle as synchronous response: "
<< CAF_TARG(node.msg, to_string) << ", "
<< CAF_MARG(node.mid, integer_value) << ", "
<< CAF_TARG(node->msg, to_string) << ", "
<< CAF_MARG(node->mid, integer_value) << ", "
<< CAF_MARG(awaited_response, integer_value));
if (awaited_response.valid() && node.mid == awaited_response) {
auto previous_node = dptr()->hm_begin(self, node);
auto res = invoke_fun(self, node.msg, node.mid, fun);
if (awaited_response.valid() && node->mid == awaited_response) {
dptr()->hm_begin(self, node);
auto res = invoke_fun(self, fun);
if (!res && handle_sync_failure_on_mismatch) {
CAF_LOG_WARNING("sync failure occured in actor "
<< "with ID " << self->id());
......@@ -139,20 +139,19 @@ class invoke_policy {
}
self->mark_arrived(awaited_response);
self->remove_handler(awaited_response);
dptr()->hm_cleanup(self, previous_node);
dptr()->hm_cleanup(self, node);
return im_success;
}
return im_skipped;
case msg_type::ordinary:
if (!awaited_response.valid()) {
auto previous_node = dptr()->hm_begin(self, node);
auto res = invoke_fun(self, node.msg, node.mid, fun);
dptr()->hm_begin(self, node);
auto res = invoke_fun(self, fun);
dptr()->hm_cleanup(self, node);
if (res) {
dptr()->hm_cleanup(self, previous_node);
return im_success;
}
// no match (restore self members)
dptr()->hm_revert(self, previous_node);
}
CAF_LOG_DEBUG_IF(awaited_response.valid(),
"ignored message; await response: "
......@@ -180,14 +179,14 @@ class invoke_policy {
// - extracts response message from handler
// - returns true if fun was successfully invoked
template <class Actor, class Fun, class MaybeResponseHdl = int>
optional<message> invoke_fun(Actor* self, message& msg, message_id& mid,
Fun& fun,
optional<message> invoke_fun(Actor* self, Fun& fun,
MaybeResponseHdl hdl = MaybeResponseHdl{}) {
# ifdef CAF_LOG_LEVEL
auto msg_str = to_string(msg);
# endif
CAF_LOG_TRACE(CAF_MARG(mid, integer_value) << ", msg = " << msg_str);
auto res = fun(msg); // might change mid
auto mid = self->current_element()->mid;
auto res = fun(self->current_element()->msg);
CAF_LOG_DEBUG_IF(res, "actor did consume message: " << msg_str);
CAF_LOG_DEBUG_IF(!res, "actor did ignore message: " << msg_str);
if (!res) {
......
......@@ -42,25 +42,18 @@ class nestable_invoke : public invoke_policy<nestable_invoke> {
}
template <class Actor>
mailbox_element* hm_begin(Actor* self, mailbox_element& node) {
auto previous = self->current_node();
self->current_node(&node);
self->push_timeout();
node.marked = true;
return previous;
void hm_begin(Actor* self, mailbox_element_ptr& node) {
node->marked = true;
node.swap(self->current_element());
}
template <class Actor>
void hm_cleanup(Actor* self, mailbox_element* previous) {
self->current_node()->marked = false;
self->current_node(previous);
self->pop_timeout();
void hm_cleanup(Actor* self, mailbox_element_ptr& node) {
auto& ref = self->current_element();
if (ref) {
ref->marked = false;
}
template <class Actor>
void hm_revert(Actor* self, mailbox_element* previous) {
// same operation for blocking, i.e., nestable, invoke
hm_cleanup(self, previous);
ref.swap(node);
}
};
......
......@@ -48,26 +48,13 @@ class not_prioritizing {
self->mailbox().cache().push_second_back(ptr.release());
}
template <class Actor, class... Ts>
bool invoke_from_cache(Actor* self, Ts&... args) {
template <class Actor, class... Vs>
bool invoke_from_cache(Actor* self, Vs&... args) {
auto& cache = self->mailbox().cache();
auto i = cache.second_begin();
auto e = cache.second_end();
CAF_LOG_DEBUG(std::distance(i, e) << " elements in cache");
while (i != e) {
switch (self->invoke_message(*i, args...)) {
case im_dropped:
i = cache.erase(i);
break;
case im_success:
cache.erase(i);
return true;
default:
++i;
break;
}
}
return false;
return cache.invoke(self, i, e, args...);
}
};
......
......@@ -39,19 +39,13 @@ class sequential_invoke : public invoke_policy<sequential_invoke> {
}
template <class Actor>
mailbox_element* hm_begin(Actor* self, mailbox_element& node) {
self->current_node(&node);
return nullptr;
void hm_begin(Actor* self, mailbox_element_ptr& node) {
node.swap(self->current_element());
}
template <class Actor>
void hm_cleanup(Actor* self, mailbox_element*) {
self->current_node(self->dummy_node());
}
template <class Actor>
void hm_revert(Actor* self, mailbox_element*) {
self->current_node(self->dummy_node());
void hm_cleanup(Actor* self, mailbox_element_ptr& node) {
node.swap(self->current_element());
}
};
......
......@@ -33,8 +33,6 @@ namespace caf {
// e.g., when calling address() in the ctor of a derived class
local_actor::local_actor()
: super(size_t{1}),
m_dummy_node(),
m_current_node(&m_dummy_node),
m_planned_exit_reason(exit_reason::not_exited) {
// nop
}
......@@ -43,6 +41,14 @@ local_actor::~local_actor() {
// nop
}
message& local_actor::last_dequeued() {
return m_current_element ? m_current_element->msg : m_dummy_message;
}
actor_addr& local_actor::last_sender() {
return m_current_element ? m_current_element->sender : m_dummy_sender;
}
void local_actor::monitor(const actor_addr& whom) {
if (whom == invalid_actor_addr) {
return;
......@@ -100,11 +106,11 @@ std::vector<group> local_actor::joined_groups() const {
}
void local_actor::reply_message(message&& what) {
auto& whom = m_current_node->sender;
auto& whom = m_current_element->sender;
if (!whom) {
return;
}
auto& mid = m_current_node->mid;
auto& mid = m_current_element->mid;
if (mid.valid() == false || mid.is_response()) {
send(actor_cast<channel>(whom), std::move(what));
} else if (!mid.is_answered()) {
......@@ -118,12 +124,11 @@ void local_actor::forward_message(const actor& dest, message_priority prio) {
if (!dest) {
return;
}
auto mid = (prio == message_priority::high)
? m_current_node->mid.with_high_priority()
: m_current_node->mid.with_normal_priority();
dest->enqueue(m_current_node->sender, mid, m_current_node->msg, host());
// treat this message as asynchronous message from now on
m_current_node->mid = invalid_message_id;
auto mid = m_current_element->mid;
m_current_element->mid = prio == message_priority::high
? mid.with_high_priority()
: mid.with_normal_priority();
dest->enqueue(std::move(m_current_element), host());
}
void local_actor::send_impl(message_priority prio, abstract_channel* dest,
......@@ -149,9 +154,12 @@ void local_actor::delayed_send_impl(message_priority prio, const channel& dest,
}
response_promise local_actor::make_response_promise() {
auto n = m_current_node;
response_promise result{address(), n->sender, n->mid.response_id()};
n->mid.mark_as_answered();
auto& ptr = m_current_element;
if (!ptr) {
return response_promise{};
}
response_promise result{address(), ptr->sender, ptr->mid.response_id()};
ptr->mid.mark_as_answered();
return result;
}
......
......@@ -50,6 +50,10 @@ void message::reset(raw_ptr new_ptr) {
m_vals.reset(new_ptr);
}
void message::swap(message& other) {
m_vals.swap(other.m_vals);
}
void* message::mutable_at(size_t p) {
CAF_REQUIRE(m_vals);
return m_vals->mutable_at(p);
......
......@@ -30,9 +30,10 @@
#include "caf/mixin/functor_based.hpp"
#include "caf/mixin/behavior_stack_based.hpp"
#include "caf/policy/not_prioritizing.hpp"
#include "caf/policy/sequential_invoke.hpp"
#include "caf/detail/intrusive_partitioned_list.hpp"
#include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/receive_policy.hpp"
......@@ -274,10 +275,18 @@ class broker : public extend<local_actor>::
accept_handle add_tcp_doorman(network::native_socket fd);
policy::invoke_message_result invoke_message(mailbox_element_ptr& msg,
behavior& bhvr,
message_id mid);
void invoke_message(mailbox_element_ptr& msg);
void invoke_message(const actor_addr& sender, message_id mid, message& msg);
void enqueue(const actor_addr&, message_id, message,
execution_unit*) override;
void enqueue(const actor_addr&, message_id,
message, execution_unit*) override;
void enqueue(mailbox_element_ptr, execution_unit*) override;
/**
* Closes all connections and acceptors.
......@@ -361,7 +370,7 @@ class broker : public extend<local_actor>::
policy::sequential_invoke m_invoke_policy;
middleman& m_mm;
std::deque<mailbox_element_ptr> m_cache;
detail::intrusive_partitioned_list<mailbox_element, detail::disposer> m_cache;
};
class broker::functor_based : public extend<broker>::
......
......@@ -130,49 +130,47 @@ void broker::doorman::io_failure(network::operation op) {
class broker::continuation {
public:
continuation(broker_ptr ptr, actor_addr from, message_id mid, message&& msg)
: m_self(std::move(ptr)),
m_from(from),
m_mid(mid),
m_data(std::move(msg)) {
continuation(broker_ptr bptr, mailbox_element_ptr mptr)
: m_self(std::move(bptr)),
m_ptr(std::move(mptr)) {
// nop
}
inline void operator()() {
CAF_PUSH_AID(m_self->id());
CAF_LOG_TRACE("");
m_self->invoke_message(m_from, m_mid, m_data);
m_self->invoke_message(m_ptr);
}
private:
broker_ptr m_self;
actor_addr m_from;
message_id m_mid;
message m_data;
mailbox_element_ptr m_ptr;
};
void broker::invoke_message(const actor_addr& sender, message_id mid,
message& msg) {
policy::invoke_message_result broker::invoke_message(mailbox_element_ptr& msg,
behavior& bhvr,
message_id mid) {
return m_invoke_policy.invoke_message(this, msg, bhvr, mid);
}
void broker::invoke_message(mailbox_element_ptr& ptr) {
CAF_LOG_TRACE(CAF_TARG(msg, to_string));
if (planned_exit_reason() != exit_reason::not_exited
|| bhvr_stack().empty()) {
CAF_LOG_DEBUG("actor already finished execution"
<< ", planned_exit_reason = " << planned_exit_reason()
<< ", bhvr_stack().empty() = " << bhvr_stack().empty());
if (mid.valid()) {
if (ptr->mid.valid()) {
detail::sync_request_bouncer srb{exit_reason()};
srb(sender, mid);
srb(ptr->sender, ptr->mid);
}
return;
}
// prepare actor for invocation of message handler
m_dummy_node.sender = sender;
m_dummy_node.mid = mid;
std::swap(msg, m_dummy_node.msg);
try {
auto bhvr = bhvr_stack().back();
auto bid = bhvr_stack().back_id();
switch (m_invoke_policy.invoke_message(this, m_dummy_node, bhvr, bid)) {
switch (invoke_message(ptr, bhvr, bid)) {
case policy::im_success: {
CAF_LOG_DEBUG("handle_message returned hm_msg_handled");
while (!bhvr_stack().empty()
......@@ -187,9 +185,9 @@ 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::make(sender, bid,
std::move(m_dummy_node.msg));
m_cache.push_back(std::move(ptr));
if (ptr) {
m_cache.push_second_back(ptr.release());
}
break;
}
}
......@@ -205,9 +203,6 @@ void broker::invoke_message(const actor_addr& sender, message_id mid,
CAF_LOG_ERROR("broker killed due to an unknown exception");
quit(exit_reason::unhandled_exception);
}
// restore dummy node
m_dummy_node.sender = invalid_actor_addr;
std::swap(m_dummy_node.msg, msg);
// cleanup if needed
if (planned_exit_reason() != exit_reason::not_exited) {
cleanup(planned_exit_reason());
......@@ -222,27 +217,23 @@ void broker::invoke_message(const actor_addr& sender, message_id mid,
}
}
void broker::invoke_message(const actor_addr& v0, message_id v1, message& v2) {
auto ptr = mailbox_element::make(v0, v1, message{});
ptr->msg.swap(v2);
invoke_message(ptr);
if (ptr) {
ptr->msg.swap(v2);
}
}
bool broker::invoke_message_from_cache() {
CAF_LOG_TRACE("");
auto bhvr = bhvr_stack().back();
auto mid = bhvr_stack().back_id();
auto i = m_cache.begin();
auto e = m_cache.end();
auto bid = bhvr_stack().back_id();
auto i = m_cache.second_begin();
auto e = m_cache.second_end();
CAF_LOG_DEBUG(std::distance(i, e) << " elements in cache");
while (i != e) {
switch (m_invoke_policy.invoke_message(this, *(i->get()), bhvr, mid)) {
case policy::im_success:
m_cache.erase(i);
return true;
case policy::im_skipped:
++i;
break;
case policy::im_dropped:
i = m_cache.erase(i);
break;
}
}
return false;
return m_cache.invoke(this, i, e, bhvr, bid);
}
void broker::write(connection_handle hdl, size_t bs, const void* buf) {
......@@ -252,10 +243,13 @@ void broker::write(connection_handle hdl, size_t bs, const void* buf) {
out.insert(out.end(), first, last);
}
void broker::enqueue(mailbox_element_ptr ptr, execution_unit*) {
parent().backend().post(continuation{this, std::move(ptr)});
}
void broker::enqueue(const actor_addr& sender, message_id mid, message msg,
execution_unit*) {
parent().backend().post(continuation{this, sender,
mid, std::move(msg)});
execution_unit* eu) {
enqueue(mailbox_element::make(sender, mid, std::move(msg)), eu);
}
broker::broker() : m_mm(*middleman::instance()) {
......
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