Commit 24635441 authored by Dominik Charousset's avatar Dominik Charousset

Fix before_processing calls in blocking actors

parent a09c309f
......@@ -156,7 +156,7 @@ public:
}
};
cond rc{std::move(stmt_)};
self->varargs_receive(rc, make_message_id(), std::forward<Ts>(xs)...);
self->varargs_receive(rc, std::forward<Ts>(xs)...);
}
};
......@@ -183,7 +183,7 @@ public:
}
};
cond rc{*this};
self->varargs_receive(rc, make_message_id(), std::forward<Ts>(xs)...);
self->varargs_receive(rc, std::forward<Ts>(xs)...);
}
};
......@@ -215,7 +215,6 @@ public:
blocking_actor* self;
bool& done;
receive_cond& rcc;
message_id mid;
detail::blocking_behavior& bhvr;
// Dispatches messages with high and normal priority to the same handler.
......@@ -253,14 +252,17 @@ public:
// -- modifiers --------------------------------------------------------------
/// Dequeues the next message from the mailbox that is
/// matched by given behavior.
/// Blocks until receiving a matching message.
template <class... Ts>
void receive(Ts&&... xs) {
accept_one_cond rc;
varargs_receive(rc, make_message_id(), std::forward<Ts>(xs)...);
varargs_receive(rc, std::forward<Ts>(xs)...);
}
/// Blocks until receiving the awaited response message.
void receive_response(message_id id, timestamp send_time,
detail::blocking_behavior& bhvr);
/// Receives messages for range `[begin, first)`.
/// Semantically equal to:
/// `for ( ; begin != end; ++begin) { receive(...); }`.
......@@ -322,9 +324,7 @@ public:
template <class... Ts>
do_receive_helper do_receive(Ts&&... xs) {
auto tup = std::make_tuple(std::forward<Ts>(xs)...);
auto cb = [=](receive_cond& rc) mutable {
varargs_tup_receive(rc, make_message_id(), tup);
};
auto cb = [=](receive_cond& rc) mutable { varargs_tup_receive(rc, tup); };
return {cb};
}
......@@ -371,8 +371,7 @@ public:
/// Receives messages until either a pre- or postcheck of `rcc` fails.
template <class... Ts>
void varargs_tup_receive(receive_cond& rcc, message_id mid,
std::tuple<Ts...>& tup) {
void varargs_tup_receive(receive_cond& rcc, std::tuple<Ts...>& tup) {
using namespace detail;
static_assert(sizeof...(Ts), "at least one argument required");
// extract how many arguments are actually the behavior part,
......@@ -389,23 +388,21 @@ public:
>::type;
make_blocking_behavior_t factory;
auto fun = apply_moved_args_prefixed(factory, tail_indices{}, tup, &bhvr);
receive_impl(rcc, mid, fun);
receive_impl(rcc, fun);
}
/// Receives messages until either a pre- or postcheck of `rcc` fails.
void varargs_tup_receive(receive_cond& rcc, message_id mid,
std::tuple<behavior&>& tup);
void varargs_tup_receive(receive_cond& rcc, std::tuple<behavior&>& tup);
/// Receives messages until either a pre- or postcheck of `rcc` fails.
template <class... Ts>
void varargs_receive(receive_cond& rcc, message_id mid, Ts&&... xs) {
void varargs_receive(receive_cond& rcc, Ts&&... xs) {
auto tup = std::forward_as_tuple(std::forward<Ts>(xs)...);
varargs_tup_receive(rcc, mid, tup);
varargs_tup_receive(rcc, tup);
}
/// Receives messages until either a pre- or postcheck of `rcc` fails.
void receive_impl(receive_cond& rcc, message_id mid,
detail::blocking_behavior& bhvr);
void receive_impl(receive_cond& rcc, detail::blocking_behavior& bhvr);
bool cleanup(error&& fail_state, execution_unit* host) override;
......
......@@ -26,6 +26,7 @@
#include "caf/behavior.hpp"
#include "caf/config.hpp"
#include "caf/detail/blocking_behavior.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/typed_actor_util.hpp"
......@@ -190,10 +191,11 @@ public:
g(err);
}
};
for (auto id : ids_) {
typename Self::accept_one_cond rc;
self->varargs_receive(rc, id, helper.wrap(), error_handler);
}
auto ts = make_timestamp();
behavior bhvr{helper.wrap(), error_handler};
auto blocking_bhvr = detail::make_blocking_behavior(&bhvr);
for (auto id : ids_)
self->receive_response(id, ts, blocking_bhvr);
}
const message_id_list& ids() const noexcept {
......
......@@ -20,6 +20,7 @@
#include "caf/behavior.hpp"
#include "caf/config.hpp"
#include "caf/detail/blocking_behavior.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/typed_actor_util.hpp"
......@@ -64,9 +65,9 @@ public:
template <class Self, class F, class OnError>
void receive(Self* self, F&& f, OnError&& g) const {
typename Self::accept_one_cond rc;
self->varargs_receive(rc, mid_, std::forward<F>(f),
std::forward<OnError>(g));
behavior bhvr{std::forward<F>(f), std::forward<OnError>(g)};
auto blocking_bhvr = detail::make_blocking_behavior(&bhvr);
self->receive_response(mid_, make_timestamp(), blocking_bhvr);
}
message_id id() const noexcept {
......
......@@ -28,6 +28,7 @@
#include "caf/system_messages.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/detail/blocking_behavior.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/typed_actor_util.hpp"
......@@ -145,8 +146,9 @@ public:
// TODO: this bypasses the policy. Either we deprecate `catch_all` or *all*
// policies must support it. Currently, we only enable this member
// function on the trivial policy for backwards compatibility.
typename actor_type::accept_one_cond rc;
self_->varargs_receive(rc, id(), std::move(g), std::move(f));
behavior bhvr{g};
auto blocking_bhvr = detail::make_blocking_behavior(&bhvr, std::move(f));
self_->receive_response(id(), make_timestamp(), blocking_bhvr);
}
// -- properties -------------------------------------------------------------
......
......@@ -131,6 +131,37 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) {
}, strong_actor_ptr{ctrl()}).detach();
}
void blocking_actor::receive_response(message_id id,
[[maybe_unused]] timestamp send_time,
detail::blocking_behavior& bhvr) {
CAF_LOG_TRACE(CAF_ARG(id) << CAF_ARG(send_time));
bool done = false;
auto f = [&](size_t, auto&, mailbox_element& x) -> intrusive::task_result {
if (id != x.mid)
return intrusive::task_result::skip;
CAF_LOG_RECEIVE_EVENT((&x));
CAF_BEFORE_PROCESSING(self, x, id, send_time);
// Blocking actors can nest receives => push/pop `current_element_`
auto prev = current_element_;
current_element_ = &x;
auto g = detail::make_scope_guard([&] { current_element_ = prev; });
if (bhvr.nested(x.content()) == none && bhvr.fallback(x).flag == rt_skip) {
// Invoke again with error if first attempt failed.
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
bhvr.nested(msg);
}
CAF_AFTER_PROCESSING(this, invoke_message_result::consumed);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(invoke_message_result::consumed);
done = true;
return intrusive::task_result::stop;
};
do {
await_data();
mailbox_.new_round(3, f);
} while (!done);
}
blocking_actor::receive_while_helper
blocking_actor::receive_while(std::function<bool()> stmt) {
return {this, std::move(stmt)};
......@@ -171,14 +202,9 @@ blocking_actor::mailbox_visitor::operator()(mailbox_element& x) {
done = true;
return intrusive::task_result::stop;
};
// Skip messages that don't match our message ID.
if (mid.is_response()) {
if (mid != x.mid) {
return intrusive::task_result::skip;
}
} else if (x.mid.is_response()) {
// Response messages get handled via receive_response().
if (x.mid.is_response())
return intrusive::task_result::skip;
}
// Automatically unlink from actors after receiving an exit.
if (x.content().match_elements<exit_msg>())
self->unlink_from(x.content().get_as<exit_msg>(0).source);
......@@ -199,19 +225,8 @@ blocking_actor::mailbox_visitor::operator()(mailbox_element& x) {
visitor.visit(sres);
return check_if_done();
}
return intrusive::task_result::skip;
}
// Response handlers must get re-invoked with an error when receiving an
// unexpected message.
if (mid.is_response()) {
auto err = make_error(sec::unexpected_response,
x.move_content_to_message());
mailbox_element_view<error> tmp{std::move(x.sender), x.mid,
std::move(x.stages), err};
self->current_element_ = &tmp;
bhvr.nested(tmp.content());
return check_if_done();
}
CAF_ANNOTATE_FALLTHROUGH;
case match_case::skip:
return intrusive::task_result::skip;
}
......@@ -229,13 +244,12 @@ blocking_actor::mailbox_visitor::operator()(mailbox_element& x) {
}
void blocking_actor::receive_impl(receive_cond& rcc,
message_id mid,
detail::blocking_behavior& bhvr) {
CAF_LOG_TRACE(CAF_ARG(mid));
CAF_LOG_TRACE("");
// Set to `true` by the visitor when done.
bool done = false;
// Make sure each receive sees all mailbox elements.
mailbox_visitor f{this, done, rcc, mid, bhvr};
mailbox_visitor f{this, done, rcc, bhvr};
mailbox().flush_cache();
// Check pre-condition once before entering the message consumption loop. The
// consumer performs any future check on pre and post conditions via
......@@ -285,7 +299,7 @@ mailbox_element_ptr blocking_actor::dequeue() {
return result;
}
void blocking_actor::varargs_tup_receive(receive_cond& rcc, message_id mid,
void blocking_actor::varargs_tup_receive(receive_cond& rcc,
std::tuple<behavior&>& tup) {
using namespace detail;
auto& bhvr = std::get<0>(tup);
......@@ -294,10 +308,10 @@ void blocking_actor::varargs_tup_receive(receive_cond& rcc, message_id mid,
bhvr.handle_timeout();
};
auto fun = make_blocking_behavior(&bhvr, std::move(tmp));
receive_impl(rcc, mid, fun);
receive_impl(rcc, fun);
} else {
auto fun = make_blocking_behavior(&bhvr);
receive_impl(rcc, mid, fun);
receive_impl(rcc, fun);
}
}
......
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