Unverified Commit 6129cc74 authored by Noir's avatar Noir Committed by GitHub

Merge pull request #1200

Detect and report broken response promises
parents 84c707d8 ff0c105e
......@@ -49,6 +49,10 @@ is based on [Keep a Changelog](https://keepachangelog.com).
string input on the command-line that starts with quotes in the same way it
would parse strings from a config file, leading to very unintuitive results in
some cases (#1113).
- Response promises now implicitly share their state when copied. Once the
reference count for the state reaches zero, CAF now produces a
`broken_promise` error if the actor failed to fulfill the promise by calling
either `dispatch` or `delegate`.
### Fixed
......
......@@ -43,7 +43,7 @@ bool from_string(caf::string_view in, fixed_stack_errc& out) {
}
bool from_integer(uint8_t in, fixed_stack_errc& out) {
if (in > 0 && in < 1) {
if (in > 0 && in < 3) {
out = static_cast<fixed_stack_errc>(in);
return true;
} else {
......
......@@ -16,7 +16,8 @@ adder_actor::behavior_type server_impl(adder_actor::pointer self,
[=](add_atom, int32_t y, int32_t z) {
auto rp = self->make_response_promise<int32_t>();
self->request(worker, infinite, add_atom_v, y, z)
.then([rp](int32_t result) mutable { rp.deliver(result); });
.then([rp](int32_t result) mutable { rp.deliver(result); },
[rp](error& err) mutable { rp.deliver(std::move(err)); });
return rp;
},
};
......
......@@ -27,38 +27,15 @@ public:
void operator()(error& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
delegate(x);
self_->respond(x);
}
void operator()(message& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
delegate(x);
self_->respond(x);
}
private:
void deliver(response_promise& rp, error& x) {
CAF_LOG_DEBUG("report error back to requesting actor");
rp.deliver(std::move(x));
}
void deliver(response_promise& rp, message& x) {
CAF_LOG_DEBUG("respond via response_promise");
// suppress empty messages for asynchronous messages
if (x.empty() && rp.async())
return;
rp.deliver(std::move(x));
}
template <class T>
void delegate(T& x) {
auto rp = self_->make_response_promise();
if (!rp.pending()) {
CAF_LOG_DEBUG("suppress response message: invalid response promise");
return;
}
deliver(rp, x);
}
Self* self_;
};
......
......@@ -23,14 +23,18 @@ struct make_response_promise_helper {
};
template <class... Ts>
struct make_response_promise_helper<typed_response_promise<Ts...>>
: make_response_promise_helper<Ts...> {};
struct make_response_promise_helper<typed_response_promise<Ts...>> {
using type = typed_response_promise<Ts...>;
};
template <>
struct make_response_promise_helper<response_promise> {
using type = response_promise;
};
template <class... Ts>
using response_promise_t = typename make_response_promise_helper<Ts...>::type;
template <class Output, class F>
struct type_checker {
static void check() {
......
......@@ -341,11 +341,15 @@ public:
/// `make_response_promise<typed_response_promise<int, int>>()`
/// is equivalent to `make_response_promise<int, int>()`.
template <class... Ts>
typename detail::make_response_promise_helper<Ts...>::type
make_response_promise() {
if (current_element_ == nullptr || current_element_->mid.is_answered())
detail::response_promise_t<Ts...> make_response_promise() {
using result_t = typename detail::make_response_promise_helper<Ts...>::type;
if (current_element_ != nullptr && !current_element_->mid.is_answered()) {
auto result = result_t{this, *current_element_};
current_element_->mid.mark_as_answered();
return result;
} else {
return {};
return {this->ctrl(), *current_element_};
}
}
/// Creates a `response_promise` to respond to a request later on.
......@@ -353,16 +357,15 @@ public:
return make_response_promise<response_promise>();
}
/// Creates a `typed_response_promise` and responds immediately.
/// Return type is deduced from arguments.
/// Return value is implicitly convertible to untyped response promise.
template <class... Ts,
class R = typename detail::make_response_promise_helper<
typename std::decay<Ts>::type...>::type>
R response(Ts&&... xs) {
auto promise = make_response_promise<R>();
promise.deliver(std::forward<Ts>(xs)...);
return promise;
template <class... Ts>
[[deprecated("simply return the result from the message handler")]] //
detail::response_promise_t<std::decay_t<Ts>...>
response(Ts&&... xs) {
if (current_element_) {
response_promise::respond_to(this, current_element_,
make_message(std::forward<Ts>(xs)...));
}
return {};
}
const char* name() const override;
......@@ -429,6 +432,11 @@ public:
message_id new_request_id(message_priority mp);
template <class T>
void respond(T& x) {
response_promise::respond_to(this, current_mailbox_element(), x);
}
/// @endcond
protected:
......
This diff is collapsed.
......@@ -159,6 +159,8 @@ enum class sec : uint8_t {
unsupported_operation,
/// A key lookup failed.
no_such_key = 65,
/// An destroyed a response promise without calling deliver or delegate on it.
broken_promise,
};
// --(rst-sec-end)--
......
......@@ -12,43 +12,78 @@
namespace caf {
/// A response promise can be used to deliver a uniquely identifiable
/// response message from the server (i.e. receiver of the request)
/// to the client (i.e. the sender of the request).
/// Enables statically typed actors to delay a response message by capturing the
/// context of a request message. This is particularly useful when an actor
/// needs to communicate to other actors in order to fulfill a request for a
/// client.
template <class... Ts>
class typed_response_promise {
public:
// -- friends ----------------------------------------------------------------
friend class local_actor;
// -- member types -----------------------------------------------------------
using forwarding_stack = response_promise::forwarding_stack;
/// Constructs an invalid response promise.
// -- constructors, destructors, and assignment operators --------------------
typed_response_promise() = default;
typed_response_promise(none_t x) : promise_(x) {
typed_response_promise(typed_response_promise&&) = default;
typed_response_promise(const typed_response_promise&) = default;
typed_response_promise& operator=(typed_response_promise&&) = default;
typed_response_promise& operator=(const typed_response_promise&) = default;
[[deprecated("use the default constructor instead")]] //
typed_response_promise(none_t x)
: promise_(x) {
// nop
}
typed_response_promise(strong_actor_ptr self, strong_actor_ptr source,
forwarding_stack stages, message_id id)
: promise_(std::move(self), std::move(source), std::move(stages), id) {
// nop
// -- properties -------------------------------------------------------------
/// @copydoc response_promise::async
bool async() const {
return promise_.async();
}
typed_response_promise(strong_actor_ptr self, mailbox_element& src)
: promise_(std::move(self), src) {
// nop
/// @copydoc response_promise::pending
bool pending() const {
return promise_.pending();
}
typed_response_promise(typed_response_promise&&) = default;
typed_response_promise(const typed_response_promise&) = default;
typed_response_promise& operator=(typed_response_promise&&) = default;
typed_response_promise& operator=(const typed_response_promise&) = default;
/// @copydoc response_promise::source
strong_actor_ptr source() const {
return promise_.source();
}
/// @copydoc response_promise::stages
forwarding_stack stages() const {
return promise_.stages();
}
/// @copydoc response_promise::next
strong_actor_ptr next() const {
return promise_.next();
}
/// @copydoc response_promise::id
message_id id() const {
return promise_.id();
}
/// Implicitly convertible to untyped response promise.
[[deprecated("Use the typed_response_promise directly.")]]
[[deprecated("use the typed_response_promise directly")]]
operator response_promise&() {
return promise_;
}
// -- delivery ---------------------------------------------------------------
/// Satisfies the promise by sending a non-error response message.
template <class... Us>
std::enable_if_t<(std::is_constructible<Ts, Us>::value && ...)>
......@@ -77,6 +112,8 @@ public:
promise_.deliver(std::move(x));
}
// -- delegation -------------------------------------------------------------
/// Satisfies the promise by delegating to another actor.
template <message_priority P = message_priority::normal, class Handle = actor,
class... Us>
......@@ -84,38 +121,22 @@ public:
return promise_.template delegate<P>(dest, std::forward<Us>(xs)...);
}
/// Returns whether this response promise replies to an asynchronous message.
bool async() const {
return promise_.async();
}
/// Queries whether this promise is a valid promise that is not satisfied yet.
bool pending() const {
return promise_.pending();
}
/// Returns the source of the corresponding request.
const strong_actor_ptr& source() const {
return promise_.source();
}
private:
// -- constructors that are visible only to friends --------------------------
/// Returns the remaining stages for the corresponding request.
const forwarding_stack& stages() const {
return promise_.stages();
typed_response_promise(local_actor* self, strong_actor_ptr source,
forwarding_stack stages, message_id id)
: promise_(self, std::move(source), std::move(stages), id) {
// nop
}
/// Returns the actor that will receive the response, i.e.,
/// `stages().front()` if `!stages().empty()` or `source()` otherwise.
strong_actor_ptr next() const {
return promise_.next();
typed_response_promise(local_actor* self, mailbox_element& src)
: promise_(self, src) {
// nop
}
/// Returns the message ID of the corresponding request.
message_id id() const {
return promise_.id();
}
// -- member variables -------------------------------------------------------
private:
response_promise promise_;
};
......
......@@ -14,105 +14,176 @@
namespace caf {
response_promise::response_promise() : self_(nullptr) {
// nop
namespace {
bool requires_response(const strong_actor_ptr& source,
const mailbox_element::forwarding_stack& stages,
message_id mid) {
return !mid.is_response() && !mid.is_answered()
&& (source || !stages.empty());
}
response_promise::response_promise(none_t) : response_promise() {
// nop
bool requires_response(const mailbox_element& src) {
return requires_response(src.sender, src.stages, src.mid);
}
response_promise::response_promise(strong_actor_ptr self,
strong_actor_ptr source,
forwarding_stack stages, message_id mid)
: id_(mid) {
} // namespace
// -- constructors, destructors, and assignment operators ----------------------
response_promise::response_promise(local_actor* self, strong_actor_ptr source,
forwarding_stack stages, message_id mid) {
CAF_ASSERT(self != nullptr);
// Form an invalid request promise when initialized from a response ID, since
// we always drop messages in this case.
if (!mid.is_response()) {
self_.swap(self);
source_.swap(source);
stages_.swap(stages);
// we always drop messages in this case. Also don't create promises for
// anonymous messages since there's nowhere to send the message to anyway.
if (requires_response(source, stages, mid)) {
state_ = make_counted<state>();
state_->self = self;
state_->source.swap(source);
state_->stages.swap(stages);
state_->id = mid;
}
}
response_promise::response_promise(strong_actor_ptr self, mailbox_element& src)
: response_promise(std::move(self), std::move(src.sender),
std::move(src.stages), src.mid) {
response_promise::response_promise(local_actor* self, mailbox_element& src)
: response_promise(self, std::move(src.sender), std::move(src.stages),
src.mid) {
// nop
}
// -- properties ---------------------------------------------------------------
bool response_promise::async() const noexcept {
return id().is_async();
}
bool response_promise::pending() const noexcept {
return state_ != nullptr && state_->self != nullptr;
}
strong_actor_ptr response_promise::source() const noexcept {
if (state_)
return state_->source;
else
return nullptr;
}
response_promise::forwarding_stack response_promise::stages() const {
if (state_)
return state_->stages;
else
return {};
}
strong_actor_ptr response_promise::next() const noexcept {
if (state_)
return state_->stages.empty() ? state_->source : state_->stages[0];
else
return nullptr;
}
message_id response_promise::id() const noexcept {
if (state_)
return state_->id;
else
return make_message_id();
}
// -- delivery -----------------------------------------------------------------
void response_promise::deliver(message msg) {
CAF_LOG_TRACE(CAF_ARG(msg));
if (pending()) {
state_->deliver_impl(std::move(msg));
state_.reset();
}
}
void response_promise::deliver(error x) {
deliver_impl(make_message(std::move(x)));
CAF_LOG_TRACE(CAF_ARG(x));
if (pending()) {
state_->deliver_impl(make_message(std::move(x)));
state_.reset();
}
}
void response_promise::deliver() {
deliver_impl(make_message());
CAF_LOG_TRACE(CAF_ARG(""));
if (pending()) {
state_->deliver_impl(make_message());
state_.reset();
}
}
bool response_promise::async() const {
return id_.is_async();
void response_promise::respond_to(local_actor* self, mailbox_element* request,
message& response) {
if (request && requires_response(*request)) {
state tmp;
tmp.self = self;
tmp.source.swap(request->sender);
tmp.stages.swap(request->stages);
tmp.id = request->mid;
tmp.deliver_impl(std::move(response));
request->mid.mark_as_answered();
}
}
local_actor* response_promise::self_dptr() const {
// TODO: We require that self_ was constructed by using a local_actor*. The
// type erasure performed by strong_actor_ptr hides that fact. We
// probably should use a different pointer type such as
// intrusive_ptr<local_actor>. However, that would mean we would have to
// include local_actor.hpp in response_promise.hpp or provide overloads
// for intrusive_ptr_add_ref and intrusive_ptr_release.
auto self_baseptr = actor_cast<abstract_actor*>(self_);
return static_cast<local_actor*>(self_baseptr);
void response_promise::respond_to(local_actor* self, mailbox_element* request,
error& response) {
if (request && requires_response(*request)) {
state tmp;
tmp.self = self;
tmp.source.swap(request->sender);
tmp.stages.swap(request->stages);
tmp.id = request->mid;
tmp.deliver_impl(make_message(std::move(response)));
request->mid.mark_as_answered();
}
}
// -- state --------------------------------------------------------------------
response_promise::state::~state() {
if (self) {
CAF_LOG_DEBUG("broken promise!");
deliver_impl(make_message(make_error(sec::broken_promise)));
}
}
execution_unit* response_promise::context() {
return self_ == nullptr ? nullptr : self_dptr()->context();
void response_promise::state::cancel() {
self = nullptr;
}
void response_promise::deliver_impl(message msg) {
void response_promise::state::deliver_impl(message msg) {
CAF_LOG_TRACE(CAF_ARG(msg));
if (self_ == nullptr) {
CAF_LOG_DEBUG("drop response: invalid promise");
return;
}
if (msg.empty() && id_.is_async()) {
if (msg.empty() && id.is_async()) {
CAF_LOG_DEBUG("drop response: empty response to asynchronous input");
self_.reset();
return;
} else {
if (stages.empty()) {
detail::profiled_send(self, self->ctrl(), source, id.response_id(),
forwarding_stack{}, self->context(),
std::move(msg));
} else {
auto next = std::move(stages.back());
stages.pop_back();
detail::profiled_send(self, std::move(source), next, id,
std::move(stages), self->context(), std::move(msg));
}
auto dptr = self_dptr();
if (!stages_.empty()) {
auto next = std::move(stages_.back());
stages_.pop_back();
detail::profiled_send(dptr, std::move(source_), next, id_,
std::move(stages_), dptr->context(), std::move(msg));
self_.reset();
return;
}
if (source_) {
detail::profiled_send(dptr, self_, source_, id_.response_id(), no_stages,
dptr->context(), std::move(msg));
self_.reset();
source_.reset();
return;
}
CAF_LOG_WARNING("malformed response promise: self != nullptr && !pending()");
cancel();
}
void response_promise::delegate_impl(abstract_actor* receiver, message msg) {
void response_promise::state::delegate_impl(abstract_actor* receiver,
message msg) {
CAF_LOG_TRACE(CAF_ARG(msg));
if (receiver == nullptr) {
if (receiver != nullptr) {
detail::profiled_send(self, std::move(source), receiver, id,
std::move(stages), self->context(), std::move(msg));
} else {
CAF_LOG_DEBUG("drop response: invalid delegation target");
return;
}
if (self_ == nullptr) {
CAF_LOG_DEBUG("drop response: invalid promise");
return;
}
auto dptr = self_dptr();
detail::profiled_send(dptr, std::move(source_), receiver, id_,
std::move(stages_), dptr->context(), std::move(msg));
self_.reset();
cancel();
}
} // namespace caf
......@@ -148,6 +148,8 @@ std::string to_string(sec x) {
return "unsupported_operation";
case sec::no_such_key:
return "no_such_key";
case sec::broken_promise:
return "broken_promise";
};
}
......@@ -350,6 +352,9 @@ bool from_string(string_view in, sec& out) {
} else if (in == "no_such_key") {
out = sec::no_such_key;
return true;
} else if (in == "broken_promise") {
out = sec::broken_promise;
return true;
} else {
return false;
}
......@@ -427,6 +432,7 @@ bool from_integer(std::underlying_type_t<sec> in,
case sec::type_clash:
case sec::unsupported_operation:
case sec::no_such_key:
case sec::broken_promise:
out = result;
return true;
};
......
......@@ -240,7 +240,7 @@ stream_slot
stream_manager::add_unchecked_outbound_path_impl(strong_actor_ptr next,
message handshake) {
CAF_LOG_TRACE(CAF_ARG(next) << CAF_ARG(handshake));
response_promise rp{self_->ctrl(), self_->ctrl(), {next}, make_message_id()};
response_promise rp{self_, self_->ctrl(), {next}, make_message_id()};
return add_unchecked_outbound_path_impl(rp, std::move(handshake));
}
......
......@@ -140,6 +140,21 @@ SCENARIO("response promises allow delaying of response messages") {
}
}
SCENARIO("response promises send errors when broken") {
auto adder_hdl = sys.spawn(adder);
auto hdl = sys.spawn(requester_v1, adder_hdl);
GIVEN("a dispatcher, and adder and a client") {
WHEN("the dispatcher terminates before calling deliver on its promise") {
inject((int, int), from(self).to(hdl).with(3, 4));
inject((exit_msg),
to(hdl).with(exit_msg{hdl.address(), exit_reason::kill}));
THEN("clients receive a broken_promise error") {
expect((error), from(hdl).to(self).with(sec::broken_promise));
}
}
}
}
SCENARIO("response promises allow delegation") {
GIVEN("a dispatcher that calls delegate on its promise") {
auto adder_hdl = sys.spawn(adder);
......
......@@ -145,6 +145,21 @@ SCENARIO("response promises allow delaying of response messages") {
}
}
SCENARIO("response promises send errors when broken") {
auto adder_hdl = sys.spawn(adder);
auto hdl = sys.spawn(requester_v1, adder_hdl);
GIVEN("a dispatcher, and adder and a client") {
WHEN("the dispatcher terminates before calling deliver on its promise") {
inject((int, int), from(self).to(hdl).with(3, 4));
inject((exit_msg),
to(hdl).with(exit_msg{hdl.address(), exit_reason::kill}));
THEN("clients receive a broken_promise error") {
expect((error), from(hdl).to(self).with(sec::broken_promise));
}
}
}
}
SCENARIO("response promises allow delegation") {
GIVEN("a dispatcher that calls delegate on its promise") {
auto adder_hdl = sys.spawn(adder);
......
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