Commit d155a7c8 authored by Dominik Charousset's avatar Dominik Charousset

Add new hook to the actor_profiler for messages

parent 1b1aae99
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#pragma once #pragma once
#include "caf/actor_clock.hpp"
#include "caf/detail/build_config.hpp" #include "caf/detail/build_config.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
...@@ -61,6 +62,33 @@ public: ...@@ -61,6 +62,33 @@ public:
virtual void after_processing(const local_actor& self, virtual void after_processing(const local_actor& self,
invoke_message_result result) invoke_message_result result)
= 0; = 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 #ifdef CAF_ENABLE_ACTOR_PROFILER
...@@ -68,9 +96,15 @@ public: ...@@ -68,9 +96,15 @@ public:
self->system().profiler_before_processing(*self, msg) self->system().profiler_before_processing(*self, msg)
# define CAF_AFTER_PROCESSING(self, result) \ # define CAF_AFTER_PROCESSING(self, result) \
self->system().profiler_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 #else
# define CAF_BEFORE_PROCESSING(self, msg) CAF_VOID_STMT # define CAF_BEFORE_PROCESSING(self, msg) CAF_VOID_STMT
# define CAF_AFTER_PROCESSING(self, result) CAF_VOID_STMT # define CAF_AFTER_PROCESSING(self, result) CAF_VOID_STMT
# define CAF_BEFORE_SENDING(self, msg) CAF_VOID_STMT
# define CAF_BEFORE_SENDING_DELAYED(self, msg) CAF_VOID_STMT
#endif #endif
} // namespace caf } // namespace caf
...@@ -575,6 +575,19 @@ public: ...@@ -575,6 +575,19 @@ public:
profiler_->after_processing(self, result); 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 /// @endcond
private: private:
......
...@@ -18,8 +18,8 @@ ...@@ -18,8 +18,8 @@
#pragma once #pragma once
#include <tuple>
#include <chrono> #include <chrono>
#include <tuple>
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
...@@ -35,6 +35,7 @@ ...@@ -35,6 +35,7 @@
#include "caf/response_type.hpp" #include "caf/response_type.hpp"
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/typed_actor_view_base.hpp"
namespace caf { namespace caf {
namespace mixin { namespace mixin {
...@@ -57,15 +58,17 @@ public: ...@@ -57,15 +58,17 @@ public:
// -- send function family --------------------------------------------------- // -- send function family ---------------------------------------------------
/// Sends `{xs...}` as an asynchronous message to `dest` with priority `mp`. /// Sends `{xs...}` as an asynchronous message to `dest` with priority `mp`.
template <message_priority P = message_priority::normal, template <message_priority P = message_priority::normal, class... Ts>
class Dest = actor, class... Ts> void send(const group& dest, Ts&&... xs) {
void send(const Dest& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
detail::type_list<detail::strip_and_convert_t<Ts>...> args_token; static_assert(!statically_typed<Subtype>(),
type_check(dest, args_token); "statically typed actors can only send() to other "
"statically typed actors; use anon_send() when "
"communicating to groups");
// TODO: consider whether it's feasible to track messages to groups
if (dest) if (dest)
dest->eq_impl(make_message_id(P), dptr()->ctrl(), dest->eq_impl(make_message_id(P), dptr()->ctrl(), dptr()->context(),
dptr()->context(), std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
} }
/// Sends `{xs...}` as an asynchronous message to `dest` with priority `mp`. /// Sends `{xs...}` as an asynchronous message to `dest` with priority `mp`.
...@@ -77,15 +80,42 @@ public: ...@@ -77,15 +80,42 @@ public:
"statically typed actors can only send() to other " "statically typed actors can only send() to other "
"statically typed actors; use anon_send() or request() when " "statically typed actors; use anon_send() or request() when "
"communicating with dynamically typed actors"); "communicating with dynamically typed actors");
if (dest) if (dest) {
dest->get()->eq_impl(make_message_id(P), dptr()->ctrl(), auto element = make_mailbox_element(dptr()->ctrl(), make_message_id(P),
dptr()->context(), std::forward<Ts>(xs)...); {}, std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING(dptr(), *element);
dest->enqueue(std::move(element), dptr()->context());
}
}
/// Sends `{xs...}` as an asynchronous message to `dest` with priority `mp`.
template <message_priority P = message_priority::normal, class Dest = actor,
class... Ts>
void send(const Dest& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
detail::type_list<detail::strip_and_convert_t<Ts>...> args_token;
type_check(dest, args_token);
if (dest) {
auto element = make_mailbox_element(dptr()->ctrl(), make_message_id(P),
{}, std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING(dptr(), *element);
dest->enqueue(std::move(element), dptr()->context());
}
} }
template <message_priority P = message_priority::normal, template <message_priority P = message_priority::normal, class Dest = actor,
class Dest = actor, class... Ts> class... Ts>
void anon_send(const Dest& dest, Ts&&... xs) { void anon_send(const Dest& dest, Ts&&... xs) {
caf::anon_send(dest, std::forward<Ts>(xs)...); static_assert(sizeof...(Ts) > 0, "no message to send");
using token = detail::type_list<detail::strip_and_convert_t<Ts>...>;
static_assert(response_type_unbox<signatures_of_t<Dest>, token>::valid,
"receiver does not accept given message");
if (dest) {
auto element = make_mailbox_element(nullptr, make_message_id(P), {},
std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING(dptr(), *element);
dest->enqueue(std::move(element), dptr()->context());
}
} }
/// Sends a message at given time point (or immediately if `timeout` has /// Sends a message at given time point (or immediately if `timeout` has
...@@ -95,16 +125,8 @@ public: ...@@ -95,16 +125,8 @@ public:
detail::enable_if_t<!std::is_same<Dest, group>::value> detail::enable_if_t<!std::is_same<Dest, group>::value>
scheduled_send(const Dest& dest, actor_clock::time_point timeout, scheduled_send(const Dest& dest, actor_clock::time_point timeout,
Ts&&... xs) { Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); scheduled_send_impl(make_message_id(P), dest, dptr()->system().clock(),
detail::type_list<detail::strip_and_convert_t<Ts>...> args_token; timeout, std::forward<Ts>(xs)...);
type_check(dest, args_token);
if (dest) {
auto& clock = dptr()->system().clock();
clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(dest),
make_mailbox_element(dptr()->ctrl(),
make_message_id(P), no_stages,
std::forward<Ts>(xs)...));
}
} }
/// Sends a message at given time point (or immediately if `timeout` has /// Sends a message at given time point (or immediately if `timeout` has
...@@ -115,6 +137,7 @@ public: ...@@ -115,6 +137,7 @@ public:
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert(!statically_typed<Subtype>(), static_assert(!statically_typed<Subtype>(),
"statically typed actors are not allowed to send to groups"); "statically typed actors are not allowed to send to groups");
// TODO: consider whether it's feasible to track messages to groups
if (dest) { if (dest) {
auto& clock = dptr()->system().clock(); auto& clock = dptr()->system().clock();
clock.schedule_message(timeout, dest, dptr()->ctrl(), clock.schedule_message(timeout, dest, dptr()->ctrl(),
...@@ -128,17 +151,10 @@ public: ...@@ -128,17 +151,10 @@ public:
detail::enable_if_t<!std::is_same<Dest, group>::value> detail::enable_if_t<!std::is_same<Dest, group>::value>
delayed_send(const Dest& dest, std::chrono::duration<Rep, Period> rel_timeout, delayed_send(const Dest& dest, std::chrono::duration<Rep, Period> rel_timeout,
Ts&&... xs) { Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); auto& clock = dptr()->system().clock();
detail::type_list<detail::strip_and_convert_t<Ts>...> args_token; auto timeout = clock.now() + rel_timeout;
type_check(dest, args_token); scheduled_send_impl(make_message_id(P), dest, dptr()->system().clock(),
if (dest) { timeout, std::forward<Ts>(xs)...);
auto& clock = dptr()->system().clock();
auto timeout = clock.now() + rel_timeout;
clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(dest),
make_mailbox_element(dptr()->ctrl(),
make_message_id(P), no_stages,
std::forward<Ts>(xs)...));
}
} }
/// Sends a message after a relative timeout. /// Sends a message after a relative timeout.
...@@ -149,6 +165,7 @@ public: ...@@ -149,6 +165,7 @@ public:
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert(!statically_typed<Subtype>(), static_assert(!statically_typed<Subtype>(),
"statically typed actors are not allowed to send to groups"); "statically typed actors are not allowed to send to groups");
// TODO: consider whether it's feasible to track messages to groups
if (dest) { if (dest) {
auto& clock = dptr()->system().clock(); auto& clock = dptr()->system().clock();
auto timeout = clock.now() + rtime; auto timeout = clock.now() + rtime;
...@@ -157,20 +174,63 @@ public: ...@@ -157,20 +174,63 @@ public:
} }
} }
template <message_priority P = message_priority::normal, class Dest = actor,
class... Ts>
void scheduled_anon_send(const Dest& dest, actor_clock::time_point timeout,
Ts&&... xs) {
scheduled_anon_send_impl(make_message_id(P), dest, dptr()->system().clock(),
timeout, std::forward<Ts>(xs)...);
}
template <message_priority P = message_priority::normal, class Dest = actor, template <message_priority P = message_priority::normal, class Dest = actor,
class Rep = int, class Period = std::ratio<1>, class... Ts> class Rep = int, class Period = std::ratio<1>, class... Ts>
void delayed_anon_send(const Dest& dest, void delayed_anon_send(const Dest& dest,
std::chrono::duration<Rep, Period> rtime, Ts&&... xs) { std::chrono::duration<Rep, Period> rel_timeout,
caf::delayed_anon_send<P>(dest, rtime, std::forward<Ts>(xs)...); Ts&&... xs) {
auto& clock = dptr()->system().clock();
auto timeout = clock.now() + rel_timeout;
scheduled_anon_send(make_message_id(P), dest, clock, timeout,
std::forward<Ts>(xs)...);
} }
template <class Rep = int, class Period = std::ratio<1>, class... Ts> template <class Rep = int, class Period = std::ratio<1>, class... Ts>
void delayed_anon_send(const group& dest, void delayed_anon_send(const group& dest,
std::chrono::duration<Rep, Period> rtime, Ts&&... xs) { std::chrono::duration<Rep, Period> rtime, Ts&&... xs) {
caf::delayed_anon_send(dest, rtime, std::forward<Ts>(xs)...); delayed_anon_send_impl(dest, rtime, std::forward<Ts>(xs)...);
} }
private: private:
template <class Dest, class... Ts>
void scheduled_send_impl(message_id mid, const Dest& dest, actor_clock& clock,
actor_clock::time_point timeout, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
detail::type_list<detail::strip_and_convert_t<Ts>...> args_token;
type_check(dest, args_token);
if (dest) {
auto element = make_mailbox_element(dptr()->ctrl(), mid, no_stages,
std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING_SCHEDULED(dptr(), timeout, *element);
clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(dest),
std::move(element));
}
}
template <class Dest, class... Ts>
void scheduled_anon_send_impl(message_id mid, const Dest& dest,
actor_clock& clock,
actor_clock::time_point timeout, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
detail::type_list<detail::strip_and_convert_t<Ts>...> args_token;
type_check(dest, args_token);
if (dest) {
auto element = make_mailbox_element(nullptr, mid, no_stages,
std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING_SCHEDULED(dptr(), timeout, *element);
clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(dest),
std::move(element));
}
}
template <class Dest, class ArgTypes> template <class Dest, class ArgTypes>
static void type_check(const Dest&, ArgTypes) { static void type_check(const Dest&, ArgTypes) {
static_assert(!statically_typed<Subtype>() || statically_typed<Dest>(), static_assert(!statically_typed<Subtype>() || statically_typed<Dest>(),
...@@ -186,9 +246,17 @@ private: ...@@ -186,9 +246,17 @@ private:
"this actor does not accept the response message"); "this actor does not accept the response message");
} }
Subtype* dptr() { template <class T = Base>
detail::enable_if_t<std::is_base_of<abstract_actor, T>::value, Subtype*>
dptr() {
return static_cast<Subtype*>(this); return static_cast<Subtype*>(this);
} }
template <class T = Subtype, class = detail::enable_if_t<std::is_base_of<
typed_actor_view_base, T>::value>>
typename T::pointer dptr() {
return static_cast<Subtype*>(this)->internal_ptr();
}
}; };
} // namespace mixin } // namespace mixin
......
...@@ -18,15 +18,13 @@ ...@@ -18,15 +18,13 @@
#pragma once #pragma once
#include "caf/scheduled_actor.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.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 { namespace caf {
struct typed_actor_view_base { };
template <class... Sigs> template <class... Sigs>
class typed_actor_view : public extend<typed_actor_view_base, class typed_actor_view : public extend<typed_actor_view_base,
typed_actor_view<Sigs...>>::template typed_actor_view<Sigs...>>::template
...@@ -35,6 +33,8 @@ public: ...@@ -35,6 +33,8 @@ public:
/// Stores the template parameter pack. /// Stores the template parameter pack.
using signatures = detail::type_list<Sigs...>; using signatures = detail::type_list<Sigs...>;
using pointer = scheduled_actor*;
typed_actor_view(scheduled_actor* ptr) : self_(ptr) { typed_actor_view(scheduled_actor* ptr) : self_(ptr) {
// nop // 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
...@@ -33,7 +33,7 @@ namespace { ...@@ -33,7 +33,7 @@ namespace {
using string_list = std::vector<std::string>; using string_list = std::vector<std::string>;
struct recorder : actor_profiler { 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: "); log.emplace_back("new: ");
auto& str = log.back(); auto& str = log.back();
str += self.name(); str += self.name();
...@@ -43,20 +43,21 @@ struct recorder : actor_profiler { ...@@ -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.emplace_back("delete: ");
log.back() += self.name(); log.back() += self.name();
} }
void before_processing(const local_actor& self, void before_processing(const local_actor& self,
const mailbox_element& element) { const mailbox_element& element) override {
log.emplace_back(self.name()); log.emplace_back(self.name());
auto& str = log.back(); auto& str = log.back();
str += " got: "; str += " got: ";
str += to_string(element.content()); 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()); log.emplace_back(self.name());
auto& str = log.back(); auto& str = log.back();
str += " "; str += " ";
...@@ -64,6 +65,23 @@ struct recorder : actor_profiler { ...@@ -64,6 +65,23 @@ struct recorder : actor_profiler {
str += " the message"; 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; string_list log;
}; };
...@@ -148,7 +166,9 @@ CAF_TEST(record actor messaging) { ...@@ -148,7 +166,9 @@ CAF_TEST(record actor messaging) {
CAF_CHECK_EQUAL(string_list({ CAF_CHECK_EQUAL(string_list({
"new: foo", "new: foo",
"new: bar, parent: foo", "new: bar, parent: foo",
"foo sends: (\"hello bar\")",
"bar got: (\"hello bar\")", "bar got: (\"hello bar\")",
// TODO: "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