Commit 546bb80c authored by Dominik Charousset's avatar Dominik Charousset

Track requests and delegated messages

parent 2239586b
...@@ -403,20 +403,8 @@ public: ...@@ -403,20 +403,8 @@ public:
detail::implicit_conversions_t< detail::implicit_conversions_t<
typename std::decay<Ts>::type>...>::delegated_type typename std::decay<Ts>::type>...>::delegated_type
delegate(const Handle& dest, Ts&&... xs) { delegate(const Handle& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "nothing to delegate"); auto rp = make_response_promise();
using token = detail::type_list<typename detail::implicit_conversions< return rp.template delegate<P>(dest, std::forward<Ts>(xs)...);
typename std::decay<Ts>::type>::type...>;
static_assert(response_type_unbox<signatures_of_t<Handle>, token>::valid,
"receiver does not accept given message");
auto mid = current_element_->mid;
current_element_->mid = P == message_priority::high
? mid.with_high_priority()
: mid.with_normal_priority();
dest->enqueue(make_mailbox_element(std::move(current_element_->sender), mid,
std::move(current_element_->stages),
std::forward<Ts>(xs)...),
context());
return {};
} }
virtual void initialize(); virtual void initialize();
......
...@@ -21,15 +21,16 @@ ...@@ -21,15 +21,16 @@
#include <tuple> #include <tuple>
#include <chrono> #include <chrono>
#include "caf/fwd.hpp"
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/message.hpp" #include "caf/check_typed_input.hpp"
#include "caf/detail/profiled_send.hpp"
#include "caf/duration.hpp" #include "caf/duration.hpp"
#include "caf/fwd.hpp"
#include "caf/message.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/response_type.hpp"
#include "caf/response_handle.hpp"
#include "caf/message_priority.hpp" #include "caf/message_priority.hpp"
#include "caf/check_typed_input.hpp" #include "caf/response_handle.hpp"
#include "caf/response_type.hpp"
namespace caf { namespace caf {
namespace mixin { namespace mixin {
...@@ -77,17 +78,17 @@ public: ...@@ -77,17 +78,17 @@ public:
>::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");
auto dptr = static_cast<Subtype*>(this); auto self = static_cast<Subtype*>(this);
auto req_id = dptr->new_request_id(P); auto req_id = self->new_request_id(P);
if (dest) { if (dest) {
dest->eq_impl(req_id, dptr->ctrl(), dptr->context(), detail::profiled_send(self, self->ctrl(), dest, req_id, {},
std::forward<Ts>(xs)...); self->context(), std::forward<Ts>(xs)...);
dptr->request_response_timeout(timeout, req_id); self->request_response_timeout(timeout, req_id);
} else { } else {
dptr->eq_impl(req_id.response_id(), dptr->ctrl(), dptr->context(), self->eq_impl(req_id.response_id(), self->ctrl(), self->context(),
make_error(sec::invalid_argument)); make_error(sec::invalid_argument));
} }
return {req_id.response_id(), dptr}; return {req_id.response_id(), self};
} }
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`. /// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`.
......
...@@ -21,11 +21,12 @@ ...@@ -21,11 +21,12 @@
#include <vector> #include <vector>
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/message.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp"
#include "caf/check_typed_input.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"
#include "caf/check_typed_input.hpp"
namespace caf { namespace caf {
...@@ -56,15 +57,15 @@ public: ...@@ -56,15 +57,15 @@ public:
detail::enable_if_t<((sizeof...(Ts) > 0) detail::enable_if_t<((sizeof...(Ts) > 0)
|| (!std::is_convertible<T, error>::value || (!std::is_convertible<T, error>::value
&& !std::is_same<detail::decay_t<T>, unit_t>::value)) && !std::is_same<detail::decay_t<T>, unit_t>::value))
&& !detail::is_expected<detail::decay_t<T>>::value> && !detail::is_expected<detail::decay_t<T>>::value>
deliver(T&& x, Ts&&... xs) { deliver(T&& x, Ts&&... xs) {
using ts = detail::type_list<detail::decay_t<T>, detail::decay_t<Ts>...>; using ts = detail::type_list<detail::decay_t<T>, detail::decay_t<Ts>...>;
static_assert(!detail::tl_exists<ts, detail::is_result>::value, static_assert(!detail::tl_exists<ts, detail::is_result>::value,
"it is not possible to deliver objects of type result<T>"); "it is not possible to deliver objects of type result<T>");
static_assert(!detail::tl_exists<ts, detail::is_expected>::value, static_assert(!detail::tl_exists<ts, detail::is_expected>::value,
"mixing expected<T> with regular values is not supported"); "mixing expected<T> with regular values is not supported");
return deliver_impl(make_message(std::forward<T>(x), return deliver_impl(
std::forward<Ts>(xs)...)); make_message(std::forward<T>(x), std::forward<Ts>(xs)...));
} }
template <class T> template <class T>
...@@ -75,28 +76,22 @@ public: ...@@ -75,28 +76,22 @@ public:
} }
/// Satisfies the promise by delegating to another actor. /// Satisfies the promise by delegating to another actor.
template <message_priority P = message_priority::normal, template <message_priority P = message_priority::normal, class Handle = actor,
class Handle = actor, class... Ts> class... Ts>
typename response_type< typename response_type<typename Handle::signatures,
typename Handle::signatures, detail::implicit_conversions_t<
detail::implicit_conversions_t<typename std::decay<Ts>::type>... typename std::decay<Ts>::type>...>::delegated_type
>::delegated_type
delegate(const Handle& dest, Ts&&... xs) { delegate(const Handle& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "nothing to delegate"); static_assert(sizeof...(Ts) > 0, "nothing to delegate");
using token = using token = detail::type_list<typename detail::implicit_conversions<
detail::type_list< typename std::decay<Ts>::type>::type...>;
typename detail::implicit_conversions<
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 (dest) { // TODO: use `if constexpr` when switching to C++17
auto mid = P == message_priority::high ? id_.with_high_priority() : id_; if (P == message_priority::high)
dest->enqueue(make_mailbox_element(std::move(source_), mid, id_ = id_.with_high_priority();
std::move(stages_), delegate_impl(actor_cast<abstract_actor*>(dest),
std::forward<Ts>(xs)...), make_message(std::forward<Ts>(xs)...));
context());
}
return {}; return {};
} }
...@@ -143,6 +138,8 @@ private: ...@@ -143,6 +138,8 @@ private:
void deliver_impl(message msg); void deliver_impl(message msg);
void delegate_impl(abstract_actor* receiver, message msg);
strong_actor_ptr self_; strong_actor_ptr self_;
strong_actor_ptr source_; strong_actor_ptr source_;
forwarding_stack stages_; forwarding_stack stages_;
......
...@@ -108,4 +108,20 @@ void response_promise::deliver_impl(message msg) { ...@@ -108,4 +108,20 @@ void response_promise::deliver_impl(message msg) {
CAF_LOG_WARNING("malformed response promise: self != nullptr && !pending()"); CAF_LOG_WARNING("malformed response promise: self != nullptr && !pending()");
} }
void response_promise::delegate_impl(abstract_actor* receiver, message msg) {
CAF_LOG_TRACE(CAF_ARG(msg));
if (receiver == nullptr) {
CAF_LOG_DEBUG("drop response: invalid delegation target");
return;
}
if (self_ == nullptr) {
CAF_LOG_DEBUG("drop response: invalid promise");
return;
}
auto self = self_ptr();
detail::profiled_send(self, std::move(source_), receiver, id_,
std::move(stages_), self->context(), std::move(msg));
self_.reset();
}
} // namespace caf } // namespace caf
...@@ -110,19 +110,22 @@ struct fixture { ...@@ -110,19 +110,22 @@ struct fixture {
scheduler_type& sched; scheduler_type& sched;
}; };
struct foo_state { # define NAMED_ACTOR_STATE(type) \
const char* name = "foo"; struct type##_state { \
}; const char* name = #type; \
}
struct bar_state { NAMED_ACTOR_STATE(bar);
const char* name = "bar"; NAMED_ACTOR_STATE(client);
}; NAMED_ACTOR_STATE(foo);
NAMED_ACTOR_STATE(server);
NAMED_ACTOR_STATE(worker);
} // namespace } // namespace
CAF_TEST_FIXTURE_SCOPE(actor_profiler_tests, fixture) CAF_TEST_FIXTURE_SCOPE(actor_profiler_tests, fixture)
CAF_TEST(record actor construction) { CAF_TEST(profilers record actor construction) {
CAF_MESSAGE("fully initialize CAF, ignore system-internal actors"); CAF_MESSAGE("fully initialize CAF, ignore system-internal actors");
run(); run();
rec.log.clear(); rec.log.clear();
...@@ -141,7 +144,7 @@ CAF_TEST(record actor construction) { ...@@ -141,7 +144,7 @@ CAF_TEST(record actor construction) {
rec.log); rec.log);
} }
CAF_TEST(record actor messaging) { CAF_TEST(profilers record asynchronous messaging) {
CAF_MESSAGE("fully initialize CAF, ignore system-internal actors"); CAF_MESSAGE("fully initialize CAF, ignore system-internal actors");
run(); run();
rec.log.clear(); rec.log.clear();
...@@ -178,6 +181,51 @@ CAF_TEST(record actor messaging) { ...@@ -178,6 +181,51 @@ CAF_TEST(record actor messaging) {
rec.log); rec.log);
} }
CAF_TEST(profilers record request / response messaging) {
CAF_MESSAGE("fully initialize CAF, ignore system-internal actors");
run();
rec.log.clear();
CAF_MESSAGE("spawn a client and a server with one worker");
auto worker = [](stateful_actor<worker_state>*) -> behavior {
return {
[](int x, int y) { return x + y; },
};
};
auto server = [](stateful_actor<server_state>* self, actor work) -> behavior {
return {
[=](int x, int y) { return self->delegate(work, x, y); },
};
};
auto client = [](stateful_actor<client_state>* self, actor serv) {
self->request(serv, infinite, 19, 23).then([](int result) {
CAF_CHECK_EQUAL(result, 42);
});
};
sys.spawn(client, sys.spawn(server, sys.spawn(worker)));
run();
for (const auto& line : rec.log) {
CAF_MESSAGE(line);
}
CAF_CHECK_EQUAL(string_list({
"new: worker",
"new: server",
"new: client",
"client sends: (19, 23)",
"server got: (19, 23)",
"server sends: (19, 23)",
"server consumed the message",
"delete: server",
"worker got: (19, 23)",
"worker sends: (42)",
"worker consumed the message",
"client got: (42)",
"client consumed the message",
"delete: worker",
"delete: client",
}),
rec.log);
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
#endif // CAF_ENABLE_ACTOR_PROFILER #endif // CAF_ENABLE_ACTOR_PROFILER
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