Commit 0f3d58c0 authored by Dominik Charousset's avatar Dominik Charousset

Add before_processing overload for responses

parent 24a13d0e
......@@ -22,6 +22,7 @@
#include "caf/config.hpp"
#include "caf/detail/build_config.hpp"
#include "caf/fwd.hpp"
#include "caf/timestamp.hpp"
namespace caf {
......@@ -47,7 +48,8 @@ public:
/// @thread-safe
virtual void remove_actor(const local_actor& self) = 0;
/// Called whenever an actor is about to process an element from its mailbox.
/// Called whenever an actor is about to process an asynchronous message from
/// its mailbox.
/// @param self The current actor.
/// @param element The current element from the mailbox.
/// @thread-safe
......@@ -55,6 +57,18 @@ public:
const mailbox_element& element)
= 0;
/// Called whenever an actor is about to process a response message from its
/// mailbox.
/// @param self The current actor.
/// @param element The current element from the mailbox.
/// @param request_id Message ID of the original request.
/// @param send_time The timestamp .
/// @note The default implementation calls `before_processing(self, element)`.
/// @thread-safe
virtual void
before_processing(const local_actor& self, const mailbox_element& element,
message_id request_id, timestamp send_time);
/// Called after an actor processed an element from its mailbox.
/// @param self The current actor.
/// @param result Stores whether the actor consumed, skipped or dropped the
......@@ -93,8 +107,8 @@ public:
};
#ifdef CAF_ENABLE_ACTOR_PROFILER
# define CAF_BEFORE_PROCESSING(self, msg) \
self->system().profiler_before_processing(*self, msg)
# define CAF_BEFORE_PROCESSING(self, ...) \
self->system().profiler_before_processing(*self, __VA_ARGS__)
# define CAF_AFTER_PROCESSING(self, result) \
self->system().profiler_after_processing(*self, result)
# define CAF_BEFORE_SENDING(self, msg) \
......@@ -102,7 +116,7 @@ public:
# define CAF_BEFORE_SENDING_SCHEDULED(self, timeout, msg) \
self->system().profiler_before_sending_scheduled(*self, timeout, msg)
#else
# define CAF_BEFORE_PROCESSING(self, msg) static_cast<void>(0)
# define CAF_BEFORE_PROCESSING(self, ...) static_cast<void>(0)
# define CAF_AFTER_PROCESSING(self, result) static_cast<void>(0)
# define CAF_BEFORE_SENDING(self, msg) static_cast<void>(0)
# define CAF_BEFORE_SENDING_SCHEDULED(self, timeout, msg) static_cast<void>(0)
......
......@@ -188,8 +188,17 @@ public:
/// A queue optimized for single-reader-many-writers.
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
/// Stores state for handling an outstanding response message.
struct response_handler {
/// Stores the callbacks for processing the response.
behavior bhvr;
/// Stores when the request was sent.
timestamp send_time;
};
/// The message ID of an outstanding response with its callback.
using pending_response = std::pair<const message_id, behavior>;
using pending_response = std::pair<const message_id, response_handler>;
/// A pointer to a scheduled actor.
using pointer = scheduled_actor*;
......@@ -688,7 +697,7 @@ public:
}
inline behavior& current_behavior() {
return !awaited_responses_.empty() ? awaited_responses_.front().second
return !awaited_responses_.empty() ? awaited_responses_.front().second.bhvr
: bhvr_stack_.back();
}
......@@ -896,7 +905,8 @@ protected:
std::forward_list<pending_response> awaited_responses_;
/// Stores callbacks for multiplexed responses.
detail::unordered_flat_map<message_id, behavior> multiplexed_responses_;
detail::unordered_flat_map<message_id, response_handler>
multiplexed_responses_;
/// Customization point for setting a default `message` callback.
default_handler default_handler_;
......
......@@ -18,10 +18,18 @@
#include "caf/actor_profiler.hpp"
#include "caf/message_id.hpp"
namespace caf {
actor_profiler::~actor_profiler() {
// nop
}
void actor_profiler::before_processing(const local_actor& self,
const mailbox_element& element,
message_id, timestamp) {
before_processing(self, element);
}
} // namespace caf
......@@ -30,56 +30,56 @@ raw_event_based_actor::raw_event_based_actor(actor_config& cfg)
invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x));
current_element_ = &x;
auto invoke = [](behavior& f, mailbox_element& in) -> bool {
return f(in.content()) != none;
};
// Short-circuit awaited responses. Don't trigger any logging if we skip here.
if (!awaited_responses_.empty()) {
auto& pr = awaited_responses_.front();
// Skip all messages until we receive the currently awaited response.
if (x.mid != pr.first)
return invoke_message_result::skipped;
CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, x, pr.first, pr.second.send_time);
auto f = std::move(pr.second.bhvr);
awaited_responses_.pop_front();
if (!invoke(f, x)) {
// Try again with error if first attempt failed.
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
f(msg);
}
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(invoke_message_result::consumed);
return invoke_message_result::consumed;
}
// Handle multiplexed responses.
if (x.mid.is_response()) {
auto mrh = multiplexed_responses_.find(x.mid);
// Neither awaited nor multiplexed, probably an expired timeout. Discard
// without bothering the profiler.
if (mrh == multiplexed_responses_.end()) {
CAF_LOG_DEBUG("drop unexpected response:" << x);
return invoke_message_result::dropped;
}
CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, x, mrh->first, mrth->second.send_time);
auto bhvr = std::move(mrh->second.bhvr);
multiplexed_responses_.erase(mrh);
if (!invoke(bhvr, x)) {
// Try again with error if first attempt failed.
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
bhvr(msg);
}
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(invoke_message_result::consumed);
return invoke_message_result::consumed;
}
CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, x);
// Wrap the actual body for the function.
auto body = [this, &x] {
// short-circuit awaited responses
if (!awaited_responses_.empty()) {
auto& pr = awaited_responses_.front();
// skip all messages until we receive the currently awaited response
if (x.mid != pr.first)
return invoke_message_result::skipped;
if (!pr.second(x.content())) {
// try again with error if first attempt failed
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
pr.second(msg);
}
awaited_responses_.pop_front();
return invoke_message_result::consumed;
}
// handle multiplexed responses
if (x.mid.is_response()) {
auto mrh = multiplexed_responses_.find(x.mid);
// neither awaited nor multiplexed, probably an expired timeout
if (mrh == multiplexed_responses_.end())
return invoke_message_result::dropped;
if (!mrh->second(x.content())) {
// try again with error if first attempt failed
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
mrh->second(msg);
}
multiplexed_responses_.erase(mrh);
return invoke_message_result::consumed;
}
auto& content = x.content();
// handle timeout messages
if (x.content().type_token() == make_type_token<timeout_msg>()) {
auto& tm = content.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
CAF_ASSERT(x.mid.is_async());
if (is_active_receive_timeout(tid)) {
CAF_LOG_DEBUG("handle timeout message");
if (bhvr_stack_.empty())
return invoke_message_result::dropped;
bhvr_stack_.back().handle_timeout();
return invoke_message_result::consumed;
}
CAF_LOG_DEBUG("dropped expired timeout message");
return invoke_message_result::dropped;
}
// handle everything else as ordinary message
detail::default_invoke_result_visitor<event_based_actor> visitor{this};
bool skipped = false;
......
......@@ -534,14 +534,16 @@ void scheduled_actor::add_awaited_response_handler(message_id response_id,
behavior bhvr) {
if (bhvr.timeout().valid())
request_response_timeout(bhvr.timeout(), response_id);
awaited_responses_.emplace_front(response_id, std::move(bhvr));
awaited_responses_.emplace_front(
response_id, response_handler{std::move(bhvr), make_timestamp()});
}
void scheduled_actor::add_multiplexed_response_handler(message_id response_id,
behavior bhvr) {
if (bhvr.timeout().valid())
request_response_timeout(bhvr.timeout(), response_id);
multiplexed_responses_.emplace(response_id, std::move(bhvr));
multiplexed_responses_.emplace(
response_id, response_handler{std::move(bhvr), make_timestamp()});
}
scheduled_actor::message_category
......@@ -630,51 +632,56 @@ scheduled_actor::categorize(mailbox_element& x) {
invoke_message_result scheduled_actor::consume(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x));
current_element_ = &x;
auto invoke = [](behavior& f, mailbox_element& in) -> bool {
return f(in.content()) != none;
};
// Short-circuit awaited responses. Don't trigger any logging if we skip here.
if (!awaited_responses_.empty()) {
auto& pr = awaited_responses_.front();
// Skip all messages until we receive the currently awaited response.
if (x.mid != pr.first)
return invoke_message_result::skipped;
CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, x, pr.first, pr.second.send_time);
auto f = std::move(pr.second.bhvr);
awaited_responses_.pop_front();
if (!invoke(f, x)) {
// Try again with error if first attempt failed.
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
f(msg);
}
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(invoke_message_result::consumed);
return invoke_message_result::consumed;
}
// Handle multiplexed responses.
if (x.mid.is_response()) {
auto mrh = multiplexed_responses_.find(x.mid);
// Neither awaited nor multiplexed, probably an expired timeout. Discard
// without bothering the profiler.
if (mrh == multiplexed_responses_.end()) {
CAF_LOG_DEBUG("drop unexpected response:" << x);
return invoke_message_result::dropped;
}
CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, x, mrh->first, mrth->second.send_time);
auto bhvr = std::move(mrh->second.bhvr);
multiplexed_responses_.erase(mrh);
if (!invoke(bhvr, x)) {
// Try again with error if first attempt failed.
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
bhvr(msg);
}
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(invoke_message_result::consumed);
return invoke_message_result::consumed;
}
CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, x);
// Wrap the actual body for the function.
auto body = [this, &x] {
// Helper function for dispatching a message to a response handler.
using ptr_t = scheduled_actor*;
using fun_t = bool (*)(ptr_t, behavior&, mailbox_element&);
auto ordinary_invoke = [](ptr_t, behavior& f, mailbox_element& in) -> bool {
return f(in.content()) != none;
};
auto select_invoke_fun = [&]() -> fun_t { return ordinary_invoke; };
// Short-circuit awaited responses.
if (!awaited_responses_.empty()) {
auto invoke = select_invoke_fun();
auto& pr = awaited_responses_.front();
// skip all messages until we receive the currently awaited response
if (x.mid != pr.first)
return invoke_message_result::skipped;
auto f = std::move(pr.second);
awaited_responses_.pop_front();
if (!invoke(this, f, x)) {
// try again with error if first attempt failed
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
f(msg);
}
return invoke_message_result::consumed;
}
// Handle multiplexed responses.
if (x.mid.is_response()) {
auto invoke = select_invoke_fun();
auto mrh = multiplexed_responses_.find(x.mid);
// neither awaited nor multiplexed, probably an expired timeout
if (mrh == multiplexed_responses_.end())
return invoke_message_result::dropped;
auto bhvr = std::move(mrh->second);
multiplexed_responses_.erase(mrh);
if (!invoke(this, bhvr, x)) {
// try again with error if first attempt failed
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
bhvr(msg);
}
return invoke_message_result::consumed;
}
// Dispatch on the content of x.
switch (categorize(x)) {
case message_category::skipped:
......
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