Unverified Commit 9e6d1e8b authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #961

Make messages visible to the actor profiler
parents e9cb0409 f50647d0
......@@ -18,6 +18,8 @@
#pragma once
#include "caf/actor_clock.hpp"
#include "caf/config.hpp"
#include "caf/detail/build_config.hpp"
#include "caf/fwd.hpp"
......@@ -61,6 +63,33 @@ public:
virtual void after_processing(const local_actor& self,
invoke_message_result result)
= 0;
/// Called whenever an actor is about to send a message. Allows the profiler
/// to inject arbitrary meta data before putting the mailbox element into the
/// mailbox of the receiver.
/// @param self The current actor.
/// @param element The outgoing mailbox element.
/// @note The profiler gets a mutable reference to `element`, but it is only
/// supposed to inject meta data. Not to alter the message itself. Doing
/// so is an easy way to introduce bugs that are very hard to track
/// down.
/// @thread-safe
virtual void before_sending(const local_actor& self, mailbox_element& element)
= 0;
/// Analoguous to `before_sending`, but called whenever an actor is about to
/// call `actor_clock::schedule_message`.
/// @param self The current actor.
/// @param timeout Time point for the message delivery.
/// @param element The outgoing mailbox element.
/// @thread-safe
virtual void before_sending_scheduled(const local_actor& self,
caf::actor_clock::time_point timeout,
mailbox_element& element)
= 0;
// TODO: the instrumentation currently only works for actor-to-actor messages,
// but not when using group communication.
};
#ifdef CAF_ENABLE_ACTOR_PROFILER
......@@ -68,9 +97,15 @@ public:
self->system().profiler_before_processing(*self, msg)
# define CAF_AFTER_PROCESSING(self, result) \
self->system().profiler_after_processing(*self, result)
# define CAF_BEFORE_SENDING(self, msg) \
self->system().profiler_before_sending(*self, msg)
# define CAF_BEFORE_SENDING_SCHEDULED(self, timeout, msg) \
self->system().profiler_before_sending_scheduled(*self, timeout, msg)
#else
# define CAF_BEFORE_PROCESSING(self, msg) CAF_VOID_STMT
# define CAF_AFTER_PROCESSING(self, result) CAF_VOID_STMT
# define CAF_BEFORE_PROCESSING(self, msg) static_cast<void>(0)
# define CAF_AFTER_PROCESSING(self, result) static_cast<void>(0)
# define CAF_BEFORE_SENDING(self, msg) static_cast<void>(0)
# define CAF_BEFORE_SENDING_SCHEDULED(self, timeout, msg) static_cast<void>(0)
#endif
} // namespace caf
......@@ -575,6 +575,19 @@ public:
profiler_->after_processing(self, result);
}
void profiler_before_sending(const local_actor& self,
mailbox_element& element) {
if (profiler_)
profiler_->before_sending(self, element);
}
void profiler_before_sending_scheduled(const local_actor& self,
caf::actor_clock::time_point timeout,
mailbox_element& element) {
if (profiler_)
profiler_->before_sending_scheduled(self, timeout, element);
}
/// @endcond
private:
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <vector>
#include "caf/actor_cast.hpp"
#include "caf/actor_clock.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/actor_profiler.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/message_id.hpp"
#include "caf/no_stages.hpp"
namespace caf {
namespace detail {
template <class Self, class Sender, class Handle, class... Ts>
void profiled_send(Self* self, Sender&& sender, const Handle& receiver,
message_id msg_id, std::vector<strong_actor_ptr> stages,
execution_unit* context, Ts&&... xs) {
if (receiver) {
auto element = make_mailbox_element(std::forward<Sender>(sender), msg_id,
std::move(stages),
std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING(self, *element);
receiver->enqueue(std::move(element), context);
}
}
template <class Self, class Sender, class Handle, class... Ts>
void profiled_send(Self* self, Sender&& sender, const Handle& receiver,
actor_clock& clock, actor_clock::time_point timeout,
message_id msg_id, Ts&&... xs) {
if (receiver) {
auto element = make_mailbox_element(std::forward<Sender>(sender), msg_id,
no_stages, std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING_SCHEDULED(self, timeout, *element);
clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(receiver),
std::move(element));
}
}
} // namespace detail
} // namespace caf
......@@ -403,20 +403,8 @@ public:
detail::implicit_conversions_t<
typename std::decay<Ts>::type>...>::delegated_type
delegate(const Handle& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "nothing to delegate");
using token = detail::type_list<typename detail::implicit_conversions<
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 {};
auto rp = make_response_promise();
return rp.template delegate<P>(dest, std::forward<Ts>(xs)...);
}
virtual void initialize();
......
......@@ -21,15 +21,16 @@
#include <tuple>
#include <chrono>
#include "caf/fwd.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/fwd.hpp"
#include "caf/message.hpp"
#include "caf/message_id.hpp"
#include "caf/response_type.hpp"
#include "caf/response_handle.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 mixin {
......@@ -77,17 +78,17 @@ public:
>::type...>;
static_assert(response_type_unbox<signatures_of_t<Handle>, token>::valid,
"receiver does not accept given message");
auto dptr = static_cast<Subtype*>(this);
auto req_id = dptr->new_request_id(P);
auto self = static_cast<Subtype*>(this);
auto req_id = self->new_request_id(P);
if (dest) {
dest->eq_impl(req_id, dptr->ctrl(), dptr->context(),
std::forward<Ts>(xs)...);
dptr->request_response_timeout(timeout, req_id);
detail::profiled_send(self, self->ctrl(), dest, req_id, {},
self->context(), std::forward<Ts>(xs)...);
self->request_response_timeout(timeout, req_id);
} 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));
}
return {req_id.response_id(), dptr};
return {req_id.response_id(), self};
}
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`.
......
This diff is collapsed.
......@@ -21,11 +21,12 @@
#include <vector>
#include "caf/actor.hpp"
#include "caf/message.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/response_type.hpp"
#include "caf/check_typed_input.hpp"
namespace caf {
......@@ -63,8 +64,8 @@ public:
"it is not possible to deliver objects of type result<T>");
static_assert(!detail::tl_exists<ts, detail::is_expected>::value,
"mixing expected<T> with regular values is not supported");
return deliver_impl(make_message(std::forward<T>(x),
std::forward<Ts>(xs)...));
return deliver_impl(
make_message(std::forward<T>(x), std::forward<Ts>(xs)...));
}
template <class T>
......@@ -75,28 +76,22 @@ public:
}
/// Satisfies the promise by delegating to another actor.
template <message_priority P = message_priority::normal,
class Handle = actor, class... Ts>
typename response_type<
typename Handle::signatures,
detail::implicit_conversions_t<typename std::decay<Ts>::type>...
>::delegated_type
template <message_priority P = message_priority::normal, class Handle = actor,
class... Ts>
typename response_type<typename Handle::signatures,
detail::implicit_conversions_t<
typename std::decay<Ts>::type>...>::delegated_type
delegate(const Handle& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "nothing to delegate");
using token =
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>;
using token = detail::type_list<typename detail::implicit_conversions<
typename std::decay<Ts>::type>::type...>;
static_assert(response_type_unbox<signatures_of_t<Handle>, token>::valid,
"receiver does not accept given message");
if (dest) {
auto mid = P == message_priority::high ? id_.with_high_priority() : id_;
dest->enqueue(make_mailbox_element(std::move(source_), mid,
std::move(stages_),
std::forward<Ts>(xs)...),
context());
}
// TODO: use `if constexpr` when switching to C++17
if (P == message_priority::high)
id_ = id_.with_high_priority();
delegate_impl(actor_cast<abstract_actor*>(dest),
make_message(std::forward<Ts>(xs)...));
return {};
}
......@@ -111,36 +106,41 @@ public:
bool async() const;
/// 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();
}
/// Returns the source of the corresponding request.
inline const strong_actor_ptr& source() const {
const strong_actor_ptr& source() const {
return source_;
}
/// Returns the remaining stages for the corresponding request.
inline const forwarding_stack& stages() const {
const forwarding_stack& stages() const {
return stages_;
}
/// Returns the actor that will receive the response, i.e.,
/// `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();
}
/// Returns the message ID of the corresponding request.
inline message_id id() const {
message_id id() const {
return id_;
}
private:
/// 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_;
......
......@@ -18,15 +18,13 @@
#pragma once
#include "caf/scheduled_actor.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/typed_actor_view_base.hpp"
namespace caf {
struct typed_actor_view_base { };
template <class... Sigs>
class typed_actor_view : public extend<typed_actor_view_base,
typed_actor_view<Sigs...>>::template
......@@ -35,6 +33,8 @@ public:
/// Stores the template parameter pack.
using signatures = detail::type_list<Sigs...>;
using pointer = scheduled_actor*;
typed_actor_view(scheduled_actor* ptr) : self_(ptr) {
// nop
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
namespace caf {
/// Tag type for @ref typed_actor_view.
struct typed_actor_view_base {};
} // namespace caf
......@@ -16,13 +16,15 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include <utility>
#include <algorithm>
#include <utility>
#include "caf/response_promise.hpp"
#include "caf/logger.hpp"
#include "caf/detail/profiled_send.hpp"
#include "caf/local_actor.hpp"
#include "caf/logger.hpp"
#include "caf/no_stages.hpp"
namespace caf {
......@@ -37,15 +39,14 @@ response_promise::response_promise(none_t) : response_promise() {
response_promise::response_promise(strong_actor_ptr self,
strong_actor_ptr source,
forwarding_stack stages, message_id mid)
: self_(std::move(self)),
source_(std::move(source)),
stages_(std::move(stages)),
id_(mid) {
: id_(mid) {
CAF_ASSERT(self != nullptr);
// Form an invalid request promise when initialized from a response ID, since
// we always drop messages in this case.
if (mid.is_response()) {
source_ = nullptr;
stages_.clear();
if (!mid.is_response()) {
self_.swap(self);
source_.swap(source);
stages_.swap(stages);
}
}
......@@ -67,32 +68,60 @@ bool response_promise::async() const {
return id_.is_async();
}
local_actor* response_promise::self_dptr() 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() {
return self_ == nullptr
? nullptr
: static_cast<local_actor*>(actor_cast<abstract_actor*>(self_))
->context();
return self_ == nullptr ? nullptr : self_dptr()->context();
}
void response_promise::deliver_impl(message msg) {
CAF_LOG_TRACE(CAF_ARG(msg));
if (self_ == nullptr) {
CAF_LOG_DEBUG("drop response: invalid promise");
return;
}
auto dptr = self_dptr();
if (!stages_.empty()) {
auto next = std::move(stages_.back());
stages_.pop_back();
next->enqueue(make_mailbox_element(std::move(source_), id_,
std::move(stages_), std::move(msg)),
context());
detail::profiled_send(dptr, std::move(source_), next, id_,
std::move(stages_), dptr->context(), std::move(msg));
self_.reset();
return;
}
if (source_) {
source_->enqueue(std::move(self_), id_.response_id(),
std::move(msg), context());
detail::profiled_send(dptr, self_, source_, id_.response_id(), no_stages,
dptr->context(), std::move(msg));
self_.reset();
source_.reset();
return;
}
CAF_LOG_INFO_IF(self_ != nullptr, "response promise already satisfied");
CAF_LOG_INFO_IF(self_ == nullptr, "invalid response promise");
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 dptr = self_dptr();
detail::profiled_send(dptr, std::move(source_), receiver, id_,
std::move(stages_), dptr->context(), std::move(msg));
self_.reset();
}
} // namespace caf
......@@ -33,7 +33,7 @@ namespace {
using string_list = std::vector<std::string>;
struct recorder : actor_profiler {
void add_actor(const local_actor& self, const local_actor* parent) {
void add_actor(const local_actor& self, const local_actor* parent) override {
log.emplace_back("new: ");
auto& str = log.back();
str += self.name();
......@@ -43,20 +43,21 @@ struct recorder : actor_profiler {
}
}
void remove_actor(const local_actor& self) {
void remove_actor(const local_actor& self) override {
log.emplace_back("delete: ");
log.back() += self.name();
}
void before_processing(const local_actor& self,
const mailbox_element& element) {
const mailbox_element& element) override {
log.emplace_back(self.name());
auto& str = log.back();
str += " got: ";
str += to_string(element.content());
}
void after_processing(const local_actor& self, invoke_message_result result) {
void after_processing(const local_actor& self,
invoke_message_result result) override {
log.emplace_back(self.name());
auto& str = log.back();
str += " ";
......@@ -64,6 +65,23 @@ struct recorder : actor_profiler {
str += " the message";
}
void before_sending(const local_actor& self,
mailbox_element& element) override {
log.emplace_back(self.name());
auto& str = log.back();
str += " sends: ";
str += to_string(element.content());
}
void before_sending_scheduled(const local_actor& self,
actor_clock::time_point,
mailbox_element& element) override {
log.emplace_back(self.name());
auto& str = log.back();
str += " sends (scheduled): ";
str += to_string(element.content());
}
string_list log;
};
......@@ -92,19 +110,22 @@ struct fixture {
scheduler_type& sched;
};
struct foo_state {
const char* name = "foo";
};
# define NAMED_ACTOR_STATE(type) \
struct type##_state { \
const char* name = #type; \
}
struct bar_state {
const char* name = "bar";
};
NAMED_ACTOR_STATE(bar);
NAMED_ACTOR_STATE(client);
NAMED_ACTOR_STATE(foo);
NAMED_ACTOR_STATE(server);
NAMED_ACTOR_STATE(worker);
} // namespace
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");
run();
rec.log.clear();
......@@ -123,7 +144,7 @@ CAF_TEST(record actor construction) {
rec.log);
}
CAF_TEST(record actor messaging) {
CAF_TEST(profilers record asynchronous messaging) {
CAF_MESSAGE("fully initialize CAF, ignore system-internal actors");
run();
rec.log.clear();
......@@ -148,7 +169,9 @@ CAF_TEST(record actor messaging) {
CAF_CHECK_EQUAL(string_list({
"new: foo",
"new: bar, parent: foo",
"foo sends: (\"hello bar\")",
"bar got: (\"hello bar\")",
"bar sends: (\"hello foo\")",
"bar consumed the message",
"foo got: (\"hello foo\")",
"delete: bar",
......@@ -158,6 +181,51 @@ CAF_TEST(record actor messaging) {
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()
#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