Unverified Commit fddc1747 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #946

Add experimental API for actor profiling
parents 8607ad2f 41d3df49
......@@ -13,6 +13,7 @@ defaultReleaseBuildFlags = [
defaultDebugBuildFlags = defaultReleaseBuildFlags + [
'CAF_ENABLE_ADDRESS_SANITIZER:BOOL=yes',
'CAF_LOG_LEVEL:STRING=TRACE',
'CAF_ENABLE_ACTOR_PROFILER:BOOL=yes',
]
// Configures the behavior of our stages.
......
......@@ -31,3 +31,4 @@
#cmakedefine CAF_NO_EXCEPTIONS
#cmakedefine CAF_ENABLE_ACTOR_PROFILER
......@@ -81,6 +81,7 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
--with-asan alias for --with-address-sanitier
--enable-asan alias for --with-address-sanitier
--with-gcov build with gcov coverage enabled
--with-actor-profiler enables the (experimental) actor_profiler API
Convenience options:
--dev-mode sets --build-type=debug, --no-examples,
......@@ -250,6 +251,9 @@ while [ $# -ne 0 ]; do
--with-gcov)
append_cache_entry CAF_ENABLE_GCOV BOOL yes
;;
--with-actor-profiler)
append_cache_entry CAF_ENABLE_ACTOR_PROFILER BOOL yes
;;
--no-memory-management)
append_cache_entry CAF_NO_MEM_MANAGEMENT BOOL yes
;;
......
......@@ -7,6 +7,7 @@ file(GLOB_RECURSE LIBCAF_CORE_HDRS "caf/*.hpp")
enum_to_string("caf/exit_reason.hpp" "exit_reason_strings.cpp")
enum_to_string("caf/intrusive/inbox_result.hpp" "inbox_result_strings.cpp")
enum_to_string("caf/intrusive/task_result.hpp" "task_result_strings.cpp")
enum_to_string("caf/invoke_message_result.hpp" "invoke_msg_result_strings.cpp")
enum_to_string("caf/message_priority.hpp" "message_priority_strings.cpp")
enum_to_string("caf/pec.hpp" "pec_strings.cpp")
enum_to_string("caf/sec.hpp" "sec_strings.cpp")
......@@ -16,6 +17,7 @@ enum_to_string("caf/stream_priority.hpp" "stream_priority_strings.cpp")
set(LIBCAF_CORE_SRCS
"${CMAKE_CURRENT_BINARY_DIR}/exit_reason_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/inbox_result_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/invoke_msg_result_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/message_priority_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/pec_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/sec_strings.cpp"
......@@ -33,6 +35,7 @@ set(LIBCAF_CORE_SRCS
src/actor_control_block.cpp
src/actor_ostream.cpp
src/actor_pool.cpp
src/actor_profiler.cpp
src/actor_proxy.cpp
src/actor_registry.cpp
src/actor_system.cpp
......
......@@ -37,11 +37,13 @@ public:
// -- constructors, destructors, and assignment operators --------------------
explicit actor_config(execution_unit* ptr = nullptr);
explicit actor_config(execution_unit* host = nullptr,
local_actor* parent = nullptr);
// -- member variables -------------------------------------------------------
execution_unit* host;
local_actor* parent;
int flags;
input_range<const group>* groups;
detail::unique_function<behavior(local_actor*)> init_fun;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 "caf/detail/build_config.hpp"
#include "caf/fwd.hpp"
namespace caf {
/// A profiler which provides a set of callbacks for several actor operations in
/// order to collect fine-grained profiling state about the system.
/// @experimental
class actor_profiler {
public:
virtual ~actor_profiler();
/// Called whenever the actor system spawns a new actor. The system calls this
/// member function after the constructor of `self` has completed but before
/// constructing the behavior.
/// @param self The new actor.
/// @param parent Points to the parent actor unless `self` is a top-level
/// actor (in this case, `parent` has the value `nullptr`).
/// @thread-safe
virtual void add_actor(const local_actor& self, const local_actor* parent)
= 0;
/// Called before the actor system calls the destructor for `self`.
/// @param ptr Points to an actor that is about to get destroyed.
/// @thread-safe
virtual void remove_actor(const local_actor& self) = 0;
/// Called whenever an actor is about to process an element from its mailbox.
/// @param self The current actor.
/// @param element The current element from the mailbox.
/// @thread-safe
virtual void before_processing(const local_actor& self,
const mailbox_element& element)
= 0;
/// Called after an actor processed an element from its mailbox.
/// @param self The current actor.
/// @param result Stores whether the actor consumed, skipped or dropped the
/// message.
/// @thread-safe
virtual void after_processing(const local_actor& self,
invoke_message_result result)
= 0;
};
#ifdef CAF_ENABLE_ACTOR_PROFILER
# define CAF_BEFORE_PROCESSING(self, msg) \
self->system().profiler_before_processing(*self, msg)
# define CAF_AFTER_PROCESSING(self, result) \
self->system().profiler_after_processing(*self, result)
#else
# define CAF_BEFORE_PROCESSING(self, msg) CAF_VOID_STMT
# define CAF_AFTER_PROCESSING(self, result) CAF_VOID_STMT
#endif
} // namespace caf
......@@ -33,6 +33,7 @@
#include "caf/actor_clock.hpp"
#include "caf/actor_config.hpp"
#include "caf/actor_marker.hpp"
#include "caf/actor_profiler.hpp"
#include "caf/actor_registry.hpp"
#include "caf/composable_behavior_based_actor.hpp"
#include "caf/detail/init_fun_factory.hpp"
......@@ -545,10 +546,35 @@ public:
auto res = make_actor<C>(next_actor_id(), node(), this,
cfg, std::forward<Ts>(xs)...);
auto ptr = static_cast<C*>(actor_cast<abstract_actor*>(res));
#ifdef CAF_ENABLE_ACTOR_PROFILER
profiler_add_actor(*ptr, cfg.parent);
#endif
ptr->launch(cfg.host, has_lazy_init_flag(Os), has_hide_flag(Os));
return res;
}
void profiler_add_actor(const local_actor& self, const local_actor* parent) {
if (profiler_)
profiler_->add_actor(self, parent);
}
void profiler_remove_actor(const local_actor& self) {
if (profiler_)
profiler_->remove_actor(self);
}
void profiler_before_processing(const local_actor& self,
const mailbox_element& element) {
if (profiler_)
profiler_->before_processing(self, element);
}
void profiler_after_processing(const local_actor& self,
invoke_message_result result) {
if (profiler_)
profiler_->after_processing(self, result);
}
/// @endcond
private:
......@@ -577,6 +603,9 @@ private:
// -- member variables -------------------------------------------------------
/// Provides system-wide callbacks for several actor operations.
actor_profiler* profiler_;
/// Used to generate ascending actor IDs.
std::atomic<size_t> ids_;
......@@ -634,4 +663,3 @@ private:
};
} // namespace caf
......@@ -27,6 +27,7 @@
#include <unordered_map>
#include "caf/actor_factory.hpp"
#include "caf/actor_profiler.hpp"
#include "caf/config_option.hpp"
#include "caf/config_option_adder.hpp"
#include "caf/config_option_set.hpp"
......@@ -300,6 +301,11 @@ public:
thread_hooks thread_hooks_;
/// Provides system-wide callbacks for several actor operations.
/// @experimental
/// @note Has no effect unless building CAF with CAF_ENABLE_ACTOR_PROFILER.
actor_profiler* profiler = nullptr;
// -- run-time type information ----------------------------------------------
portable_name_map type_names_by_rtti;
......
......@@ -88,6 +88,7 @@ class actor_companion;
class actor_config;
class actor_control_block;
class actor_pool;
class actor_profiler;
class actor_proxy;
class actor_registry;
class actor_system;
......@@ -177,6 +178,7 @@ enum class atom_value : uint64_t;
enum class byte : uint8_t;
enum class sec : uint8_t;
enum class stream_priority;
enum class invoke_message_result;
// -- aliases ------------------------------------------------------------------
......
......@@ -22,17 +22,21 @@
namespace caf {
enum invoke_message_result {
im_success,
im_skipped,
im_dropped
/// Stores the result of a message invocation.
enum class invoke_message_result {
/// Indicates that the actor consumed the message.
consumed,
/// Indicates that the actor left the message in the mailbox.
skipped,
/// Indicates that the actor discarded the message based on meta data. For
/// example, timeout messages for already received requests usually get
/// dropped without calling any user-defined code.
dropped,
};
inline std::string to_string(invoke_message_result x) {
return x == im_success ? "im_success"
: (x == im_skipped ? "im_skipped" : "im_dropped" );
}
/// @relates invoke_message_result
std::string to_string(invoke_message_result);
} // namespace caf
......@@ -96,15 +96,17 @@ public:
template <class T, spawn_options Os = no_spawn_options, class... Ts>
infer_handle_from_class_t<T> spawn(Ts&&... xs) {
actor_config cfg{context()};
return eval_opts(Os, system().spawn_class<T, make_unbound(Os)>(
cfg, std::forward<Ts>(xs)...));
actor_config cfg{context(), this};
return eval_opts(Os,
system().spawn_class<T, make_unbound(Os)>(cfg,
std::forward<Ts>(
xs)...));
}
template <class T, spawn_options Os = no_spawn_options>
infer_handle_from_state_t<T> spawn() {
using impl = composable_behavior_based_actor<T>;
actor_config cfg{context()};
actor_config cfg{context(), this};
return eval_opts(Os, system().spawn_class<impl, make_unbound(Os)>(cfg));
}
......@@ -114,7 +116,7 @@ public:
static constexpr bool spawnable = detail::spawnable<F, impl, Ts...>();
static_assert(spawnable,
"cannot spawn function-based actor with given arguments");
actor_config cfg{context()};
actor_config cfg{context(), this};
static constexpr spawn_options unbound = make_unbound(Os);
detail::bool_token<spawnable> enabled;
return eval_opts(Os,
......@@ -125,50 +127,65 @@ public:
template <class T, spawn_options Os = no_spawn_options, class Groups,
class... Ts>
actor spawn_in_groups(const Groups& gs, Ts&&... xs) {
actor_config cfg{context()};
return eval_opts(Os, system().spawn_class_in_groups<T, make_unbound(Os)>(
cfg, gs.begin(), gs.end(), std::forward<Ts>(xs)...));
actor_config cfg{context(), this};
return eval_opts(Os, system()
.spawn_class_in_groups<
T, make_unbound(Os)>(cfg, gs.begin(), gs.end(),
std::forward<Ts>(xs)...));
}
template <class T, spawn_options Os = no_spawn_options, class... Ts>
actor spawn_in_groups(std::initializer_list<group> gs, Ts&&... xs) {
actor_config cfg{context()};
return eval_opts(Os, system().spawn_class_in_groups<T, make_unbound(Os)>(
cfg, gs.begin(), gs.end(), std::forward<Ts>(xs)...));
actor_config cfg{context(), this};
return eval_opts(Os, system()
.spawn_class_in_groups<
T, make_unbound(Os)>(cfg, gs.begin(), gs.end(),
std::forward<Ts>(xs)...));
}
template <class T, spawn_options Os = no_spawn_options, class... Ts>
actor spawn_in_group(const group& grp, Ts&&... xs) {
actor_config cfg{context()};
actor_config cfg{context(), this};
auto first = &grp;
return eval_opts(Os, system().spawn_class_in_groups<T, make_unbound(Os)>(
cfg, first, first + 1, std::forward<Ts>(xs)...));
return eval_opts(Os, system()
.spawn_class_in_groups<
T, make_unbound(Os)>(cfg, first, first + 1,
std::forward<Ts>(xs)...));
}
template <spawn_options Os = no_spawn_options, class Groups, class F,
class... Ts>
actor spawn_in_groups(const Groups& gs, F fun, Ts&&... xs) {
actor_config cfg{context()};
return eval_opts(
Os, system().spawn_fun_in_groups<make_unbound(Os)>(
cfg, gs.begin(), gs.end(), fun, std::forward<Ts>(xs)...));
actor_config cfg{context(), this};
return eval_opts(Os,
system()
.spawn_fun_in_groups<make_unbound(Os)>(cfg, gs.begin(),
gs.end(), fun,
std::forward<Ts>(
xs)...));
}
template <spawn_options Os = no_spawn_options, class F, class... Ts>
actor spawn_in_groups(std::initializer_list<group> gs, F fun, Ts&&... xs) {
actor_config cfg{context()};
return eval_opts(
Os, system().spawn_fun_in_groups<make_unbound(Os)>(
cfg, gs.begin(), gs.end(), fun, std::forward<Ts>(xs)...));
actor_config cfg{context(), this};
return eval_opts(Os,
system()
.spawn_fun_in_groups<make_unbound(Os)>(cfg, gs.begin(),
gs.end(), fun,
std::forward<Ts>(
xs)...));
}
template <spawn_options Os = no_spawn_options, class F, class... Ts>
actor spawn_in_group(const group& grp, F fun, Ts&&... xs) {
actor_config cfg{context()};
actor_config cfg{context(), this};
auto first = &grp;
return eval_opts(Os,
system().spawn_fun_in_groups<make_unbound(Os)>(
cfg, first, first + 1, fun, std::forward<Ts>(xs)...));
system()
.spawn_fun_in_groups<make_unbound(Os)>(cfg, first,
first + 1, fun,
std::forward<Ts>(
xs)...));
}
// -- sending asynchronous messages ------------------------------------------
......@@ -290,7 +307,7 @@ public:
/// Returns a pointer to the currently processed mailbox element.
/// @private
inline void current_mailbox_element(mailbox_element* ptr) {
inline void current_mailbox_element(mailbox_element* ptr) {
current_element_ = ptr;
}
......@@ -333,10 +350,8 @@ public:
/// Return type is deduced from arguments.
/// Return value is implicitly convertible to untyped response promise.
template <class... Ts,
class R =
typename detail::make_response_promise_helper<
typename std::decay<Ts>::type...
>::type>
class R = typename detail::make_response_promise_helper<
typename std::decay<Ts>::type...>::type>
R response(Ts&&... xs) {
auto promise = make_response_promise<R>();
promise.deliver(std::forward<Ts>(xs)...);
......@@ -382,27 +397,23 @@ public:
return (mid.is_request()) ? mid.response_id() : message_id();
}
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");
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),
? 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 {};
......
This diff is collapsed.
......@@ -22,8 +22,9 @@
namespace caf {
actor_config::actor_config(execution_unit* ptr)
: host(ptr),
actor_config::actor_config(execution_unit* host, local_actor* parent)
: host(host),
parent(parent),
flags(abstract_channel::is_abstract_actor_flag),
groups(nullptr) {
// nop
......@@ -33,12 +34,9 @@ std::string to_string(const actor_config& x) {
// Note: x.groups is an input range. Traversing it is emptying it, hence we
// cannot look inside the range here.
std::string result = "actor_config(";
bool first = false;
auto add = [&](int flag, const char* name) {
if ((x.flags & flag) != 0) {
if (first)
first = false;
else
if (result.back() != '(')
result += ", ";
result += name;
}
......@@ -48,7 +46,7 @@ std::string to_string(const actor_config& x) {
add(abstract_actor::is_detached_flag, "detached_flag");
add(abstract_actor::is_blocking_flag, "blocking_flag");
add(abstract_actor::is_hidden_flag, "hidden_flag");
result += ")";
result += ')';
return result;
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/actor_profiler.hpp"
namespace caf {
actor_profiler::~actor_profiler() {
// nop
}
} // namespace caf
......@@ -216,16 +216,17 @@ const char* actor_system::module::name() const noexcept {
}
actor_system::actor_system(actor_system_config& cfg)
: ids_(0),
types_(*this),
logger_(new caf::logger(*this), false),
registry_(*this),
groups_(*this),
dummy_execution_unit_(this),
await_actors_before_shutdown_(true),
detached_(0),
cfg_(cfg),
logger_dtor_done_(false) {
: profiler_(cfg.profiler),
ids_(0),
types_(*this),
logger_(new caf::logger(*this), false),
registry_(*this),
groups_(*this),
dummy_execution_unit_(this),
await_actors_before_shutdown_(true),
detached_(0),
cfg_(cfg),
logger_dtor_done_(false) {
CAF_SET_LOGGER_SYS(this);
for (auto& hook : cfg.thread_hooks_)
hook->init(*this);
......
......@@ -26,6 +26,7 @@
#include "caf/detail/invoke_result_visitor.hpp"
#include "caf/detail/set_thread_name.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/invoke_message_result.hpp"
#include "caf/logger.hpp"
namespace caf {
......@@ -156,67 +157,75 @@ void blocking_actor::fail_state(error err) {
}
intrusive::task_result
blocking_actor::mailbox_visitor:: operator()(mailbox_element& x) {
blocking_actor::mailbox_visitor::operator()(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x));
CAF_LOG_RECEIVE_EVENT((&x));
auto check_if_done = [&]() -> intrusive::task_result {
// Stop consuming items when reaching the end of the user-defined receive
// loop either via post or pre condition.
if (rcc.post() && rcc.pre())
return intrusive::task_result::resume;
done = true;
return intrusive::task_result::stop;
};
// Skip messages that don't match our message ID.
if (mid.is_response()) {
if (mid != x.mid) {
CAF_LOG_SKIP_EVENT();
CAF_BEFORE_PROCESSING(self, x);
// Wrap the actual body for the function.
auto body = [this, &x] {
auto check_if_done = [&]() -> intrusive::task_result {
// Stop consuming items when reaching the end of the user-defined receive
// loop either via post or pre condition.
if (rcc.post() && rcc.pre())
return intrusive::task_result::resume;
done = true;
return intrusive::task_result::stop;
};
// Skip messages that don't match our message ID.
if (mid.is_response()) {
if (mid != x.mid) {
return intrusive::task_result::skip;
}
} else if (x.mid.is_response()) {
return intrusive::task_result::skip;
}
} else if (x.mid.is_response()) {
CAF_LOG_SKIP_EVENT();
return intrusive::task_result::skip;
}
// Automatically unlink from actors after receiving an exit.
if (x.content().match_elements<exit_msg>())
self->unlink_from(x.content().get_as<exit_msg>(0).source);
// Blocking actors can nest receives => push/pop `current_element_`
auto prev_element = self->current_element_;
self->current_element_ = &x;
auto g = detail::make_scope_guard([&] {
self->current_element_ = prev_element;
});
// Dispatch on x.
detail::default_invoke_result_visitor<blocking_actor> visitor{self};
switch (bhvr.nested(visitor, x.content())) {
default:
return check_if_done();
case match_case::no_match:
{ // Blocking actors can have fallback handlers for catch-all rules.
// Automatically unlink from actors after receiving an exit.
if (x.content().match_elements<exit_msg>())
self->unlink_from(x.content().get_as<exit_msg>(0).source);
// Blocking actors can nest receives => push/pop `current_element_`
auto prev_element = self->current_element_;
self->current_element_ = &x;
auto g = detail::make_scope_guard(
[&] { self->current_element_ = prev_element; });
// Dispatch on x.
detail::default_invoke_result_visitor<blocking_actor> visitor{self};
switch (bhvr.nested(visitor, x.content())) {
default:
return check_if_done();
case match_case::no_match: { // Blocking actors can have fallback handlers
// for catch-all rules.
auto sres = bhvr.fallback(*self->current_element_);
if (sres.flag != rt_skip) {
visitor.visit(sres);
CAF_LOG_FINALIZE_EVENT();
return check_if_done();
}
}
// Response handlers must get re-invoked with an error when receiving an
// unexpected message.
if (mid.is_response()) {
auto err = make_error(sec::unexpected_response,
x.move_content_to_message());
mailbox_element_view<error> tmp{std::move(x.sender), x.mid,
std::move(x.stages), err};
self->current_element_ = &tmp;
bhvr.nested(tmp.content());
CAF_LOG_FINALIZE_EVENT();
return check_if_done();
}
CAF_ANNOTATE_FALLTHROUGH;
case match_case::skip:
CAF_LOG_SKIP_EVENT();
return intrusive::task_result::skip;
// Response handlers must get re-invoked with an error when receiving an
// unexpected message.
if (mid.is_response()) {
auto err = make_error(sec::unexpected_response,
x.move_content_to_message());
mailbox_element_view<error> tmp{std::move(x.sender), x.mid,
std::move(x.stages), err};
self->current_element_ = &tmp;
bhvr.nested(tmp.content());
return check_if_done();
}
CAF_ANNOTATE_FALLTHROUGH;
case match_case::skip:
return intrusive::task_result::skip;
}
};
// Post-process the returned value from the function body.
auto result = body();
if (result == intrusive::task_result::skip) {
CAF_AFTER_PROCESSING(self, invoke_message_result::skipped);
CAF_LOG_SKIP_EVENT();
} else {
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
CAF_LOG_FINALIZE_EVENT();
}
return result;
}
void blocking_actor::receive_impl(receive_cond& rcc,
......
......@@ -18,29 +18,29 @@
#include "caf/local_actor.hpp"
#include <string>
#include <condition_variable>
#include <string>
#include "caf/sec.hpp"
#include "caf/atom.hpp"
#include "caf/logger.hpp"
#include "caf/scheduler.hpp"
#include "caf/resumable.hpp"
#include "caf/actor_cast.hpp"
#include "caf/exit_reason.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/actor_system.hpp"
#include "caf/atom.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/default_attachable.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/exit_reason.hpp"
#include "caf/logger.hpp"
#include "caf/resumable.hpp"
#include "caf/scheduler.hpp"
#include "caf/sec.hpp"
namespace caf {
local_actor::local_actor(actor_config& cfg)
: monitorable_actor(cfg),
context_(cfg.host),
current_element_(nullptr),
initial_behavior_fac_(std::move(cfg.init_fun)) {
: monitorable_actor(cfg),
context_(cfg.host),
current_element_(nullptr),
initial_behavior_fac_(std::move(cfg.init_fun)) {
// nop
}
......@@ -50,6 +50,9 @@ local_actor::~local_actor() {
void local_actor::on_destroy() {
CAF_PUSH_AID_FROM_PTR(this);
#ifdef CAF_ENABLE_ACTOR_PROFILER
system().profiler_remove_actor(*this);
#endif
if (!getf(is_cleaned_up_flag)) {
on_exit();
cleanup(exit_reason::unreachable, nullptr);
......@@ -68,15 +71,16 @@ void local_actor::request_response_timeout(const duration& d, message_id mid) {
void local_actor::monitor(abstract_actor* ptr, message_priority priority) {
if (ptr != nullptr)
ptr->attach(default_attachable::make_monitor(ptr->address(), address(),
priority));
ptr->attach(
default_attachable::make_monitor(ptr->address(), address(), priority));
}
void local_actor::demonitor(const actor_addr& whom) {
CAF_LOG_TRACE(CAF_ARG(whom));
auto ptr = actor_cast<strong_actor_ptr>(whom);
if (ptr) {
default_attachable::observe_token tk{address(), default_attachable::monitor};
default_attachable::observe_token tk{address(),
default_attachable::monitor};
ptr->get()->detach(tk);
}
}
......
......@@ -23,7 +23,7 @@
namespace caf {
raw_event_based_actor::raw_event_based_actor(actor_config& cfg)
: event_based_actor(cfg) {
: event_based_actor(cfg) {
// nop
}
......@@ -31,93 +31,102 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x));
current_element_ = &x;
CAF_LOG_RECEIVE_EVENT(current_element_);
// short-circuit awaited responses
if (!awaited_responses_.empty()) {
auto& pr = awaited_responses_.front();
// skip all messages until we receive the currently awaited response
if (x.mid != pr.first)
return im_skipped;
if (!pr.second(x.content())) {
// try again with error if first attempt failed
auto msg = make_message(make_error(sec::unexpected_response,
x.move_content_to_message()));
pr.second(msg);
CAF_BEFORE_PROCESSING(this, x);
// Wrap the actual body for the function.
auto body = [this, &x] {
// short-circuit awaited responses
if (!awaited_responses_.empty()) {
auto& pr = awaited_responses_.front();
// skip all messages until we receive the currently awaited response
if (x.mid != pr.first)
return invoke_message_result::skipped;
if (!pr.second(x.content())) {
// try again with error if first attempt failed
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
pr.second(msg);
}
awaited_responses_.pop_front();
return invoke_message_result::consumed;
}
awaited_responses_.pop_front();
return im_success;
}
// handle multiplexed responses
if (x.mid.is_response()) {
auto mrh = multiplexed_responses_.find(x.mid);
// neither awaited nor multiplexed, probably an expired timeout
if (mrh == multiplexed_responses_.end())
return im_dropped;
if (!mrh->second(x.content())) {
// try again with error if first attempt failed
auto msg = make_message(make_error(sec::unexpected_response,
x.move_content_to_message()));
mrh->second(msg);
// handle multiplexed responses
if (x.mid.is_response()) {
auto mrh = multiplexed_responses_.find(x.mid);
// neither awaited nor multiplexed, probably an expired timeout
if (mrh == multiplexed_responses_.end())
return invoke_message_result::dropped;
if (!mrh->second(x.content())) {
// try again with error if first attempt failed
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
mrh->second(msg);
}
multiplexed_responses_.erase(mrh);
return invoke_message_result::consumed;
}
multiplexed_responses_.erase(mrh);
return im_success;
}
auto& content = x.content();
// handle timeout messages
if (x.content().type_token() == make_type_token<timeout_msg>()) {
auto& tm = content.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
CAF_ASSERT(x.mid.is_async());
if (is_active_receive_timeout(tid)) {
CAF_LOG_DEBUG("handle timeout message");
if (bhvr_stack_.empty())
return im_dropped;
bhvr_stack_.back().handle_timeout();
return im_success;
auto& content = x.content();
// handle timeout messages
if (x.content().type_token() == make_type_token<timeout_msg>()) {
auto& tm = content.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
CAF_ASSERT(x.mid.is_async());
if (is_active_receive_timeout(tid)) {
CAF_LOG_DEBUG("handle timeout message");
if (bhvr_stack_.empty())
return invoke_message_result::dropped;
bhvr_stack_.back().handle_timeout();
return invoke_message_result::consumed;
}
CAF_LOG_DEBUG("dropped expired timeout message");
return invoke_message_result::dropped;
}
CAF_LOG_DEBUG("dropped expired timeout message");
return im_dropped;
}
// handle everything else as ordinary message
detail::default_invoke_result_visitor<event_based_actor> visitor{this};
bool skipped = false;
auto had_timeout = getf(has_timeout_flag);
if (had_timeout)
unsetf(has_timeout_flag);
// restore timeout at scope exit if message was skipped
auto timeout_guard = detail::make_scope_guard([&] {
if (skipped && had_timeout)
setf(has_timeout_flag);
});
auto call_default_handler = [&] {
auto sres = call_handler(default_handler_, this, x);
switch (sres.flag) {
// handle everything else as ordinary message
detail::default_invoke_result_visitor<event_based_actor> visitor{this};
bool skipped = false;
auto had_timeout = getf(has_timeout_flag);
if (had_timeout)
unsetf(has_timeout_flag);
// restore timeout at scope exit if message was skipped
auto timeout_guard = detail::make_scope_guard([&] {
if (skipped && had_timeout)
setf(has_timeout_flag);
});
auto call_default_handler = [&] {
auto sres = call_handler(default_handler_, this, x);
switch (sres.flag) {
default:
break;
case rt_error:
case rt_value:
visitor.visit(sres);
break;
case rt_skip:
skipped = true;
}
};
if (bhvr_stack_.empty()) {
call_default_handler();
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
}
auto& bhvr = bhvr_stack_.back();
switch (bhvr(visitor, x.content())) {
default:
break;
case rt_error:
case rt_value:
visitor.visit(sres);
break;
case rt_skip:
case match_case::skip:
skipped = true;
break;
case match_case::no_match:
call_default_handler();
}
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
};
if (bhvr_stack_.empty()) {
call_default_handler();
return !skipped ? im_success : im_skipped;
}
auto& bhvr = bhvr_stack_.back();
switch (bhvr(visitor, x.content())) {
default:
break;
case match_case::skip:
skipped = true;
break;
case match_case::no_match:
call_default_handler();
}
return !skipped ? im_success : im_skipped;
// should be unreachable
CAF_CRITICAL("invalid message type");
// Post-process the returned value from the function body.
auto result = body();
CAF_AFTER_PROCESSING(this, result);
CAF_LOG_SKIP_OR_FINALIZE_EVENT(result);
return result;
}
} // namespace caf
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE actor_profiler
#include "caf/actor_profiler.hpp"
#include "caf/test/dsl.hpp"
#include "caf/config.hpp"
#ifdef CAF_ENABLE_ACTOR_PROFILER
using namespace caf;
namespace {
using string_list = std::vector<std::string>;
struct recorder : actor_profiler {
void add_actor(const local_actor& self, const local_actor* parent) {
log.emplace_back("new: ");
auto& str = log.back();
str += self.name();
if (parent != nullptr) {
str += ", parent: ";
str += parent->name();
}
}
void remove_actor(const local_actor& self) {
log.emplace_back("delete: ");
log.back() += self.name();
}
void before_processing(const local_actor& self,
const mailbox_element& element) {
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) {
log.emplace_back(self.name());
auto& str = log.back();
str += " ";
str += to_string(result);
str += " the message";
}
string_list log;
};
actor_system_config& init(actor_system_config& cfg, recorder& rec) {
test_coordinator_fixture<>::init_config(cfg);
cfg.profiler = &rec;
return cfg;
}
struct fixture {
using scheduler_type = caf::scheduler::test_coordinator;
fixture()
: sys(init(cfg, rec)),
sched(dynamic_cast<scheduler_type&>(sys.scheduler())) {
// nop
}
void run() {
sched.run();
}
recorder rec;
actor_system_config cfg;
actor_system sys;
scheduler_type& sched;
};
struct foo_state {
const char* name = "foo";
};
struct bar_state {
const char* name = "bar";
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(actor_profiler_tests, fixture)
CAF_TEST(record actor construction) {
CAF_MESSAGE("fully initialize CAF, ignore system-internal actors");
run();
rec.log.clear();
CAF_MESSAGE("spawn a foo and a bar");
auto bar = [](stateful_actor<bar_state>*) {};
auto foo = [bar](stateful_actor<foo_state>* self) { self->spawn(bar); };
auto foo_actor = sys.spawn(foo);
run();
foo_actor = nullptr;
CAF_CHECK_EQUAL(string_list({
"new: foo",
"new: bar, parent: foo",
"delete: bar",
"delete: foo",
}),
rec.log);
}
CAF_TEST(record actor messaging) {
CAF_MESSAGE("fully initialize CAF, ignore system-internal actors");
run();
rec.log.clear();
CAF_MESSAGE("spawn a foo and a bar");
auto bar = [](stateful_actor<bar_state>*) -> behavior {
return {
[](const std::string& str) {
CAF_CHECK_EQUAL(str, "hello bar");
return "hello foo";
},
};
};
auto foo = [bar](stateful_actor<foo_state>* self) -> behavior {
auto b = self->spawn(bar);
self->send(b, "hello bar");
return {
[](const std::string& str) { CAF_CHECK_EQUAL(str, "hello foo"); },
};
};
sys.spawn(foo);
run();
CAF_CHECK_EQUAL(string_list({
"new: foo",
"new: bar, parent: foo",
"bar got: (\"hello bar\")",
"bar consumed the message",
"foo got: (\"hello foo\")",
"delete: bar",
"foo consumed the message",
"delete: foo",
}),
rec.log);
}
CAF_TEST_FIXTURE_SCOPE_END()
#endif // CAF_ENABLE_ACTOR_PROFILER
......@@ -60,13 +60,13 @@ void manager::detach(execution_unit*, bool invoke_disconnect_message) {
auto mptr = make_mailbox_element(nullptr, make_message_id(), {},
detach_message());
switch (raw_ptr->consume(*mptr)) {
case im_success:
case invoke_message_result::consumed:
raw_ptr->finalize();
break;
case im_skipped:
case invoke_message_result::skipped:
raw_ptr->push_to_cache(std::move(mptr));
break;
case im_dropped:
case invoke_message_result::dropped:
CAF_LOG_INFO("broker dropped disconnect message");
break;
}
......
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