Commit 2239586b authored by Dominik Charousset's avatar Dominik Charousset

Track response messages in the actor profiler

parent 1568ad7f
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#include <vector> #include <vector>
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_clock.hpp" #include "caf/actor_clock.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/actor_profiler.hpp" #include "caf/actor_profiler.hpp"
...@@ -32,27 +33,26 @@ ...@@ -32,27 +33,26 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
template <class Self, class Handle, class... Ts> template <class Self, class Sender, class Handle, class... Ts>
void profiled_send(Self* self, actor_control_block* sender, void profiled_send(Self* self, Sender&& sender, const Handle& receiver,
const Handle& receiver, message_id msg_id, message_id msg_id, std::vector<strong_actor_ptr> stages,
std::vector<strong_actor_ptr> stages,
execution_unit* context, Ts&&... xs) { execution_unit* context, Ts&&... xs) {
if (receiver) { if (receiver) {
auto element = make_mailbox_element(sender, msg_id, std::move(stages), auto element = make_mailbox_element(std::forward<Sender>(sender), msg_id,
std::move(stages),
std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING(self, *element); CAF_BEFORE_SENDING(self, *element);
receiver->enqueue(std::move(element), context); receiver->enqueue(std::move(element), context);
} }
} }
template <class Self, class Handle, class... Ts> template <class Self, class Sender, class Handle, class... Ts>
void profiled_send(Self* self, actor_control_block* sender, void profiled_send(Self* self, Sender&& sender, const Handle& receiver,
const Handle& receiver, actor_clock& clock, actor_clock& clock, actor_clock::time_point timeout,
actor_clock::time_point timeout, message_id msg_id, message_id msg_id, Ts&&... xs) {
Ts&&... xs) {
if (receiver) { if (receiver) {
auto element = make_mailbox_element(sender, msg_id, no_stages, auto element = make_mailbox_element(std::forward<Sender>(sender), msg_id,
std::forward<Ts>(xs)...); no_stages, std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING_SCHEDULED(self, timeout, *element); CAF_BEFORE_SENDING_SCHEDULED(self, timeout, *element);
clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(receiver), clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(receiver),
std::move(element)); std::move(element));
......
...@@ -111,32 +111,34 @@ public: ...@@ -111,32 +111,34 @@ public:
bool async() const; bool async() const;
/// Queries whether this promise is a valid promise that is not satisfied yet. /// Queries whether this promise is a valid promise that is not satisfied yet.
inline bool pending() const { bool pending() const {
return source_ != nullptr || !stages_.empty(); return source_ != nullptr || !stages_.empty();
} }
/// Returns the source of the corresponding request. /// Returns the source of the corresponding request.
inline const strong_actor_ptr& source() const { const strong_actor_ptr& source() const {
return source_; return source_;
} }
/// Returns the remaining stages for the corresponding request. /// Returns the remaining stages for the corresponding request.
inline const forwarding_stack& stages() const { const forwarding_stack& stages() const {
return stages_; return stages_;
} }
/// Returns the actor that will receive the response, i.e., /// Returns the actor that will receive the response, i.e.,
/// `stages().front()` if `!stages().empty()` or `source()` otherwise. /// `stages().front()` if `!stages().empty()` or `source()` otherwise.
inline strong_actor_ptr next() const { strong_actor_ptr next() const {
return stages_.empty() ? source_ : stages_.front(); return stages_.empty() ? source_ : stages_.front();
} }
/// Returns the message ID of the corresponding request. /// Returns the message ID of the corresponding request.
inline message_id id() const { message_id id() const {
return id_; return id_;
} }
private: private:
local_actor* self_ptr() const;
execution_unit* context(); execution_unit* context();
void deliver_impl(message msg); void deliver_impl(message msg);
......
...@@ -16,13 +16,15 @@ ...@@ -16,13 +16,15 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include <utility>
#include <algorithm> #include <algorithm>
#include <utility>
#include "caf/response_promise.hpp" #include "caf/response_promise.hpp"
#include "caf/logger.hpp" #include "caf/detail/profiled_send.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/logger.hpp"
#include "caf/no_stages.hpp"
namespace caf { namespace caf {
...@@ -37,15 +39,14 @@ response_promise::response_promise(none_t) : response_promise() { ...@@ -37,15 +39,14 @@ response_promise::response_promise(none_t) : response_promise() {
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)
: self_(std::move(self)), : id_(mid) {
source_(std::move(source)), CAF_ASSERT(self != nullptr);
stages_(std::move(stages)),
id_(mid) {
// 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.
if (mid.is_response()) { if (!mid.is_response()) {
source_ = nullptr; self_.swap(self);
stages_.clear(); source_.swap(source);
stages_.swap(stages);
} }
} }
...@@ -67,32 +68,44 @@ bool response_promise::async() const { ...@@ -67,32 +68,44 @@ bool response_promise::async() const {
return id_.is_async(); return id_.is_async();
} }
local_actor* response_promise::self_ptr() 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);
}
execution_unit* response_promise::context() { execution_unit* response_promise::context() {
return self_ == nullptr return self_ == nullptr ? nullptr : self_ptr()->context();
? nullptr
: static_cast<local_actor*>(actor_cast<abstract_actor*>(self_))
->context();
} }
void response_promise::deliver_impl(message msg) { void response_promise::deliver_impl(message msg) {
CAF_LOG_TRACE(CAF_ARG(msg)); CAF_LOG_TRACE(CAF_ARG(msg));
if (self_ == nullptr) {
CAF_LOG_DEBUG("drop response: invalid promise");
return;
}
auto self = self_ptr();
if (!stages_.empty()) { if (!stages_.empty()) {
auto next = std::move(stages_.back()); auto next = std::move(stages_.back());
stages_.pop_back(); stages_.pop_back();
next->enqueue(make_mailbox_element(std::move(source_), id_, detail::profiled_send(self, std::move(source_), next, id_,
std::move(stages_), std::move(msg)), std::move(stages_), self->context(), std::move(msg));
context()); self_.reset();
return; return;
} }
if (source_) { if (source_) {
source_->enqueue(std::move(self_), id_.response_id(), detail::profiled_send(self, self_, source_, id_.response_id(), no_stages,
std::move(msg), context()); self->context(), std::move(msg));
self_.reset();
source_.reset(); source_.reset();
return; return;
} }
CAF_LOG_INFO_IF(self_ != nullptr, "response promise already satisfied"); CAF_LOG_WARNING("malformed response promise: self != nullptr && !pending()");
CAF_LOG_INFO_IF(self_ == nullptr, "invalid response promise");
} }
} // namespace caf } // namespace caf
...@@ -168,7 +168,7 @@ CAF_TEST(record actor messaging) { ...@@ -168,7 +168,7 @@ CAF_TEST(record actor messaging) {
"new: bar, parent: foo", "new: bar, parent: foo",
"foo sends: (\"hello bar\")", "foo sends: (\"hello bar\")",
"bar got: (\"hello bar\")", "bar got: (\"hello bar\")",
// TODO: "bar sends: (\"hello foo\")", "bar sends: (\"hello foo\")",
"bar consumed the message", "bar consumed the message",
"foo got: (\"hello foo\")", "foo got: (\"hello foo\")",
"delete: bar", "delete: bar",
......
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