Commit ff0c105e authored by Dominik Charousset's avatar Dominik Charousset

Fix response of void handlers, streamline API

parent 542e39ea
...@@ -13,11 +13,9 @@ namespace { ...@@ -13,11 +13,9 @@ namespace {
behavior ping(event_based_actor* self, actor pong_actor, int n) { behavior ping(event_based_actor* self, actor pong_actor, int n) {
self->send(pong_actor, ping_atom_v, n); self->send(pong_actor, ping_atom_v, n);
return { return {
[=](pong_atom, int x) -> result<ping_atom, int> { [=](pong_atom, int x) {
if (x > 1) if (x > 1)
return {ping_atom_v, x - 1}; self->send(pong_actor, ping_atom_v, x - 1);
else
return {};
}, },
}; };
} }
......
...@@ -27,38 +27,15 @@ public: ...@@ -27,38 +27,15 @@ public:
void operator()(error& x) override { void operator()(error& x) override {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
delegate(x); self_->respond(x);
} }
void operator()(message& x) override { void operator()(message& x) override {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
delegate(x); self_->respond(x);
} }
private: 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_; Self* self_;
}; };
......
...@@ -23,14 +23,18 @@ struct make_response_promise_helper { ...@@ -23,14 +23,18 @@ struct make_response_promise_helper {
}; };
template <class... Ts> template <class... Ts>
struct make_response_promise_helper<typed_response_promise<Ts...>> struct make_response_promise_helper<typed_response_promise<Ts...>> {
: make_response_promise_helper<Ts...> {}; using type = typed_response_promise<Ts...>;
};
template <> template <>
struct make_response_promise_helper<response_promise> { struct make_response_promise_helper<response_promise> {
using type = 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> template <class Output, class F>
struct type_checker { struct type_checker {
static void check() { static void check() {
......
...@@ -341,11 +341,15 @@ public: ...@@ -341,11 +341,15 @@ public:
/// `make_response_promise<typed_response_promise<int, int>>()` /// `make_response_promise<typed_response_promise<int, int>>()`
/// is equivalent to `make_response_promise<int, int>()`. /// is equivalent to `make_response_promise<int, int>()`.
template <class... Ts> template <class... Ts>
typename detail::make_response_promise_helper<Ts...>::type detail::response_promise_t<Ts...> make_response_promise() {
make_response_promise() { using result_t = typename detail::make_response_promise_helper<Ts...>::type;
if (current_element_ == nullptr || current_element_->mid.is_answered()) 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 {};
return {this->ctrl(), *current_element_}; }
} }
/// Creates a `response_promise` to respond to a request later on. /// Creates a `response_promise` to respond to a request later on.
...@@ -353,16 +357,15 @@ public: ...@@ -353,16 +357,15 @@ public:
return make_response_promise<response_promise>(); return make_response_promise<response_promise>();
} }
/// Creates a `typed_response_promise` and responds immediately. template <class... Ts>
/// Return type is deduced from arguments. [[deprecated("simply return the result from the message handler")]] //
/// Return value is implicitly convertible to untyped response promise. detail::response_promise_t<std::decay_t<Ts>...>
template <class... Ts, response(Ts&&... xs) {
class R = typename detail::make_response_promise_helper< if (current_element_) {
typename std::decay<Ts>::type...>::type> response_promise::respond_to(this, current_element_,
R response(Ts&&... xs) { make_message(std::forward<Ts>(xs)...));
auto promise = make_response_promise<R>(); }
promise.deliver(std::forward<Ts>(xs)...); return {};
return promise;
} }
const char* name() const override; const char* name() const override;
...@@ -429,6 +432,11 @@ public: ...@@ -429,6 +432,11 @@ public:
message_id new_request_id(message_priority mp); 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 /// @endcond
protected: protected:
......
...@@ -23,17 +23,21 @@ namespace caf { ...@@ -23,17 +23,21 @@ namespace caf {
/// communicate to other actors in order to fulfill a request for a client. /// 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:
// -- friends ----------------------------------------------------------------
friend class local_actor;
friend class stream_manager;
template <class...>
friend class typed_response_promise;
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using forwarding_stack = std::vector<strong_actor_ptr>; using forwarding_stack = std::vector<strong_actor_ptr>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
response_promise(strong_actor_ptr self, strong_actor_ptr source,
forwarding_stack stages, message_id id);
response_promise(strong_actor_ptr self, mailbox_element& src);
response_promise() = default; response_promise() = default;
response_promise(response_promise&&) = default; response_promise(response_promise&&) = default;
...@@ -161,6 +165,27 @@ public: ...@@ -161,6 +165,27 @@ public:
} }
private: private:
// -- constructors that are visible only to friends --------------------------
response_promise(local_actor* self, strong_actor_ptr source,
forwarding_stack stages, message_id id);
response_promise(local_actor* self, mailbox_element& src);
// -- utility functions that are visible only to friends ---------------------
/// Sends @p response as if creating a response promise from @p self and
/// @p request and then calling `deliver` on it but avoids extra overhead such
/// as heap allocations for the promise.
static void respond_to(local_actor* self, mailbox_element* request,
message& response);
/// @copydoc respond_to
static void respond_to(local_actor* self, mailbox_element* request,
error& response);
// -- nested types -----------------------------------------------------------
// Note: response promises must remain local to their owner. Hence, we don't // Note: response promises must remain local to their owner. Hence, we don't
// need a thread-safe reference count for the state. // need a thread-safe reference count for the state.
class CAF_CORE_EXPORT state { class CAF_CORE_EXPORT state {
...@@ -177,7 +202,7 @@ private: ...@@ -177,7 +202,7 @@ private:
void delegate_impl(abstract_actor* receiver, message msg); void delegate_impl(abstract_actor* receiver, message msg);
mutable size_t ref_count = 1; mutable size_t ref_count = 1;
strong_actor_ptr self; local_actor* self;
strong_actor_ptr source; strong_actor_ptr source;
forwarding_stack stages; forwarding_stack stages;
message_id id; message_id id;
...@@ -192,6 +217,8 @@ private: ...@@ -192,6 +217,8 @@ private:
} }
}; };
// -- member variables -------------------------------------------------------
intrusive_ptr<state> state_; intrusive_ptr<state> state_;
}; };
......
...@@ -19,23 +19,16 @@ namespace caf { ...@@ -19,23 +19,16 @@ namespace caf {
template <class... Ts> template <class... Ts>
class typed_response_promise { class typed_response_promise {
public: public:
// -- friends ----------------------------------------------------------------
friend class local_actor;
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using forwarding_stack = response_promise::forwarding_stack; using forwarding_stack = response_promise::forwarding_stack;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
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
}
typed_response_promise(strong_actor_ptr self, mailbox_element& src)
: promise_(std::move(self), src) {
// nop
}
typed_response_promise() = default; typed_response_promise() = default;
typed_response_promise(typed_response_promise&&) = default; typed_response_promise(typed_response_promise&&) = default;
...@@ -129,6 +122,21 @@ public: ...@@ -129,6 +122,21 @@ public:
} }
private: private:
// -- constructors that are visible only to friends --------------------------
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
}
typed_response_promise(local_actor* self, mailbox_element& src)
: promise_(self, src) {
// nop
}
// -- member variables -------------------------------------------------------
response_promise promise_; response_promise promise_;
}; };
......
...@@ -14,27 +14,41 @@ ...@@ -14,27 +14,41 @@
namespace caf { namespace caf {
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());
}
bool requires_response(const mailbox_element& src) {
return requires_response(src.sender, src.stages, src.mid);
}
} // namespace
// -- constructors, destructors, and assignment operators ---------------------- // -- constructors, destructors, and assignment operators ----------------------
response_promise::response_promise(strong_actor_ptr self, response_promise::response_promise(local_actor* self, strong_actor_ptr source,
strong_actor_ptr source,
forwarding_stack stages, message_id mid) { forwarding_stack stages, message_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. Also don't create promises for // 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. // anonymous messages since there's nowhere to send the message to anyway.
if (!mid.is_response() && (source || !stages.empty())) { if (requires_response(source, stages, mid)) {
state_ = make_counted<state>(); state_ = make_counted<state>();
state_->self.swap(self); state_->self = self;
state_->source.swap(source); state_->source.swap(source);
state_->stages.swap(stages); state_->stages.swap(stages);
state_->id = mid; state_->id = mid;
} }
} }
response_promise::response_promise(strong_actor_ptr self, mailbox_element& src) response_promise::response_promise(local_actor* self, mailbox_element& src)
: response_promise(std::move(self), std::move(src.sender), : response_promise(self, std::move(src.sender), std::move(src.stages),
std::move(src.stages), src.mid) { src.mid) {
// nop // nop
} }
...@@ -102,6 +116,32 @@ void response_promise::deliver() { ...@@ -102,6 +116,32 @@ void response_promise::deliver() {
} }
} }
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();
}
}
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 -------------------------------------------------------------------- // -- state --------------------------------------------------------------------
response_promise::state::~state() { response_promise::state::~state() {
...@@ -112,7 +152,7 @@ response_promise::state::~state() { ...@@ -112,7 +152,7 @@ response_promise::state::~state() {
} }
void response_promise::state::cancel() { void response_promise::state::cancel() {
self.reset(); self = nullptr;
} }
void response_promise::state::deliver_impl(message msg) { void response_promise::state::deliver_impl(message msg) {
...@@ -120,16 +160,15 @@ void response_promise::state::deliver_impl(message msg) { ...@@ -120,16 +160,15 @@ void response_promise::state::deliver_impl(message msg) {
if (msg.empty() && id.is_async()) { if (msg.empty() && id.is_async()) {
CAF_LOG_DEBUG("drop response: empty response to asynchronous input"); CAF_LOG_DEBUG("drop response: empty response to asynchronous input");
} else { } else {
auto dptr = actor_cast<local_actor*>(self);
if (stages.empty()) { if (stages.empty()) {
detail::profiled_send(dptr, self, source, id.response_id(), detail::profiled_send(self, self->ctrl(), source, id.response_id(),
forwarding_stack{}, dptr->context(), forwarding_stack{}, self->context(),
std::move(msg)); std::move(msg));
} else { } else {
auto next = std::move(stages.back()); auto next = std::move(stages.back());
stages.pop_back(); stages.pop_back();
detail::profiled_send(dptr, std::move(source), next, id, detail::profiled_send(self, std::move(source), next, id,
std::move(stages), dptr->context(), std::move(msg)); std::move(stages), self->context(), std::move(msg));
} }
} }
cancel(); cancel();
...@@ -139,9 +178,8 @@ void response_promise::state::delegate_impl(abstract_actor* receiver, ...@@ -139,9 +178,8 @@ void response_promise::state::delegate_impl(abstract_actor* receiver,
message msg) { message msg) {
CAF_LOG_TRACE(CAF_ARG(msg)); CAF_LOG_TRACE(CAF_ARG(msg));
if (receiver != nullptr) { if (receiver != nullptr) {
auto dptr = actor_cast<local_actor*>(self); detail::profiled_send(self, std::move(source), receiver, id,
detail::profiled_send(dptr, std::move(source), receiver, id, std::move(stages), self->context(), std::move(msg));
std::move(stages), dptr->context(), std::move(msg));
} else { } else {
CAF_LOG_DEBUG("drop response: invalid delegation target"); CAF_LOG_DEBUG("drop response: invalid delegation target");
} }
......
...@@ -240,7 +240,7 @@ stream_slot ...@@ -240,7 +240,7 @@ stream_slot
stream_manager::add_unchecked_outbound_path_impl(strong_actor_ptr next, stream_manager::add_unchecked_outbound_path_impl(strong_actor_ptr next,
message handshake) { message handshake) {
CAF_LOG_TRACE(CAF_ARG(next) << CAF_ARG(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)); return add_unchecked_outbound_path_impl(rp, std::move(handshake));
} }
......
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