Commit a6cf6260 authored by Dominik Charousset's avatar Dominik Charousset

Detect and report broken response promises

parent c089c764
...@@ -46,6 +46,10 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -46,6 +46,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 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 would parse strings from a config file, leading to very unintuitive results in
some cases (#1113). 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 ### Fixed
......
...@@ -11,68 +11,129 @@ ...@@ -11,68 +11,129 @@
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/check_typed_input.hpp" #include "caf/check_typed_input.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/response_type.hpp" #include "caf/response_type.hpp"
namespace caf { namespace caf {
/// A response promise can be used to deliver a uniquely identifiable /// Enables actors to delay a response message by capturing the context of a
/// response message from the server (i.e. receiver of the request) /// request message. This is particularly useful when an actor needs to
/// to the client (i.e. the sender of the request). /// communicate to other actors in order to fulfill a request for a client.
class CAF_CORE_EXPORT response_promise { class CAF_CORE_EXPORT response_promise {
public: public:
using forwarding_stack = std::vector<strong_actor_ptr>; // -- member types -----------------------------------------------------------
/// Constructs an invalid response promise. using forwarding_stack = std::vector<strong_actor_ptr>;
response_promise();
response_promise(none_t); // -- constructors, destructors, and assignment operators --------------------
response_promise(strong_actor_ptr self, strong_actor_ptr source, response_promise(strong_actor_ptr self, strong_actor_ptr source,
forwarding_stack stages, message_id id); forwarding_stack stages, message_id id);
response_promise(strong_actor_ptr self, mailbox_element& src); response_promise(strong_actor_ptr self, mailbox_element& src);
response_promise() = default;
response_promise(response_promise&&) = default; response_promise(response_promise&&) = default;
response_promise(const response_promise&) = default; response_promise(const response_promise&) = default;
response_promise& operator=(response_promise&&) = default; response_promise& operator=(response_promise&&) = default;
response_promise& operator=(const response_promise&) = default; response_promise& operator=(const response_promise&) = default;
/// Satisfies the promise by sending a non-error response message. [[deprecated("use the default constructor instead")]] response_promise(none_t)
template <class T, class... Ts> : response_promise() {
detail::enable_if_t<((sizeof...(Ts) > 0) // nop
|| (!std::is_convertible<T, error>::value }
&& !std::is_same<detail::decay_t<T>, unit_t>::value))
&& !detail::is_expected<detail::decay_t<T>>::value> // -- properties -------------------------------------------------------------
deliver(T&& x, Ts&&... xs) {
using ts = detail::type_list<detail::decay_t<T>, detail::decay_t<Ts>...>; /// Returns whether this response promise replies to an asynchronous message.
static_assert(!detail::tl_exists<ts, detail::is_result>::value, bool async() const noexcept;
"it is not possible to deliver objects of type result<T>");
static_assert(!detail::tl_exists<ts, detail::is_expected>::value, /// Queries whether this promise is a valid promise that is not satisfied yet.
bool pending() const noexcept;
/// Returns the source of the corresponding request.
strong_actor_ptr source() const noexcept;
/// Returns the remaining stages for the corresponding request.
forwarding_stack stages() const;
/// Returns the actor that will receive the response, i.e.,
/// `stages().front()` if `!stages().empty()` or `source()` otherwise.
strong_actor_ptr next() const noexcept;
/// Returns the message ID of the corresponding request.
message_id id() const noexcept;
// -- delivery ---------------------------------------------------------------
/// Satisfies the promise by sending the given message.
/// @note Drops empty messages silently when responding to an asynchronous
/// request message, i.e., if `async() == true`.
/// @post `pending() == false`
void deliver(message msg);
/// Satisfies the promise by sending an error message.
/// @post `pending() == false`
void deliver(error x);
/// Satisfies the promise by sending an empty message.
/// @note Sends no message if the request message was asynchronous, i.e., if
/// `async() == true`.
/// @post `pending() == false`
void deliver();
/// Satisfies the promise by sending an empty message.
/// @note Sends no message if the request message was asynchronous, i.e., if
/// `async() == true`.
/// @post `pending() == false`
void deliver(unit_t) {
deliver();
}
/// Satisfies the promise by sending `make_message(xs...)`.
/// @post `pending() == false`
template <class... Ts>
void deliver(Ts... xs) {
using arg_types = detail::type_list<Ts...>;
static_assert(!detail::tl_exists<arg_types, detail::is_result>::value,
"delivering a result<T> is not supported");
static_assert(!detail::tl_exists<arg_types, detail::is_expected>::value,
"mixing expected<T> with regular values is not supported"); "mixing expected<T> with regular values is not supported");
if constexpr (sizeof...(Ts) == 0 if (pending()) {
&& std::is_same<message, std::decay_t<T>>::value) state_->deliver_impl(make_message(std::move(xs)...));
deliver_impl(std::forward<T>(x)); state_.reset();
else }
deliver_impl(make_message(std::forward<T>(x), std::forward<Ts>(xs)...));
} }
/// Satisfies the promise by sending either an error or a non-error response /// Satisfies the promise by sending the content of `x`, i.e., either a value
/// message. /// of type `T` or an @ref error.
/// @post `pending() == false`
template <class T> template <class T>
void deliver(expected<T> x) { void deliver(expected<T> x) {
if (pending()) {
if (x) { if (x) {
if constexpr (std::is_same<T, void>::value) if constexpr (std::is_same<T, void>::value
deliver(); || std::is_same<T, unit_t>::value)
state_->deliver_impl(make_message());
else else
deliver(std::move(*x)); state_->deliver_impl(make_message(std::move(*x)));
} else { } else {
deliver(std::move(x.error())); state_->deliver_impl(make_message(std::move(x.error())));
} }
state_.reset();
} }
}
// -- delegation -------------------------------------------------------------
/// Satisfies the promise by delegating to another actor. /// Satisfies the promise by delegating to another actor.
/// @post `pending() == false`
template <message_priority P = message_priority::normal, class Handle = actor, template <message_priority P = message_priority::normal, class Handle = actor,
class... Ts> class... Ts>
delegated_response_type_t< delegated_response_type_t<
...@@ -84,73 +145,54 @@ public: ...@@ -84,73 +145,54 @@ public:
typename std::decay<Ts>::type>::type...>; typename std::decay<Ts>::type>::type...>;
static_assert(response_type_unbox<signatures_of_t<Handle>, token>::valid, static_assert(response_type_unbox<signatures_of_t<Handle>, token>::valid,
"receiver does not accept given message"); "receiver does not accept given message");
if (pending()) {
if constexpr (P == message_priority::high) if constexpr (P == message_priority::high)
id_ = id_.with_high_priority(); state_->id = state_->id.with_high_priority();
if constexpr (std::is_same<detail::type_list<message>, if constexpr (std::is_same<detail::type_list<message>,
detail::type_list<std::decay_t<Ts>...>>::value) detail::type_list<std::decay_t<Ts>...>>::value)
delegate_impl(actor_cast<abstract_actor*>(dest), std::forward<Ts>(xs)...); state_->delegate_impl(actor_cast<abstract_actor*>(dest),
std::forward<Ts>(xs)...);
else else
delegate_impl(actor_cast<abstract_actor*>(dest), state_->delegate_impl(actor_cast<abstract_actor*>(dest),
make_message(std::forward<Ts>(xs)...)); make_message(std::forward<Ts>(xs)...));
state_.reset();
}
return {}; return {};
} }
/// Satisfies the promise by sending an error response message. private:
void deliver(error x); // Note: response promises must remain local to their owner. Hence, we don't
// need a thread-safe reference count for the state.
/// Satisfies the promise by sending an empty message if this promise has a class state {
/// valid message ID, i.e., `async() == false`. public:
void deliver(); state() = default;
state(const state&) = delete;
/// Satisfies the promise by sending an empty message if this promise has a state& operator=(const state&) = delete;
/// valid message ID, i.e., `async() == false`. ~state();
void deliver(unit_t) {
deliver();
}
/// Returns whether this response promise replies to an asynchronous message. void cancel();
bool async() const;
/// Queries whether this promise is a valid promise that is not satisfied yet. void deliver_impl(message msg);
bool pending() const {
return source_ != nullptr || !stages_.empty();
}
/// Returns the source of the corresponding request. void delegate_impl(abstract_actor* receiver, message msg);
const strong_actor_ptr& source() const {
return source_;
}
/// Returns the remaining stages for the corresponding request. mutable size_t ref_count = 1;
const forwarding_stack& stages() const { strong_actor_ptr self;
return stages_; strong_actor_ptr source;
} forwarding_stack stages;
message_id id;
/// Returns the actor that will receive the response, i.e., friend void intrusive_ptr_add_ref(const state* ptr) {
/// `stages().front()` if `!stages().empty()` or `source()` otherwise. ++ptr->ref_count;
strong_actor_ptr next() const {
return stages_.empty() ? source_ : stages_.front();
} }
/// Returns the message ID of the corresponding request. friend void intrusive_ptr_release(const state* ptr) {
message_id id() const { if (--ptr->ref_count == 0)
return id_; delete ptr;
} }
};
private: intrusive_ptr<state> state_;
/// Returns a downcasted version of `self_`.
local_actor* self_dptr() const;
execution_unit* context();
void deliver_impl(message msg);
void delegate_impl(abstract_actor* receiver, message msg);
strong_actor_ptr self_;
strong_actor_ptr source_;
forwarding_stack stages_;
message_id id_;
}; };
} // namespace caf } // namespace caf
...@@ -159,6 +159,8 @@ enum class sec : uint8_t { ...@@ -159,6 +159,8 @@ enum class sec : uint8_t {
unsupported_operation, unsupported_operation,
/// A key lookup failed. /// A key lookup failed.
no_such_key = 65, no_such_key = 65,
/// An destroyed a response promise without calling deliver or delegate on it.
broken_promise,
}; };
// --(rst-sec-end)-- // --(rst-sec-end)--
......
...@@ -12,20 +12,18 @@ ...@@ -12,20 +12,18 @@
namespace caf { namespace caf {
/// A response promise can be used to deliver a uniquely identifiable /// Enables statically typed actors to delay a response message by capturing the
/// response message from the server (i.e. receiver of the request) /// context of a request message. This is particularly useful when an actor
/// to the client (i.e. the sender of the request). /// needs to communicate to other actors in order to fulfill a request for a
/// client.
template <class... Ts> template <class... Ts>
class typed_response_promise { class typed_response_promise {
public: public:
using forwarding_stack = response_promise::forwarding_stack; // -- member types -----------------------------------------------------------
/// Constructs an invalid response promise. using forwarding_stack = response_promise::forwarding_stack;
typed_response_promise() = default;
typed_response_promise(none_t x) : promise_(x) { // -- constructors, destructors, and assignment operators --------------------
// nop
}
typed_response_promise(strong_actor_ptr self, strong_actor_ptr source, typed_response_promise(strong_actor_ptr self, strong_actor_ptr source,
forwarding_stack stages, message_id id) forwarding_stack stages, message_id id)
...@@ -38,17 +36,61 @@ public: ...@@ -38,17 +36,61 @@ public:
// nop // nop
} }
typed_response_promise() = default;
typed_response_promise(typed_response_promise&&) = default; typed_response_promise(typed_response_promise&&) = default;
typed_response_promise(const typed_response_promise&) = default; typed_response_promise(const typed_response_promise&) = default;
typed_response_promise& operator=(typed_response_promise&&) = default; typed_response_promise& operator=(typed_response_promise&&) = default;
typed_response_promise& operator=(const typed_response_promise&) = default; typed_response_promise& operator=(const typed_response_promise&) = default;
/// Implicitly convertible to untyped response promise. [[deprecated("use the default constructor instead")]] //
[[deprecated("Use the typed_response_promise directly.")]] typed_response_promise(none_t x)
: promise_(x) {
// nop
}
// -- properties -------------------------------------------------------------
/// @copydoc response_promise::async
bool async() const {
return promise_.async();
}
/// @copydoc response_promise::pending
bool pending() const {
return promise_.pending();
}
/// @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();
}
[[deprecated("use the typed_response_promise directly")]]
operator response_promise&() { operator response_promise&() {
return promise_; return promise_;
} }
// -- delivery ---------------------------------------------------------------
/// Satisfies the promise by sending a non-error response message. /// Satisfies the promise by sending a non-error response message.
template <class... Us> template <class... Us>
std::enable_if_t<(std::is_constructible<Ts, Us>::value && ...)> std::enable_if_t<(std::is_constructible<Ts, Us>::value && ...)>
...@@ -77,6 +119,8 @@ public: ...@@ -77,6 +119,8 @@ public:
promise_.deliver(std::move(x)); promise_.deliver(std::move(x));
} }
// -- delegation -------------------------------------------------------------
/// Satisfies the promise by delegating to another actor. /// Satisfies the promise by delegating to another actor.
template <message_priority P = message_priority::normal, class Handle = actor, template <message_priority P = message_priority::normal, class Handle = actor,
class... Us> class... Us>
...@@ -84,37 +128,6 @@ public: ...@@ -84,37 +128,6 @@ public:
return promise_.template delegate<P>(dest, std::forward<Us>(xs)...); 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();
}
/// Returns the remaining stages for the corresponding request.
const forwarding_stack& stages() const {
return promise_.stages();
}
/// 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();
}
/// Returns the message ID of the corresponding request.
message_id id() const {
return promise_.id();
}
private: private:
response_promise promise_; response_promise promise_;
}; };
......
...@@ -14,25 +14,21 @@ ...@@ -14,25 +14,21 @@
namespace caf { namespace caf {
response_promise::response_promise() : self_(nullptr) { // -- constructors, destructors, and assignment operators ----------------------
// nop
}
response_promise::response_promise(none_t) : response_promise() {
// nop
}
response_promise::response_promise(strong_actor_ptr self, response_promise::response_promise(strong_actor_ptr self,
strong_actor_ptr source, strong_actor_ptr source,
forwarding_stack stages, message_id mid) forwarding_stack stages, message_id mid) {
: id_(mid) {
CAF_ASSERT(self != nullptr); CAF_ASSERT(self != nullptr);
// Form an invalid request promise when initialized from a response ID, since // Form an invalid request promise when initialized from a response ID, since
// we always drop messages in this case. // we always drop messages in this case. Also don't create promises for
if (!mid.is_response()) { // anonymous messages since there's nowhere to send the message to anyway.
self_.swap(self); if (!mid.is_response() && (source || !stages.empty())) {
source_.swap(source); state_ = make_counted<state>();
stages_.swap(stages); state_->self.swap(self);
state_->source.swap(source);
state_->stages.swap(stages);
state_->id = mid;
} }
} }
...@@ -42,77 +38,114 @@ response_promise::response_promise(strong_actor_ptr self, mailbox_element& src) ...@@ -42,77 +38,114 @@ response_promise::response_promise(strong_actor_ptr self, mailbox_element& src)
// nop // nop
} }
void response_promise::deliver(error x) { // -- properties ---------------------------------------------------------------
deliver_impl(make_message(std::move(x)));
bool response_promise::async() const noexcept {
return id().is_async();
} }
void response_promise::deliver() { bool response_promise::pending() const noexcept {
deliver_impl(make_message()); return state_ != nullptr && state_->self != nullptr;
} }
bool response_promise::async() const { strong_actor_ptr response_promise::source() const noexcept {
return id_.is_async(); if (state_)
return state_->source;
else
return nullptr;
} }
local_actor* response_promise::self_dptr() const { response_promise::forwarding_stack response_promise::stages() const {
// TODO: We require that self_ was constructed by using a local_actor*. The if (state_)
// type erasure performed by strong_actor_ptr hides that fact. We return state_->stages;
// probably should use a different pointer type such as else
// intrusive_ptr<local_actor>. However, that would mean we would have to return {};
// 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);
} }
execution_unit* response_promise::context() { strong_actor_ptr response_promise::next() const noexcept {
return self_ == nullptr ? nullptr : self_dptr()->context(); if (state_)
return state_->stages.empty() ? state_->source : state_->stages[0];
else
return nullptr;
} }
void response_promise::deliver_impl(message msg) { 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)); CAF_LOG_TRACE(CAF_ARG(msg));
if (self_ == nullptr) { if (pending()) {
CAF_LOG_DEBUG("drop response: invalid promise"); state_->deliver_impl(std::move(msg));
return; state_.reset();
} }
if (msg.empty() && id_.is_async()) { }
CAF_LOG_DEBUG("drop response: empty response to asynchronous input");
self_.reset(); void response_promise::deliver(error x) {
return; CAF_LOG_TRACE(CAF_ARG(x));
if (pending()) {
state_->deliver_impl(make_message(std::move(x)));
state_.reset();
} }
auto dptr = self_dptr(); }
if (!stages_.empty()) {
auto next = std::move(stages_.back()); void response_promise::deliver() {
stages_.pop_back(); CAF_LOG_TRACE(CAF_ARG(""));
detail::profiled_send(dptr, std::move(source_), next, id_, if (pending()) {
std::move(stages_), dptr->context(), std::move(msg)); state_->deliver_impl(make_message());
self_.reset(); state_.reset();
return;
} }
if (source_) { }
detail::profiled_send(dptr, self_, source_, id_.response_id(), no_stages,
dptr->context(), std::move(msg)); // -- state --------------------------------------------------------------------
self_.reset();
source_.reset(); response_promise::state::~state() {
return; if (self) {
CAF_LOG_DEBUG("broken promise!");
deliver_impl(make_message(make_error(sec::broken_promise)));
} }
CAF_LOG_WARNING("malformed response promise: self != nullptr && !pending()");
} }
void response_promise::delegate_impl(abstract_actor* receiver, message msg) { void response_promise::state::cancel() {
self.reset();
}
void response_promise::state::deliver_impl(message msg) {
CAF_LOG_TRACE(CAF_ARG(msg)); CAF_LOG_TRACE(CAF_ARG(msg));
if (receiver == nullptr) { if (msg.empty() && id.is_async()) {
CAF_LOG_DEBUG("drop response: invalid delegation target"); CAF_LOG_DEBUG("drop response: empty response to asynchronous input");
return; } else {
auto dptr = actor_cast<local_actor*>(self);
if (stages.empty()) {
detail::profiled_send(dptr, self, source, id.response_id(),
forwarding_stack{}, dptr->context(),
std::move(msg));
} else {
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));
} }
if (self_ == nullptr) {
CAF_LOG_DEBUG("drop response: invalid promise");
return;
} }
auto dptr = self_dptr(); cancel();
detail::profiled_send(dptr, std::move(source_), receiver, id_, }
std::move(stages_), dptr->context(), std::move(msg));
self_.reset(); void response_promise::state::delegate_impl(abstract_actor* receiver,
message msg) {
CAF_LOG_TRACE(CAF_ARG(msg));
if (receiver != nullptr) {
auto dptr = actor_cast<local_actor*>(self);
detail::profiled_send(dptr, std::move(source), receiver, id,
std::move(stages), dptr->context(), std::move(msg));
} else {
CAF_LOG_DEBUG("drop response: invalid delegation target");
}
cancel();
} }
} // namespace caf } // namespace caf
...@@ -148,6 +148,8 @@ std::string to_string(sec x) { ...@@ -148,6 +148,8 @@ std::string to_string(sec x) {
return "unsupported_operation"; return "unsupported_operation";
case sec::no_such_key: case sec::no_such_key:
return "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) { ...@@ -350,6 +352,9 @@ bool from_string(string_view in, sec& out) {
} else if (in == "no_such_key") { } else if (in == "no_such_key") {
out = sec::no_such_key; out = sec::no_such_key;
return true; return true;
} else if (in == "broken_promise") {
out = sec::broken_promise;
return true;
} else { } else {
return false; return false;
} }
...@@ -427,6 +432,7 @@ bool from_integer(std::underlying_type_t<sec> in, ...@@ -427,6 +432,7 @@ bool from_integer(std::underlying_type_t<sec> in,
case sec::type_clash: case sec::type_clash:
case sec::unsupported_operation: case sec::unsupported_operation:
case sec::no_such_key: case sec::no_such_key:
case sec::broken_promise:
out = result; out = result;
return true; return true;
}; };
......
...@@ -140,6 +140,21 @@ SCENARIO("response promises allow delaying of response messages") { ...@@ -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") { SCENARIO("response promises allow delegation") {
GIVEN("a dispatcher that calls delegate on its promise") { GIVEN("a dispatcher that calls delegate on its promise") {
auto adder_hdl = sys.spawn(adder); auto adder_hdl = sys.spawn(adder);
......
...@@ -145,6 +145,21 @@ SCENARIO("response promises allow delaying of response messages") { ...@@ -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") { SCENARIO("response promises allow delegation") {
GIVEN("a dispatcher that calls delegate on its promise") { GIVEN("a dispatcher that calls delegate on its promise") {
auto adder_hdl = sys.spawn(adder); 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