Commit f64d01b5 authored by Lingxi-Li's avatar Lingxi-Li

Fix and improve response promise

parent fbba3248
...@@ -58,7 +58,7 @@ public: ...@@ -58,7 +58,7 @@ public:
this->send(x.first, std::move(x.second)); this->send(x.first, std::move(x.second));
this->become( this->become(
// collect results // collect results
others >> [=] { others >> [=]() mutable {
join_(value_, this->current_message()); join_(value_, this->current_message());
if (--awaited_results_ == 0) { if (--awaited_results_ == 0) {
rp.deliver(make_message(value_)); rp.deliver(make_message(value_));
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include <cstdint> #include <cstdint>
#include <exception> #include <exception>
#include <functional> #include <functional>
#include <type_traits>
#include <forward_list> #include <forward_list>
#include <unordered_map> #include <unordered_map>
...@@ -55,6 +56,7 @@ ...@@ -55,6 +56,7 @@
#include "caf/check_typed_input.hpp" #include "caf/check_typed_input.hpp"
#include "caf/monitorable_actor.hpp" #include "caf/monitorable_actor.hpp"
#include "caf/invoke_message_result.hpp" #include "caf/invoke_message_result.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/detail/disposer.hpp" #include "caf/detail/disposer.hpp"
#include "caf/detail/behavior_stack.hpp" #include "caf/detail/behavior_stack.hpp"
...@@ -341,6 +343,16 @@ public: ...@@ -341,6 +343,16 @@ public:
/// Creates a `response_promise` to response to a request later on. /// Creates a `response_promise` to response to a request later on.
response_promise make_response_promise(); response_promise make_response_promise();
/// Creates a `response_promise` and responds immediately.
/// Non-requests do not get an error response message.
template <class... Ts>
typed_response_promise<typename std::decay<Ts>::type...>
response(Ts&&... xs) {
auto promise = make_response_promise();
promise.deliver(std::forward<Ts>(xs)...);
return promise;
}
/// Sets a custom exception handler for this actor. If multiple handlers are /// Sets a custom exception handler for this actor. If multiple handlers are
/// defined, only the functor that was added *last* is being executed. /// defined, only the functor that was added *last* is being executed.
template <class F> template <class F>
......
...@@ -34,49 +34,42 @@ namespace caf { ...@@ -34,49 +34,42 @@ namespace caf {
/// to the client (i.e. the sender of the request). /// to the client (i.e. the sender of the request).
class response_promise { class response_promise {
public: public:
response_promise() = default; /// Constructs an invalid response promise.
response_promise();
response_promise(local_actor* self, mailbox_element& src);
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;
using forwarding_stack = std::vector<actor_addr>; /// Satisfies the promise by sending a non-error response message.
//response_promise(local_actor* self, actor_addr source,
// forwarding_stack stages,
// message_id response_id);
response_promise(local_actor* self, mailbox_element& src);
/// Queries whether this promise is still valid, i.e., no response
/// was yet delivered to the client.
inline bool valid() const {
// handle is valid if it has a receiver or a next stage
return source_ || ! stages_.empty();
}
inline explicit operator bool() const {
return valid();
}
/// Sends the response_message and invalidate this promise.
template <class T, class... Ts> template <class T, class... Ts>
typename std::enable_if< typename std::enable_if<
(sizeof...(Ts) > 0) ||
! std::is_convertible<T, error>::value ! std::is_convertible<T, error>::value
>::type >::type
deliver(T&&x, Ts&&... xs) const { deliver(T&&x, Ts&&... xs) {
deliver_impl(make_message(std::forward<T>(x), std::forward<Ts>(xs)...)); deliver_impl(make_message(std::forward<T>(x), std::forward<Ts>(xs)...));
} }
/// Sends an error as response unless the sender used asynchronous messaging /// Satisfies the promise by sending an error response message.
/// and invalidate this promise. /// For non-requests, nothing is done.
void deliver(error x) const; void deliver(error x);
/// Queries whether this promise is a valid promise that is not satisfied yet.
inline bool pending() const {
return ! stages_.empty() || source_;
}
private: private:
void deliver_impl(message response_message) const; using forwarding_stack = std::vector<actor_addr>;
void deliver_impl(message response_message);
local_actor* self_; local_actor* self_;
mutable actor_addr source_; actor_addr source_;
mutable forwarding_stack stages_; forwarding_stack stages_;
message_id id_; message_id id_;
}; };
......
...@@ -22,27 +22,57 @@ ...@@ -22,27 +22,57 @@
#include "caf/response_promise.hpp" #include "caf/response_promise.hpp"
#include "caf/detail/type_list.hpp"
namespace caf { 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).
template <class... Ts> template <class... Ts>
class typed_response_promise { class typed_response_promise {
public: public:
/// Constructs an invalid response promise.
typed_response_promise() = default; typed_response_promise() = default;
inline typed_response_promise(response_promise promise)
: promise_(std::move(promise)) {
// nop
}
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=(const typed_response_promise&) = default; typed_response_promise& operator=(const typed_response_promise&) = default;
typed_response_promise(response_promise promise) : promise_(promise) { /// Satisfies the promise by sending a non-error response message.
// nop template <class U, class... Us>
typename std::enable_if<
(sizeof...(Us) > 0) || ! std::is_convertible<U, error>::value
>::type
deliver(U&& x, Us&&... xs) {
static_assert(
std::is_same<detail::type_list<Ts...>,
detail::type_list<typename std::decay<U>::type,
typename std::decay<Us>::type...>>::value,
"typed_response_promise: message type mismatched");
promise_.deliver(std::forward<U>(x), std::forward<Us>(xs)...);
}
/// Satisfies the promise by sending an error response message.
/// For non-requests, nothing is done.
inline void deliver(error x) {
promise_.deliver(std::move(x));
} }
explicit operator bool() const { /// Returns `*this` as an untyped response promise.
// handle is valid if it has a receiver inline operator response_promise& () {
return static_cast<bool>(promise_); return promise_;
} }
template <class... Us> /// Queries whether this promise is a valid promise that is not satisfied yet.
void deliver(Us&&... xs) const { inline bool pending() const {
promise_.deliver(make_message(std::forward<Us>(xs)...)); return promise_.pending();
} }
private: private:
......
...@@ -333,7 +333,7 @@ response_promise fetch_response_promise(local_actor*, response_promise& hdl) { ...@@ -333,7 +333,7 @@ response_promise fetch_response_promise(local_actor*, response_promise& hdl) {
// enables `return request(...).then(...)` // enables `return request(...).then(...)`
bool handle_message_id_res(local_actor* self, message& res, bool handle_message_id_res(local_actor* self, message& res,
response_promise hdl) { response_promise hdl) {
CAF_ASSERT(hdl); CAF_ASSERT(hdl.pending());
CAF_LOG_TRACE(CAF_ARG(res)); CAF_LOG_TRACE(CAF_ARG(res));
if (res.match_elements<atom_value, uint64_t>() if (res.match_elements<atom_value, uint64_t>()
&& res.get_as<atom_value>(0) == atom("MESSAGE_ID")) { && res.get_as<atom_value>(0) == atom("MESSAGE_ID")) {
...@@ -345,8 +345,7 @@ bool handle_message_id_res(local_actor* self, message& res, ...@@ -345,8 +345,7 @@ bool handle_message_id_res(local_actor* self, message& res,
if (ref_opt) { if (ref_opt) {
behavior inner{std::move(ref_opt->second.first)}; behavior inner{std::move(ref_opt->second.first)};
ref_opt->second.first.assign( ref_opt->second.first.assign(
others >> [=] { others >> [=]() mutable {
// inner is const inside this lambda and mutable a C++14 feature
auto ires = const_cast<behavior&>(inner)(self->current_message()); auto ires = const_cast<behavior&>(inner)(self->current_message());
if (ires && ! handle_message_id_res(self, *ires, hdl)) if (ires && ! handle_message_id_res(self, *ires, hdl))
hdl.deliver(*ires); hdl.deliver(*ires);
...@@ -375,7 +374,7 @@ bool post_process_invoke_res(local_actor* self, bool is_sync_request, ...@@ -375,7 +374,7 @@ bool post_process_invoke_res(local_actor* self, bool is_sync_request,
auto rp = fetch_response_promise(self, hdl); auto rp = fetch_response_promise(self, hdl);
// return true if self has answered to the original request, // return true if self has answered to the original request,
// e.g., by forwarding or delegating it // e.g., by forwarding or delegating it
if (! rp) if (! rp.pending())
return res.valid(); return res.valid();
// fulfill the promise // fulfill the promise
if (res) { if (res) {
...@@ -1001,12 +1000,11 @@ void local_actor::delayed_send_impl(message_id mid, const channel& dest, ...@@ -1001,12 +1000,11 @@ void local_actor::delayed_send_impl(message_id mid, const channel& dest,
response_promise local_actor::make_response_promise() { response_promise local_actor::make_response_promise() {
auto& ptr = current_element_; auto& ptr = current_element_;
if (! ptr) if (! ptr)
return response_promise{}; return {};
auto& mid = ptr->mid; auto& mid = ptr->mid;
if (mid.is_answered()) if (mid.is_answered())
return response_promise{}; return {};
response_promise result{this, *ptr}; return {this, *ptr};
return result;
} }
const char* local_actor::name() const { const char* local_actor::name() const {
......
...@@ -18,23 +18,17 @@ ...@@ -18,23 +18,17 @@
******************************************************************************/ ******************************************************************************/
#include <utility> #include <utility>
#include <algorithm>
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/response_promise.hpp" #include "caf/response_promise.hpp"
namespace caf { namespace caf {
/* response_promise::response_promise()
response_promise::response_promise(local_actor* self, actor_addr source, : self_(nullptr) {
forwarding_stack stages, // nop
message_id id)
: self_(self),
source_(std::move(source)),
stages_(std::move(stages)),
id_(id) {
CAF_ASSERT(id.is_response() || ! id.valid());
} }
*/
response_promise::response_promise(local_actor* self, mailbox_element& src) response_promise::response_promise(local_actor* self, mailbox_element& src)
: self_(self), : self_(self),
...@@ -44,24 +38,30 @@ response_promise::response_promise(local_actor* self, mailbox_element& src) ...@@ -44,24 +38,30 @@ response_promise::response_promise(local_actor* self, mailbox_element& src)
src.mid.mark_as_answered(); src.mid.mark_as_answered();
} }
void response_promise::deliver_impl(message msg) const { void response_promise::deliver(error x) {
if (! valid()) if (id_.valid())
deliver_impl(make_message(std::move(x)));
}
void response_promise::deliver_impl(message msg) {
if (! stages_.empty()) {
auto next = std::move(stages_.back());
stages_.pop_back();
next->enqueue(mailbox_element::make(std::move(source_), id_,
std::move(stages_), std::move(msg)),
self_->context());
return; return;
if (stages_.empty()) { }
if (source_) {
source_->enqueue(self_->address(), id_.response_id(), source_->enqueue(self_->address(), id_.response_id(),
std::move(msg), self_->context()); std::move(msg), self_->context());
source_ = invalid_actor_addr;
return; return;
} }
auto next = std::move(stages_.back()); if (self_)
stages_.pop_back(); CAF_LOG_ERROR("response promise already satisfied");
next->enqueue(mailbox_element::make(std::move(source_), id_, else
std::move(stages_), std::move(msg)), CAF_LOG_ERROR("invalid response promise");
self_->context());
}
void response_promise::deliver(error x) const {
if (id_.valid())
deliver_impl(make_message(std::move(x)));
} }
} // namespace caf } // namespace caf
...@@ -30,13 +30,18 @@ using namespace caf; ...@@ -30,13 +30,18 @@ using namespace caf;
namespace { namespace {
using foo_actor = typed_actor<replies_to<get_atom, int>::with<int>, using foo_actor = typed_actor<replies_to<int>::with<int>,
replies_to<get_atom, int>::with<int>,
replies_to<get_atom, int, int>::with<int, int>, replies_to<get_atom, int, int>::with<int, int>,
replies_to<get_atom, double>::with<double>,
replies_to<get_atom, double, double>
::with<double, double>,
reacts_to<put_atom, int, int>, reacts_to<put_atom, int, int>,
reacts_to<put_atom, int, int, int>>; reacts_to<put_atom, int, int, int>>;
using foo_promise = typed_response_promise<int>; using foo_promise = typed_response_promise<int>;
using foo2_promise = typed_response_promise<int, int>; using foo2_promise = typed_response_promise<int, int>;
using foo3_promise = typed_response_promise<double>;
class foo_actor_impl : public foo_actor::base { class foo_actor_impl : public foo_actor::base {
public: public:
...@@ -46,6 +51,12 @@ public: ...@@ -46,6 +51,12 @@ public:
behavior_type make_behavior() override { behavior_type make_behavior() override {
return { return {
[=](int x) -> foo_promise {
auto resp = response(x * 2);
CAF_CHECK(! resp.pending());
resp.deliver(x * 4); // has no effect
return resp;
},
[=](get_atom, int x) -> foo_promise { [=](get_atom, int x) -> foo_promise {
auto calculator = spawn([](event_based_actor* self) -> behavior { auto calculator = spawn([](event_based_actor* self) -> behavior {
return { return {
...@@ -72,8 +83,24 @@ public: ...@@ -72,8 +83,24 @@ public:
send(calculator, next_id_, x, y); send(calculator, next_id_, x, y);
auto& entry = promises2_[next_id_++]; auto& entry = promises2_[next_id_++];
entry = make_response_promise(); entry = make_response_promise();
// verify move semantics
CAF_CHECK(entry.pending());
foo2_promise tmp(std::move(entry));
CAF_CHECK(! entry.pending());
CAF_CHECK(tmp.pending());
entry = std::move(tmp);
CAF_CHECK(entry.pending());
CAF_CHECK(! tmp.pending());
return entry; return entry;
}, },
[=](get_atom, double) -> foo3_promise {
foo3_promise resp = make_response_promise();
resp.deliver(make_error(sec::unexpected_message));
return resp;
},
[=](get_atom, double x, double y) {
return response(x * 2, y * 2);
},
[=](put_atom, int promise_id, int x) { [=](put_atom, int promise_id, int x) {
auto i = promises_.find(promise_id); auto i = promises_.find(promise_id);
if (i == promises_.end()) if (i == promises_.end())
...@@ -98,7 +125,19 @@ private: ...@@ -98,7 +125,19 @@ private:
}; };
struct fixture { struct fixture {
fixture()
: self(system, true),
foo(system.spawn<foo_actor_impl>()) {
// nop
}
~fixture() {
self->send_exit(foo, exit_reason::kill);
}
actor_system system; actor_system system;
scoped_actor self;
foo_actor foo;
}; };
} // namespace <anonymous> } // namespace <anonymous>
...@@ -106,8 +145,10 @@ struct fixture { ...@@ -106,8 +145,10 @@ struct fixture {
CAF_TEST_FIXTURE_SCOPE(typed_spawn_tests, fixture) CAF_TEST_FIXTURE_SCOPE(typed_spawn_tests, fixture)
CAF_TEST(typed_response_promise) { CAF_TEST(typed_response_promise) {
scoped_actor self{system}; typed_response_promise<int> resp;
auto foo = self->spawn<foo_actor_impl>(); resp.deliver(1); // delivers on an invalid promise has no effect
CAF_CHECK_EQUAL(static_cast<void*>(&static_cast<response_promise&>(resp)),
static_cast<void*>(&resp));
self->request(foo, get_atom::value, 42).receive( self->request(foo, get_atom::value, 42).receive(
[](int x) { [](int x) {
CAF_CHECK_EQUAL(x, 84); CAF_CHECK_EQUAL(x, 84);
...@@ -119,7 +160,67 @@ CAF_TEST(typed_response_promise) { ...@@ -119,7 +160,67 @@ CAF_TEST(typed_response_promise) {
CAF_CHECK_EQUAL(y, 104); CAF_CHECK_EQUAL(y, 104);
} }
); );
self->send_exit(foo, exit_reason::user_shutdown); self->request(foo, get_atom::value, 3.14, 3.14).receive(
[](double x, double y) {
CAF_CHECK_EQUAL(x, 3.14 * 2);
CAF_CHECK_EQUAL(y, 3.14 * 2);
},
[](const error& err) {
CAF_ERROR("unexpected error response message received: "
<< to_string(err));
}
);
}
CAF_TEST(typed_response_promise_chained) {
auto composed = foo * foo * foo;
self->request(composed, 1).receive(
[](int v) {
CAF_CHECK_EQUAL(v, 8);
},
[](const error& err) {
CAF_ERROR("unexpected error response message received: "
<< to_string(err));
}
);
}
// verify that only requests get an error response message
CAF_TEST(error_response_message) {
self->request(foo, get_atom::value, 3.14).receive(
[](double) {
CAF_ERROR("unexpected ordinary response message received");
},
[](const error& err) {
CAF_CHECK_EQUAL(err.code(), static_cast<uint8_t>(sec::unexpected_message));
}
);
self->send(foo, get_atom::value, 3.14);
self->send(foo, get_atom::value, 42);
self->receive(
[](int x) {
CAF_CHECK_EQUAL(x, 84);
},
[](double x) {
CAF_ERROR("unexpected ordinary response message received: " << x);
}
);
}
// verify that delivering to a satisfied promise has no effect
CAF_TEST(satisfied_promise) {
self->send(foo, 1);
self->send(foo, get_atom::value, 3.14, 3.14);
int i = 0;
self->receive_for(i, 2) (
[](int x) {
CAF_CHECK_EQUAL(x, 1 * 2);
},
[](double x, double y) {
CAF_CHECK_EQUAL(x, 3.14 * 2);
CAF_CHECK_EQUAL(y, 3.14 * 2);
}
);
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
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