Commit 656499e5 authored by Dominik Charousset's avatar Dominik Charousset

Simplify calls to CAF_AFTER_PROCESSING

parent e608ad30
...@@ -594,6 +594,14 @@ bool operator==(const logger::field& x, const logger::field& y); ...@@ -594,6 +594,14 @@ bool operator==(const logger::field& x, const logger::field& y);
#define CAF_LOG_FINALIZE_EVENT() \ #define CAF_LOG_FINALIZE_EVENT() \
CAF_LOG_IMPL(CAF_LOG_FLOW_COMPONENT, CAF_LOG_LEVEL_DEBUG, "FINALIZE") CAF_LOG_IMPL(CAF_LOG_FLOW_COMPONENT, CAF_LOG_LEVEL_DEBUG, "FINALIZE")
#define CAF_LOG_SKIP_OR_FINALIZE_EVENT(invoke_result) \
do { \
if (invoke_result == caf::invoke_message_result::skipped) \
CAF_LOG_SKIP_EVENT(); \
else \
CAF_LOG_FINALIZE_EVENT(); \
} while (false)
#define CAF_LOG_TERMINATE_EVENT(thisptr, rsn) \ #define CAF_LOG_TERMINATE_EVENT(thisptr, rsn) \
CAF_LOG_IMPL(CAF_LOG_FLOW_COMPONENT, CAF_LOG_LEVEL_DEBUG, \ CAF_LOG_IMPL(CAF_LOG_FLOW_COMPONENT, CAF_LOG_LEVEL_DEBUG, \
"TERMINATE ; ID =" << thisptr->id() \ "TERMINATE ; ID =" << thisptr->id() \
...@@ -616,6 +624,8 @@ bool operator==(const logger::field& x, const logger::field& y); ...@@ -616,6 +624,8 @@ bool operator==(const logger::field& x, const logger::field& y);
#define CAF_LOG_SKIP_EVENT() CAF_VOID_STMT #define CAF_LOG_SKIP_EVENT() CAF_VOID_STMT
#define CAF_LOG_SKIP_OR_FINALIZE_EVENT(unused) CAF_VOID_STMT
#define CAF_LOG_FINALIZE_EVENT() CAF_VOID_STMT #define CAF_LOG_FINALIZE_EVENT() CAF_VOID_STMT
#define CAF_LOG_TERMINATE_EVENT(thisptr, rsn) CAF_VOID_STMT #define CAF_LOG_TERMINATE_EVENT(thisptr, rsn) CAF_VOID_STMT
......
...@@ -157,74 +157,75 @@ void blocking_actor::fail_state(error err) { ...@@ -157,74 +157,75 @@ void blocking_actor::fail_state(error err) {
} }
intrusive::task_result intrusive::task_result
blocking_actor::mailbox_visitor:: operator()(mailbox_element& x) { blocking_actor::mailbox_visitor::operator()(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
CAF_LOG_RECEIVE_EVENT((&x)); CAF_LOG_RECEIVE_EVENT((&x));
CAF_BEFORE_PROCESSING(self, x); CAF_BEFORE_PROCESSING(self, x);
auto check_if_done = [&]() -> intrusive::task_result { // Wrap the actual body for the function.
// Stop consuming items when reaching the end of the user-defined receive auto body = [this, &x] {
// loop either via post or pre condition. auto check_if_done = [&]() -> intrusive::task_result {
if (rcc.post() && rcc.pre()) // Stop consuming items when reaching the end of the user-defined receive
return intrusive::task_result::resume; // loop either via post or pre condition.
done = true; if (rcc.post() && rcc.pre())
return intrusive::task_result::stop; return intrusive::task_result::resume;
}; done = true;
// Skip messages that don't match our message ID. return intrusive::task_result::stop;
if (mid.is_response()) { };
if (mid != x.mid) { // Skip messages that don't match our message ID.
CAF_AFTER_PROCESSING(self, invoke_message_result::skipped); if (mid.is_response()) {
CAF_LOG_SKIP_EVENT(); if (mid != x.mid) {
return intrusive::task_result::skip;
}
} else if (x.mid.is_response()) {
return intrusive::task_result::skip; return intrusive::task_result::skip;
} }
} else if (x.mid.is_response()) { // Automatically unlink from actors after receiving an exit.
CAF_AFTER_PROCESSING(self, invoke_message_result::skipped); if (x.content().match_elements<exit_msg>())
CAF_LOG_SKIP_EVENT(); self->unlink_from(x.content().get_as<exit_msg>(0).source);
return intrusive::task_result::skip; // Blocking actors can nest receives => push/pop `current_element_`
} auto prev_element = self->current_element_;
// Automatically unlink from actors after receiving an exit. self->current_element_ = &x;
if (x.content().match_elements<exit_msg>()) auto g = detail::make_scope_guard(
self->unlink_from(x.content().get_as<exit_msg>(0).source); [&] { self->current_element_ = prev_element; });
// Blocking actors can nest receives => push/pop `current_element_` // Dispatch on x.
auto prev_element = self->current_element_; detail::default_invoke_result_visitor<blocking_actor> visitor{self};
self->current_element_ = &x; switch (bhvr.nested(visitor, x.content())) {
auto g = detail::make_scope_guard([&] { default:
self->current_element_ = prev_element; return check_if_done();
}); case match_case::no_match: { // Blocking actors can have fallback handlers
// Dispatch on x. // for catch-all rules.
detail::default_invoke_result_visitor<blocking_actor> visitor{self};
switch (bhvr.nested(visitor, x.content())) {
default:
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
return check_if_done();
case match_case::no_match:
{ // Blocking actors can have fallback handlers for catch-all rules.
auto sres = bhvr.fallback(*self->current_element_); auto sres = bhvr.fallback(*self->current_element_);
if (sres.flag != rt_skip) { if (sres.flag != rt_skip) {
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
visitor.visit(sres); visitor.visit(sres);
CAF_LOG_FINALIZE_EVENT();
return check_if_done(); return check_if_done();
} }
} }
// Response handlers must get re-invoked with an error when receiving an // Response handlers must get re-invoked with an error when receiving an
// unexpected message. // unexpected message.
if (mid.is_response()) { if (mid.is_response()) {
auto err = make_error(sec::unexpected_response, auto err = make_error(sec::unexpected_response,
x.move_content_to_message()); x.move_content_to_message());
mailbox_element_view<error> tmp{std::move(x.sender), x.mid, mailbox_element_view<error> tmp{std::move(x.sender), x.mid,
std::move(x.stages), err}; std::move(x.stages), err};
self->current_element_ = &tmp; self->current_element_ = &tmp;
bhvr.nested(tmp.content()); bhvr.nested(tmp.content());
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed); return check_if_done();
CAF_LOG_FINALIZE_EVENT(); }
return check_if_done(); CAF_ANNOTATE_FALLTHROUGH;
} case match_case::skip:
CAF_ANNOTATE_FALLTHROUGH; return intrusive::task_result::skip;
case match_case::skip: }
CAF_AFTER_PROCESSING(self, invoke_message_result::skipped); };
CAF_LOG_SKIP_EVENT(); // Post-process the returned value from the function body.
return intrusive::task_result::skip; auto result = body();
if (result == intrusive::task_result::skip) {
CAF_AFTER_PROCESSING(self, invoke_message_result::skipped);
CAF_LOG_SKIP_EVENT();
} else {
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
CAF_LOG_FINALIZE_EVENT();
} }
return result;
} }
void blocking_actor::receive_impl(receive_cond& rcc, void blocking_actor::receive_impl(receive_cond& rcc,
......
...@@ -32,104 +32,100 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) { ...@@ -32,104 +32,100 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
current_element_ = &x; current_element_ = &x;
CAF_LOG_RECEIVE_EVENT(current_element_); CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, x); CAF_BEFORE_PROCESSING(this, x);
// short-circuit awaited responses // Wrap the actual body for the function.
if (!awaited_responses_.empty()) { auto body = [this, &x] {
auto& pr = awaited_responses_.front(); // short-circuit awaited responses
// skip all messages until we receive the currently awaited response if (!awaited_responses_.empty()) {
if (x.mid != pr.first) auto& pr = awaited_responses_.front();
return invoke_message_result::skipped; // skip all messages until we receive the currently awaited response
if (!pr.second(x.content())) { if (x.mid != pr.first)
// try again with error if first attempt failed return invoke_message_result::skipped;
auto msg = make_message( if (!pr.second(x.content())) {
make_error(sec::unexpected_response, x.move_content_to_message())); // try again with error if first attempt failed
pr.second(msg); auto msg = make_message(
} make_error(sec::unexpected_response, x.move_content_to_message()));
awaited_responses_.pop_front(); pr.second(msg);
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed); }
return invoke_message_result::consumed; 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()) {
CAF_AFTER_PROCESSING(this, invoke_message_result::dropped);
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); // handle multiplexed responses
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed); if (x.mid.is_response()) {
return invoke_message_result::consumed; auto mrh = multiplexed_responses_.find(x.mid);
} // neither awaited nor multiplexed, probably an expired timeout
auto& content = x.content(); if (mrh == multiplexed_responses_.end())
// 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()) {
CAF_AFTER_PROCESSING(this, invoke_message_result::dropped);
return invoke_message_result::dropped; 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);
} }
bhvr_stack_.back().handle_timeout(); multiplexed_responses_.erase(mrh);
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed);
return invoke_message_result::consumed; return invoke_message_result::consumed;
} }
CAF_LOG_DEBUG("dropped expired timeout message"); auto& content = x.content();
CAF_AFTER_PROCESSING(this, invoke_message_result::dropped); // handle timeout messages
return invoke_message_result::dropped; if (x.content().type_token() == make_type_token<timeout_msg>()) {
} auto& tm = content.get_as<timeout_msg>(0);
// handle everything else as ordinary message auto tid = tm.timeout_id;
detail::default_invoke_result_visitor<event_based_actor> visitor{this}; CAF_ASSERT(x.mid.is_async());
bool skipped = false; if (is_active_receive_timeout(tid)) {
auto had_timeout = getf(has_timeout_flag); CAF_LOG_DEBUG("handle timeout message");
if (had_timeout) if (bhvr_stack_.empty())
unsetf(has_timeout_flag); return invoke_message_result::dropped;
// restore timeout at scope exit if message was skipped bhvr_stack_.back().handle_timeout();
auto timeout_guard = detail::make_scope_guard([&] { return invoke_message_result::consumed;
if (skipped && had_timeout) }
setf(has_timeout_flag); CAF_LOG_DEBUG("dropped expired timeout message");
}); return invoke_message_result::dropped;
auto call_default_handler = [&] { }
auto sres = call_handler(default_handler_, this, x); // handle everything else as ordinary message
switch (sres.flag) { detail::default_invoke_result_visitor<event_based_actor> visitor{this};
bool skipped = false;
auto had_timeout = getf(has_timeout_flag);
if (had_timeout)
unsetf(has_timeout_flag);
// restore timeout at scope exit if message was skipped
auto timeout_guard = detail::make_scope_guard([&] {
if (skipped && had_timeout)
setf(has_timeout_flag);
});
auto call_default_handler = [&] {
auto sres = call_handler(default_handler_, this, x);
switch (sres.flag) {
default:
break;
case rt_error:
case rt_value:
visitor.visit(sres);
break;
case rt_skip:
skipped = true;
}
};
if (bhvr_stack_.empty()) {
call_default_handler();
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
}
auto& bhvr = bhvr_stack_.back();
switch (bhvr(visitor, x.content())) {
default: default:
break; break;
case rt_error: case match_case::skip:
case rt_value:
visitor.visit(sres);
break;
case rt_skip:
skipped = true; skipped = true;
break;
case match_case::no_match:
call_default_handler();
} }
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
}; };
if (bhvr_stack_.empty()) { // Post-process the returned value from the function body.
call_default_handler(); auto result = body();
auto result = !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
CAF_AFTER_PROCESSING(this, result);
return result;
}
auto& bhvr = bhvr_stack_.back();
switch (bhvr(visitor, x.content())) {
default:
break;
case match_case::skip:
skipped = true;
break;
case match_case::no_match:
call_default_handler();
}
auto result = !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
CAF_AFTER_PROCESSING(this, result); CAF_AFTER_PROCESSING(this, result);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(result);
return result; return result;
} }
......
...@@ -630,111 +630,107 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) { ...@@ -630,111 +630,107 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
current_element_ = &x; current_element_ = &x;
CAF_LOG_RECEIVE_EVENT(current_element_); CAF_LOG_RECEIVE_EVENT(current_element_);
CAF_BEFORE_PROCESSING(this, x); CAF_BEFORE_PROCESSING(this, x);
// Helper function for dispatching a message to a response handler. // Wrap the actual body for the function.
using ptr_t = scheduled_actor*; auto body = [this, &x] {
using fun_t = bool (*)(ptr_t, behavior&, mailbox_element&); // Helper function for dispatching a message to a response handler.
auto ordinary_invoke = [](ptr_t, behavior& f, mailbox_element& in) -> bool { using ptr_t = scheduled_actor*;
return f(in.content()) != none; using fun_t = bool (*)(ptr_t, behavior&, mailbox_element&);
}; auto ordinary_invoke = [](ptr_t, behavior& f, mailbox_element& in) -> bool {
auto select_invoke_fun = [&]() -> fun_t { return ordinary_invoke; }; return f(in.content()) != none;
// Short-circuit awaited responses. };
if (!awaited_responses_.empty()) { auto select_invoke_fun = [&]() -> fun_t { return ordinary_invoke; };
auto invoke = select_invoke_fun(); // Short-circuit awaited responses.
auto& pr = awaited_responses_.front(); if (!awaited_responses_.empty()) {
// skip all messages until we receive the currently awaited response auto invoke = select_invoke_fun();
if (x.mid != pr.first) { auto& pr = awaited_responses_.front();
CAF_AFTER_PROCESSING(this, invoke_message_result::skipped); // skip all messages until we receive the currently awaited response
return invoke_message_result::skipped; if (x.mid != pr.first)
} return invoke_message_result::skipped;
auto f = std::move(pr.second); auto f = std::move(pr.second);
awaited_responses_.pop_front(); awaited_responses_.pop_front();
if (!invoke(this, f, x)) { if (!invoke(this, f, x)) {
// try again with error if first attempt failed // try again with error if first attempt failed
auto msg = make_message( auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message())); make_error(sec::unexpected_response, x.move_content_to_message()));
f(msg); f(msg);
} }
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed); return invoke_message_result::consumed;
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()) {
CAF_AFTER_PROCESSING(this, invoke_message_result::dropped);
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);
} }
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed); // Handle multiplexed responses.
return invoke_message_result::consumed; if (x.mid.is_response()) {
} auto invoke = select_invoke_fun();
// Dispatch on the content of x. auto mrh = multiplexed_responses_.find(x.mid);
switch (categorize(x)) { // neither awaited nor multiplexed, probably an expired timeout
case message_category::skipped: if (mrh == multiplexed_responses_.end())
CAF_AFTER_PROCESSING(this, invoke_message_result::skipped); return invoke_message_result::dropped;
return invoke_message_result::skipped; auto bhvr = std::move(mrh->second);
case message_category::internal: multiplexed_responses_.erase(mrh);
CAF_LOG_DEBUG("handled system message"); if (!invoke(this, bhvr, x)) {
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed); // 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; return invoke_message_result::consumed;
case message_category::ordinary: { }
detail::default_invoke_result_visitor<scheduled_actor> visitor{this}; // Dispatch on the content of x.
bool skipped = false; switch (categorize(x)) {
auto had_timeout = getf(has_timeout_flag); case message_category::skipped:
if (had_timeout) return invoke_message_result::skipped;
unsetf(has_timeout_flag); case message_category::internal:
// restore timeout at scope exit if message was skipped CAF_LOG_DEBUG("handled system message");
auto timeout_guard = detail::make_scope_guard([&] { return invoke_message_result::consumed;
if (skipped && had_timeout) case message_category::ordinary: {
setf(has_timeout_flag); detail::default_invoke_result_visitor<scheduled_actor> visitor{this};
}); bool skipped = false;
auto call_default_handler = [&] { auto had_timeout = getf(has_timeout_flag);
auto sres = call_handler(default_handler_, this, x); if (had_timeout)
switch (sres.flag) { unsetf(has_timeout_flag);
// restore timeout at scope exit if message was skipped
auto timeout_guard = detail::make_scope_guard([&] {
if (skipped && had_timeout)
setf(has_timeout_flag);
});
auto call_default_handler = [&] {
auto sres = call_handler(default_handler_, this, x);
switch (sres.flag) {
default:
break;
case rt_error:
case rt_value:
visitor.visit(sres);
break;
case rt_skip:
skipped = true;
}
};
if (bhvr_stack_.empty()) {
call_default_handler();
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
}
auto& bhvr = bhvr_stack_.back();
switch (bhvr(visitor, x.content())) {
default: default:
break; break;
case rt_error: case match_case::skip:
case rt_value:
visitor.visit(sres);
break;
case rt_skip:
skipped = true; skipped = true;
break;
case match_case::no_match:
call_default_handler();
} }
}; return !skipped ? invoke_message_result::consumed
if (bhvr_stack_.empty()) { : invoke_message_result::skipped;
call_default_handler();
auto result = !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
CAF_AFTER_PROCESSING(this, result);
return result;
} }
auto& bhvr = bhvr_stack_.back();
switch (bhvr(visitor, x.content())) {
default:
break;
case match_case::skip:
skipped = true;
break;
case match_case::no_match:
call_default_handler();
}
auto result = !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
CAF_AFTER_PROCESSING(this, result);
return result;
} }
} // Unreachable.
// Unreachable. CAF_CRITICAL("invalid message type");
CAF_CRITICAL("invalid message type"); };
// Post-process the returned value from the function body.
auto result = body();
CAF_AFTER_PROCESSING(this, result);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(result);
return result;
} }
/// Tries to consume `x`. /// Tries to consume `x`.
......
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