Commit 08603f37 authored by Dominik Charousset's avatar Dominik Charousset

Integrate optional actor metrics

parent 951b9e15
......@@ -82,6 +82,7 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS}
src/detail/get_mac_addresses.cpp
src/detail/get_process_id.cpp
src/detail/get_root_uuid.cpp
src/detail/glob_match.cpp
src/detail/ini_consumer.cpp
src/detail/invoke_result_visitor.cpp
src/detail/message_builder_element.cpp
......
......@@ -133,6 +133,7 @@ public:
static constexpr int is_initialized_flag = 0x0010; // event-based actors
static constexpr int is_blocking_flag = 0x0020; // blocking_actor
static constexpr int is_detached_flag = 0x0040; // local_actor
static constexpr int collects_metrics_flag = 0x0080; // local_actor
static constexpr int is_serializable_flag = 0x0100; // local_actor
static constexpr int is_migrated_from_flag = 0x0200; // local_actor
static constexpr int has_used_aout_flag = 0x0400; // local_actor
......
......@@ -105,6 +105,10 @@ public:
void on_destroy() override;
void setup_metrics() {
// nop
}
protected:
void on_cleanup(const error& reason) override;
......
......@@ -38,6 +38,10 @@ public:
/// Invokes cleanup code.
virtual void kill_proxy(execution_unit* ctx, error reason) = 0;
void setup_metrics() {
// nop
}
};
} // namespace caf
......@@ -173,7 +173,7 @@ public:
virtual void demonitor(const node_id& node, const actor_addr& observer) = 0;
};
/// Metrics collected by the actor system by default.
/// Metrics that the actor system collects by default.
/// @warning Do not modify these metrics in user code. Some may be used by the
/// system for synchronization.
struct base_metrics_t {
......@@ -186,6 +186,24 @@ public:
/// Tracks the current number of running actors in the system.
telemetry::int_gauge* running_actors;
/// Counts the total number of messages that wait in a mailbox.
telemetry::int_gauge* queued_messages;
};
/// Metrics that some actors may collect in addition to the base metrics. All
/// families in this set use the label dimension *name* (the user-defined name
/// of the actor).
struct actor_metric_families_t {
/// Samples how long the actor needs to process messages.
telemetry::dbl_histogram_family* processing_time_family = nullptr;
/// Samples how long a message waits in the mailbox before the actor
/// processes it.
telemetry::dbl_histogram_family* mailbox_time_family = nullptr;
/// Counts how many messages are currently waiting in the mailbox.
telemetry::int_gauge_family* mailbox_size_family = nullptr;
};
/// @warning The system stores a reference to `cfg`, which means the
......@@ -418,8 +436,8 @@ public:
/// range `[first, last)`.
/// @private
template <spawn_options Os, class Iter, class F, class... Ts>
infer_handle_from_fun_t<F>
spawn_fun_in_groups(actor_config& cfg, Iter first, Iter second, F& fun,
infer_handle_from_fun_t<F> spawn_fun_in_groups(actor_config& cfg, Iter first,
Iter second, F& fun,
Ts&&... xs) {
using impl = infer_impl_from_fun_t<F>;
check_invariants<impl>();
......@@ -527,6 +545,14 @@ public:
/// @warning must be called by thread which is about to terminate
void thread_terminates();
const auto& metrics_actors_includes() const noexcept {
return metrics_actors_includes_;
}
const auto& metrics_actors_excludes() const noexcept {
return metrics_actors_excludes_;
}
template <class C, spawn_options Os, class... Ts>
infer_handle_from_class_t<C> spawn_impl(actor_config& cfg, Ts&&... xs) {
static_assert(is_unbound(Os),
......@@ -571,8 +597,8 @@ public:
profiler_->after_processing(self, result);
}
void
profiler_before_sending(const local_actor& self, mailbox_element& element) {
void profiler_before_sending(const local_actor& self,
mailbox_element& element) {
if (profiler_)
profiler_->before_sending(self, element);
}
......@@ -588,10 +614,14 @@ public:
return base_metrics_;
}
const base_metrics_t& base_metrics() const noexcept {
const auto& base_metrics() const noexcept {
return base_metrics_;
}
const auto& actor_metric_families() const noexcept {
return actor_metric_families_;
}
tracing_data_factory* tracing_context() const noexcept {
return tracing_context_;
}
......@@ -685,6 +715,17 @@ private:
/// Stores the system-wide factory for deserializing tracing data.
tracing_data_factory* tracing_context_;
/// Caches the configuration parameter `caf.metrics-filters.actors.includes`
/// for faster lookups at runtime.
std::vector<std::string> metrics_actors_includes_;
/// Caches the configuration parameter `caf.metrics-filters.actors.excludes`
/// for faster lookups at runtime.
std::vector<std::string> metrics_actors_excludes_;
/// Caches families for optional actor metrics.
actor_metric_families_t actor_metric_families_;
};
} // namespace caf
......@@ -46,6 +46,10 @@ public:
message_types_set message_types() const override;
void setup_metrics() {
// nop
}
protected:
void on_cleanup(const error& reason) override;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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::detail {
// gitignore-style pathname globbing.
bool glob_match(const char* str, const char* glob);
} // namespace caf::detail
......@@ -262,6 +262,13 @@ using dbl_histogram = histogram<double>;
using int_counter = counter<int64_t>;
using int_histogram = histogram<int64_t>;
using dbl_counter_family = metric_family_impl<dbl_counter>;
using dbl_histogram_family = metric_family_impl<dbl_histogram>;
using dbl_gauge_family = metric_family_impl<dbl_gauge>;
using int_counter_family = metric_family_impl<int_counter>;
using int_histogram_family = metric_family_impl<int_histogram>;
using int_gauge_family = metric_family_impl<int_gauge>;
} // namespace telemetry
namespace detail {
......
......@@ -49,6 +49,7 @@
#include "caf/response_type.hpp"
#include "caf/resumable.hpp"
#include "caf/spawn_options.hpp"
#include "caf/telemetry/histogram.hpp"
#include "caf/timespan.hpp"
#include "caf/typed_actor.hpp"
#include "caf/typed_response_promise.hpp"
......@@ -64,6 +65,18 @@ public:
/// Defines a monotonic clock suitable for measuring intervals.
using clock_type = std::chrono::steady_clock;
/// Optional metrics collected by individual actors when configured to do so.
struct metrics_t {
/// Samples how long the actor needs to process messages.
telemetry::dbl_histogram* processing_time = nullptr;
/// Samples how long messages wait in the mailbox before being processed.
telemetry::dbl_histogram* mailbox_time = nullptr;
/// Counts how many messages are currently waiting in the mailbox.
telemetry::int_gauge* mailbox_size = nullptr;
};
// -- constructors, destructors, and assignment operators --------------------
local_actor(actor_config& cfg);
......@@ -72,6 +85,11 @@ public:
void on_destroy() override;
// This function performs additional steps to initialize actor-specific
// metrics. It calls virtual functions and thus cannot run as part of the
// constructor.
void setup_metrics();
// -- pure virtual modifiers -------------------------------------------------
virtual void launch(execution_unit* eu, bool lazy, bool hide) = 0;
......@@ -360,6 +378,10 @@ public:
/// @cond PRIVATE
auto& builtin_metrics() noexcept {
return metrics_;
}
template <class ActorHandle>
ActorHandle eval_opts(spawn_options opts, ActorHandle res) {
if (has_monitor_flag(opts))
......@@ -408,6 +430,8 @@ protected:
/// Factory function for returning initial behavior in function-based actors.
detail::unique_function<behavior(local_actor*)> initial_behavior_fac_;
metrics_t metrics_;
};
} // namespace caf
......@@ -18,6 +18,7 @@
#pragma once
#include <chrono>
#include <cstddef>
#include <memory>
......@@ -57,6 +58,27 @@ public:
/// Stores the payload.
message payload;
/// Stores a timestamp for when this element got enqueued.
std::chrono::steady_clock::time_point enqueue_time;
/// Sets `enqueue_time` to the current time.
void set_enqueue_time() {
enqueue_time = std::chrono::steady_clock::now();
}
/// Returns the time between enqueueing the message and `t`.
double seconds_until(std::chrono::steady_clock::time_point t) const {
namespace ch = std::chrono;
using fractional_seconds = ch::duration<double>;
return ch::duration_cast<fractional_seconds>(t - enqueue_time).count();
}
/// Returns the time since calling `set_enqueue_time` in seconds.
double seconds_since_enqueue() const {
namespace ch = std::chrono;
return seconds_until(std::chrono::steady_clock::now());
}
mailbox_element() = default;
mailbox_element(strong_actor_ptr sender, message_id mid,
......
......@@ -44,12 +44,14 @@ R make_actor(actor_id aid, node_id nid, actor_system* sys, Ts&&... xs) {
std::forward<Ts>(xs)...);
}
CAF_LOG_SPAWN_EVENT(ptr->data, args);
ptr->data.setup_metrics();
return {&(ptr->ctrl), false};
}
#endif
CAF_PUSH_AID(aid);
auto ptr = new actor_storage<T>(aid, std::move(nid), sys,
std::forward<Ts>(xs)...);
ptr->data.setup_metrics();
return {&(ptr->ctrl), false};
}
......
......@@ -60,6 +60,7 @@
#include "caf/scheduled_actor.hpp"
#include "caf/sec.hpp"
#include "caf/stream_manager.hpp"
#include "caf/telemetry/timer.hpp"
#include "caf/to_string.hpp"
namespace caf {
......@@ -197,6 +198,7 @@ public:
scheduled_actor* self;
size_t& handled_msgs;
size_t max_throughput;
bool collect_metrics;
/// Consumes upstream messages.
intrusive::task_result operator()(size_t, upstream_queue&,
......@@ -217,6 +219,24 @@ public:
// Consumes asynchronous messages.
intrusive::task_result operator()(mailbox_element& x);
template <class F>
intrusive::task_result run(mailbox_element& x, F body) {
if (collect_metrics) {
auto t0 = std::chrono::steady_clock::now();
auto mbox_time = x.seconds_until(t0);
auto res = body();
if (res != intrusive::task_result::skip) {
auto& builtins = self->builtin_metrics();
telemetry::timer::observe(builtins.processing_time, t0);
builtins.mailbox_time->observe(mbox_time);
builtins.mailbox_size->dec();
}
return res;
} else {
return body();
}
}
};
// -- static helper functions ------------------------------------------------
......
......@@ -79,14 +79,16 @@ public:
const char* name() const override {
if constexpr (detail::has_name<State>::value) {
if constexpr (std::is_convertible<decltype(state.name),
const char*>::value)
return state.name;
else
return state.name.c_str();
if constexpr (!std::is_member_pointer<decltype(&State::name)>::value) {
if constexpr (std::is_convertible<decltype(State::name),
const char*>::value) {
return State::name;
}
} else {
return Base::name();
non_static_name_member(state.name);
}
}
return Base::name();
}
union {
......@@ -95,6 +97,14 @@ public:
/// its reference count drops to zero.
State state;
};
template <class T>
[[deprecated("non-static 'State::name' members have no effect since 0.18")]]
// This function only exists to raise a deprecated warning.
static void
non_static_name_member(const T&) {
// nop
}
};
} // namespace caf
......
......@@ -37,11 +37,8 @@ public:
}
~timer() {
using dbl_sec = std::chrono::duration<double>;
if (h_) {
auto end = clock_type::now();
h_->observe(std::chrono::duration_cast<dbl_sec>(end - start_).count());
}
if (h_)
observe(h_, start_);
}
auto histogram_ptr() const noexcept {
......@@ -52,6 +49,12 @@ public:
return start_;
}
static void observe(dbl_histogram* h, clock_type::time_point start) {
using dbl_sec = std::chrono::duration<double>;
auto end = clock_type::now();
h->observe(std::chrono::duration_cast<dbl_sec>(end - start).count());
}
private:
dbl_histogram* h_;
clock_type::time_point start_;
......
......@@ -46,11 +46,9 @@ struct kvstate {
using topic_set = std::unordered_set<std::string>;
std::unordered_map<key_type, std::pair<mapped_type, subscriber_set>> data;
std::unordered_map<strong_actor_ptr, topic_set> subscribers;
static const char* name;
static inline const char* name = "caf.system.config-server";
};
const char* kvstate::name = "config_server";
behavior config_serv_impl(stateful_actor<kvstate>* self) {
CAF_LOG_TRACE("");
std::string wildcard = "*";
......@@ -148,11 +146,9 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) {
// on another node, users can spwan actors remotely.
struct spawn_serv_state {
static const char* name;
static inline const char* name = "caf.system.spawn-server";
};
const char* spawn_serv_state::name = "spawn_server";
behavior spawn_serv_impl(stateful_actor<spawn_serv_state>* self) {
CAF_LOG_TRACE("");
return {
......@@ -200,13 +196,13 @@ actor_system::module::~module() {
const char* actor_system::module::name() const noexcept {
switch (id()) {
case scheduler:
return "Scheduler";
return "scheduler";
case middleman:
return "Middleman";
return "middleman";
case openssl_manager:
return "OpenSSL Manager";
return "openssl-manager";
case network_manager:
return "Network Manager";
return "metwork-manager";
default:
return "???";
}
......@@ -221,12 +217,43 @@ namespace {
auto make_base_metrics(telemetry::metric_registry& reg) {
return actor_system::base_metrics_t{
// Initialize the base metrics.
reg.counter_singleton("caf", "rejected-messages",
reg.counter_singleton("caf.system", "rejected-messages",
"Number of rejected messages.", "1", true),
reg.counter_singleton("caf", "processed-messages",
reg.counter_singleton("caf.system", "processed-messages",
"Number of processed messages.", "1", true),
reg.gauge_singleton("caf", "running-actors",
reg.gauge_singleton("caf.system", "running-actors",
"Number of currently running actors."),
reg.gauge_singleton("caf.system", "queued-messages",
"Number of messages in all mailboxes.", "1", true),
};
}
auto make_actor_metric_families(telemetry::metric_registry& reg) {
// Handling a single message generally should take microseconds. Going up to
// several milliseconds usually indicates a problem (or blocking operations)
// but may still be expected for very compute-intense tasks. Single messages
// that approach seconds to process most likely indicate a severe issue.
// Hence, the default bucket settings focus on micro- and milliseconds.
std::array<double, 9> default_buckets{{
.00001, // 10us
.0001, // 100us
.0005, // 500us
.001, // 1ms
.01, // 10ms
.1, // 100ms
.5, // 500ms
1., // 1s
5., // 5s
}};
return actor_system::actor_metric_families_t{
reg.histogram_family<double>(
"caf.actor", "processing-time", {"name"}, default_buckets,
"Time an actor needs to process messages.", "seconds"),
reg.histogram_family<double>(
"caf.actor", "mailbox-time", {"name"}, default_buckets,
"Time a message waits in the mailbox before processing.", "seconds"),
reg.gauge_family("caf.actor", "mailbox-size", {"name"},
"Number of messages in the mailbox."),
};
}
......@@ -249,6 +276,17 @@ actor_system::actor_system(actor_system_config& cfg)
CAF_SET_LOGGER_SYS(this);
for (auto& hook : cfg.thread_hooks_)
hook->init(*this);
// Cache some configuration parameters for faster lookups at runtime.
using string_list = std::vector<std::string>;
if (auto lst = get_if<string_list>(&cfg,
"caf.metrics-filters.actors.includes"))
metrics_actors_includes_ = std::move(*lst);
if (auto lst = get_if<string_list>(&cfg,
"caf.metrics-filters.actors.excludes"))
metrics_actors_excludes_ = std::move(*lst);
if (!metrics_actors_includes_.empty())
actor_metric_families_ = make_actor_metric_families(metrics_);
// Spin up modules.
for (auto& f : cfg.module_factories) {
auto mod_ptr = f(*this);
modules_[mod_ptr->id()].reset(mod_ptr);
......
......@@ -28,6 +28,7 @@
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/invoke_message_result.hpp"
#include "caf/logger.hpp"
#include "caf/telemetry/timer.hpp"
namespace caf {
......@@ -68,10 +69,17 @@ void blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_LOG_SEND_EVENT(ptr);
auto mid = ptr->mid;
auto src = ptr->sender;
auto collects_metrics = getf(abstract_actor::collects_metrics_flag);
if (collects_metrics) {
ptr->set_enqueue_time();
metrics_.mailbox_size->inc();
}
// returns false if mailbox has been closed
if (!mailbox().synchronized_push_back(mtx_, cv_, std::move(ptr))) {
CAF_LOG_REJECT_EVENT();
home_system().base_metrics().rejected_messages->inc();
if (collects_metrics)
metrics_.mailbox_size->dec();
if (mid.is_request()) {
detail::sync_request_bouncer srb{exit_reason()};
srb(src, mid);
......@@ -86,7 +94,7 @@ mailbox_element* blocking_actor::peek_at_next_mailbox_element() {
}
const char* blocking_actor::name() const {
return "blocking_actor";
return "user.blocking-actor";
}
void blocking_actor::launch(execution_unit*, bool, bool hide) {
......@@ -216,15 +224,33 @@ blocking_actor::mailbox_visitor::operator()(mailbox_element& x) {
return visit(f, sres);
};
// Post-process the returned value from the function body.
if (!self->getf(abstract_actor::collects_metrics_flag)) {
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;
} else {
auto t0 = std::chrono::steady_clock::now();
auto mbox_time = x.seconds_until(t0);
auto result = body();
if (result == intrusive::task_result::skip) {
CAF_AFTER_PROCESSING(self, invoke_message_result::skipped);
CAF_LOG_SKIP_EVENT();
auto& builtins = self->builtin_metrics();
telemetry::timer::observe(builtins.processing_time, t0);
builtins.mailbox_time->observe(mbox_time);
builtins.mailbox_size->dec();
} else {
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
CAF_LOG_FINALIZE_EVENT();
}
return result;
}
}
void blocking_actor::receive_impl(receive_cond& rcc, message_id mid,
......@@ -325,8 +351,14 @@ bool blocking_actor::cleanup(error&& fail_state, execution_unit* host) {
mailbox_.close();
// TODO: messages that are stuck in the cache can get lost
detail::sync_request_bouncer bounce{fail_state};
while (mailbox_.queue().new_round(1000, bounce).consumed_items)
; // nop
auto dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
while (dropped > 0) {
if (getf(abstract_actor::collects_metrics_flag)) {
auto val = static_cast<int64_t>(dropped);
metrics_.mailbox_size->dec(val);
}
dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
}
}
// Dispatch to parent's `cleanup` function.
return super::cleanup(std::move(fail_state), host);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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. *
******************************************************************************/
// Based on work of others. Retrieved on 2020-07-01 from
// https://github.com/Genivia/ugrep/blob/d2fb133/src/glob.cpp.
// Original header / license:
/******************************************************************************\
* Copyright (c) 2019, Robert van Engelen, Genivia Inc. All rights reserved. *
* *
* Redistribution and use in source and binary forms, with or without *
* modification, are permitted provided that the following conditions are met: *
* *
* (1) Redistributions of source code must retain the above copyright notice, *
* this list of conditions and the following disclaimer. *
* *
* (2) Redistributions in binary form must reproduce the above copyright *
* notice, this list of conditions and the following disclaimer in the *
* documentation and/or other materials provided with the distribution. *
* *
* (3) The name of the author may not be used to endorse or promote products *
* derived from this software without specific prior written permission. *
* *
* THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED *
* WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF *
* MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO *
* EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, *
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, *
* PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; *
* OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, *
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR *
* OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF *
* ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *
\******************************************************************************/
#include "caf/detail/glob_match.hpp"
#include <cstdio>
#include <cstring>
#include "caf/config.hpp"
namespace caf::detail {
namespace {
static constexpr char path_separator =
#ifdef CAF_WINDOWS
'\\'
#else
'/'
#endif
;
int utf8(const char** s);
// match text against glob, return true or false
bool match(const char* text, const char* glob) {
// to iteratively backtrack on *
const char* text1_backup = nullptr;
const char* glob1_backup = nullptr;
// to iteratively backtrack on **
const char* text2_backup = nullptr;
const char* glob2_backup = nullptr;
// match until end of text
while (*text != '\0') {
switch (*glob) {
case '*':
if (*++glob == '*') {
// trailing ** match everything after /
if (*++glob == '\0')
return true;
// ** followed by a / match zero or more directories
if (*glob != '/')
return false;
// iteratively backtrack on **
text1_backup = NULL;
glob1_backup = NULL;
text2_backup = text;
glob2_backup = ++glob;
continue;
}
// iteratively backtrack on *
text1_backup = text;
glob1_backup = glob;
continue;
case '?':
// match anything except /
if (*text == path_separator)
break;
utf8(&text);
glob++;
continue;
case '[': {
int chr = utf8(&text), last = 0x10ffff;
// match anything except /
if (chr == path_separator)
break;
bool matched = false;
bool reverse = glob[1] == '^' || glob[1] == '!';
// inverted character class
if (reverse)
glob++;
glob++;
// match character class
while (*glob != '\0' && *glob != ']')
if (last < 0x10ffff && *glob == '-' && glob[1] != ']'
&& glob[1] != '\0'
? chr <= utf8(&++glob) && chr >= last
: chr == (last = utf8(&glob)))
matched = true;
if (matched == reverse)
break;
if (*glob)
glob++;
continue;
}
case '\\':
// literal match \-escaped character
glob++;
[[fallthrough]];
default:
#ifdef CAF_WINDOWS
if (*glob != *text && !(*glob == '/' && *text == '\\'))
#else
if (*glob != *text)
#endif
break;
text++;
glob++;
continue;
}
if (glob1_backup != NULL && *text1_backup != path_separator) {
// backtrack on the last *, do not jump over /
text = ++text1_backup;
glob = glob1_backup;
continue;
}
if (glob2_backup != NULL) {
// backtrack on the last **
text = ++text2_backup;
glob = glob2_backup;
continue;
}
return false;
}
while (*glob == '*')
glob++;
return *glob == '\0';
}
int utf8(const char** s) {
int c1, c2, c3, c = (unsigned char) **s;
if (c != '\0')
(*s)++;
if (c < 0x80)
return c;
c1 = (unsigned char) **s;
if (c < 0xC0 || (c == 0xC0 && c1 != 0x80) || c == 0xC1 || (c1 & 0xC0) != 0x80)
return 0xFFFD;
if (c1 != '\0')
(*s)++;
c1 &= 0x3F;
if (c < 0xE0)
return (((c & 0x1F) << 6) | c1);
c2 = (unsigned char) **s;
if ((c == 0xE0 && c1 < 0x20) || (c2 & 0xC0) != 0x80)
return 0xFFFD;
if (c2 != '\0')
(*s)++;
c2 &= 0x3F;
if (c < 0xF0)
return (((c & 0x0F) << 12) | (c1 << 6) | c2);
c3 = (unsigned char) **s;
if (c3 != '\0')
(*s)++;
if ((c == 0xF0 && c1 < 0x10) || (c == 0xF4 && c1 >= 0x10) || c >= 0xF5
|| (c3 & 0xC0) != 0x80)
return 0xFFFD;
return (((c & 0x07) << 18) | (c1 << 12) | (c2 << 6) | (c3 & 0x3F));
}
} // namespace
// pathname or basename matching, returns true or false
bool glob_match(const char* str, const char* glob) {
// an empty glob matches nothing and an empty string matches no glob
if (glob[0] == '\0' || str[0] == '\0')
return false;
// if str starts with ./ then remove these pairs
while (str[0] == '.' && str[1] == path_separator)
str += 2;
// if str starts with / then remove it
if (str[0] == path_separator)
++str;
// a leading / in the glob means globbing the str after removing the /
if (glob[0] == '/')
++glob;
return match(str, glob);
}
} // namespace caf::detail
......@@ -27,6 +27,7 @@
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/default_attachable.hpp"
#include "caf/detail/glob_match.hpp"
#include "caf/exit_reason.hpp"
#include "caf/logger.hpp"
#include "caf/resumable.hpp"
......@@ -35,6 +36,36 @@
namespace caf {
namespace {
local_actor::metrics_t make_instance_metrics(local_actor* self) {
const auto& sys = self->home_system();
const auto& includes = sys.metrics_actors_includes();
const auto& excludes = sys.metrics_actors_excludes();
const auto* name = self->name();
auto matches = [name](const std::string& glob) {
return detail::glob_match(name, glob.c_str());
};
if (includes.empty()
|| std::none_of(includes.begin(), includes.end(), matches)
|| std::any_of(excludes.begin(), excludes.end(), matches))
return {
nullptr,
nullptr,
nullptr,
};
self->setf(abstract_actor::collects_metrics_flag);
const auto& families = sys.actor_metric_families();
string_view sv{name, strlen(name)};
return {
families.processing_time_family->get_or_add({{"name", sv}}),
families.mailbox_time_family->get_or_add({{"name", sv}}),
families.mailbox_size_family->get_or_add({{"name", sv}}),
};
}
} // namespace
local_actor::local_actor(actor_config& cfg)
: monitorable_actor(cfg),
context_(cfg.host),
......@@ -59,6 +90,10 @@ void local_actor::on_destroy() {
}
}
void local_actor::setup_metrics() {
metrics_ = make_instance_metrics(this);
}
void local_actor::request_response_timeout(timespan timeout, message_id mid) {
CAF_LOG_TRACE(CAF_ARG(timeout) << CAF_ARG(mid));
if (timeout == infinite)
......@@ -113,7 +148,7 @@ void local_actor::send_exit(const strong_actor_ptr& dest, error reason) {
}
const char* local_actor::name() const {
return "actor";
return "user.local-actor";
}
error local_actor::save_state(serializer&, const unsigned int) {
......
......@@ -34,7 +34,7 @@
namespace caf {
const char* monitorable_actor::name() const {
return "monitorable_actor";
return "user.monitorable-actor";
}
void monitorable_actor::attach(attachable_ptr ptr) {
......
......@@ -21,14 +21,12 @@
#include "caf/actor_ostream.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/config.hpp"
#include "caf/inbound_path.hpp"
#include "caf/to_string.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp"
#include "caf/detail/private_thread.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/inbound_path.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/to_string.hpp"
using namespace std::string_literals;
......@@ -166,6 +164,11 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
CAF_LOG_SEND_EVENT(ptr);
auto mid = ptr->mid;
auto sender = ptr->sender;
auto collects_metrics = getf(abstract_actor::collects_metrics_flag);
if (collects_metrics) {
ptr->set_enqueue_time();
metrics_.mailbox_size->inc();
}
switch (mailbox().push_back(std::move(ptr))) {
case intrusive::inbox_result::unblocked_reader: {
CAF_LOG_ACCEPT_EVENT(true);
......@@ -185,6 +188,8 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
case intrusive::inbox_result::queue_closed: {
CAF_LOG_REJECT_EVENT();
home_system().base_metrics().rejected_messages->inc();
if (collects_metrics)
metrics_.mailbox_size->dec();
if (mid.is_request()) {
detail::sync_request_bouncer f{exit_reason()};
f(sender, mid);
......@@ -204,7 +209,7 @@ mailbox_element* scheduled_actor::peek_at_next_mailbox_element() {
// -- overridden functions of local_actor --------------------------------------
const char* scheduled_actor::name() const {
return "scheduled_actor";
return "user.scheduled-actor";
}
void scheduled_actor::launch(execution_unit* eu, bool lazy, bool hide) {
......@@ -253,8 +258,14 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
get_normal_queue().flush_cache();
get_urgent_queue().flush_cache();
detail::sync_request_bouncer bounce{fail_state};
while (mailbox_.queue().new_round(1000, bounce).consumed_items > 0)
; // nop
auto dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
while (dropped > 0) {
if (getf(abstract_actor::collects_metrics_flag)) {
auto val = static_cast<int64_t>(dropped);
metrics_.mailbox_size->dec(val);
}
dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
}
}
// Dispatch to parent's `cleanup` function.
return super::cleanup(std::move(fail_state), host);
......@@ -293,6 +304,7 @@ intrusive::task_result
scheduled_actor::mailbox_visitor::operator()(size_t, upstream_queue&,
mailbox_element& x) {
CAF_ASSERT(x.content().match_elements<upstream_msg>());
return run(x, [&] {
self->current_mailbox_element(&x);
CAF_LOG_RECEIVE_EVENT((&x));
CAF_BEFORE_PROCESSING(self, x);
......@@ -302,6 +314,7 @@ scheduled_actor::mailbox_visitor::operator()(size_t, upstream_queue&,
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
return ++handled_msgs < max_throughput ? intrusive::task_result::resume
: intrusive::task_result::stop_all;
});
}
namespace {
......@@ -327,8 +340,7 @@ struct downstream_msg_visitor {
CAF_ASSERT(
inptr->slots == dm.slots
|| (dm.slots.sender == 0 && dm.slots.receiver == inptr->slots.receiver));
// TODO: replace with `if constexpr` when switching to C++17
if (std::is_same<T, downstream_msg::close>::value
if constexpr (std::is_same<T, downstream_msg::close>::value
|| std::is_same<T, downstream_msg::forced_close>::value) {
inptr.reset();
qs_ref.erase_later(dm.slots.receiver);
......@@ -355,6 +367,7 @@ intrusive::task_result scheduled_actor::mailbox_visitor::operator()(
size_t, downstream_queue& qs, stream_slot,
policy::downstream_messages::nested_queue_type& q, mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(handled_msgs));
return run(x, [&, this] {
self->current_mailbox_element(&x);
CAF_LOG_RECEIVE_EVENT((&x));
CAF_BEFORE_PROCESSING(self, x);
......@@ -365,22 +378,26 @@ intrusive::task_result scheduled_actor::mailbox_visitor::operator()(
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
return ++handled_msgs < max_throughput ? res
: intrusive::task_result::stop_all;
});
}
intrusive::task_result
scheduled_actor::mailbox_visitor::operator()(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(handled_msgs));
return run(x, [&, this] {
switch (self->reactivate(x)) {
case activation_result::terminated:
return intrusive::task_result::stop;
case activation_result::success:
return ++handled_msgs < max_throughput ? intrusive::task_result::resume
return ++handled_msgs < max_throughput
? intrusive::task_result::resume
: intrusive::task_result::stop_all;
case activation_result::skipped:
return intrusive::task_result::skip;
default:
return intrusive::task_result::resume;
}
});
}
resumable::resume_result scheduled_actor::resume(execution_unit* ctx,
......@@ -403,7 +420,8 @@ resumable::resume_result scheduled_actor::resume(execution_unit* ctx,
set_stream_timeout(tout);
}
};
mailbox_visitor f{this, handled_msgs, max_throughput};
mailbox_visitor f{this, handled_msgs, max_throughput,
getf(abstract_actor::collects_metrics_flag)};
mailbox_element_ptr ptr;
// Timeout for calling `advance_streams`.
while (handled_msgs < max_throughput) {
......
......@@ -55,6 +55,10 @@ public:
mh_(what->content());
}
void setup_metrics() {
// nop
}
private:
message_handler mh_;
};
......
......@@ -48,7 +48,9 @@ struct ms_timestamp {
ms_timestamp& operator=(const ms_timestamp&) = default;
};
struct underline_to_hyphen {
// Converts separators such as '.' and '-' to underlines to follow the
// Prometheus naming conventions.
struct separator_to_underline {
string_view str;
};
......@@ -60,7 +62,7 @@ template <class... Ts>
void append(prometheus::char_buffer&, string_view, Ts&&...);
template <class... Ts>
void append(prometheus::char_buffer&, underline_to_hyphen, Ts&&...);
void append(prometheus::char_buffer&, separator_to_underline, Ts&&...);
template <class... Ts>
void append(prometheus::char_buffer&, char, Ts&&...);
......@@ -94,9 +96,18 @@ void append(prometheus::char_buffer& buf, string_view str, Ts&&... xs) {
}
template <class... Ts>
void append(prometheus::char_buffer& buf, underline_to_hyphen x, Ts&&... xs) {
for (auto c : x.str)
buf.emplace_back(c != '-' ? c : '_');
void append(prometheus::char_buffer& buf, separator_to_underline x,
Ts&&... xs) {
for (auto c : x.str) {
switch (c) {
default:
buf.emplace_back(c);
break;
case '-':
case '.':
buf.emplace_back('_');
}
}
append(buf, std::forward<Ts>(xs)...);
}
......@@ -131,8 +142,8 @@ append(prometheus::char_buffer& buf, T val, Ts&&... xs) {
template <class... Ts>
void append(prometheus::char_buffer& buf, const metric_family* family,
Ts&&... xs) {
append(buf, underline_to_hyphen{family->prefix()}, '_',
underline_to_hyphen{family->name()});
append(buf, separator_to_underline{family->prefix()}, '_',
separator_to_underline{family->name()});
if (family->unit() != "1"_sv)
append(buf, '_', family->unit());
if (family->is_sum())
......
......@@ -42,7 +42,7 @@ metric_registry::metric_registry() : config_(nullptr) {
}
metric_registry::metric_registry(const actor_system_config& cfg) {
config_ = get_if<settings>(&cfg, "metrics");
config_ = get_if<settings>(&cfg, "caf.metrics");
}
metric_registry::~metric_registry() {
......
......@@ -44,12 +44,10 @@ behavior pong(stateful_actor<pong_state>*) {
}
struct ping_state {
static const char* name;
static inline const char* name = "ping";
bool had_first_timeout = false; // unused in ping_singleN functions
};
const char* ping_state::name = "ping";
using ping_actor = stateful_actor<ping_state>;
using fptr = behavior (*)(ping_actor*, bool*, const actor&);
......
......@@ -123,21 +123,19 @@ CAF_TEST(stateful actors without explicit name use the name of the parent) {
struct state {
// empty
};
test_name<state>("scheduled_actor");
test_name<state>("user.scheduled-actor");
}
CAF_TEST(states with C string names override the default name) {
struct state {
const char* name = "testee";
};
test_name<state>("testee");
}
namespace {
CAF_TEST(states with STL string names override the default name) {
struct state {
std::string name = "testee2";
};
test_name<state>("testee2");
struct named_state {
static inline const char* name = "testee";
};
} // namespace
CAF_TEST(states with static C string names override the default name) {
test_name<named_state>("testee");
}
CAF_TEST(states can accept constructor arguments and provide a behavior) {
......@@ -167,10 +165,9 @@ CAF_TEST(states can accept constructor arguments and provide a behavior) {
}
CAF_TEST(states optionally take the self pointer as first argument) {
struct state_type {
struct state_type : named_state {
event_based_actor* self;
int x;
const char* name = "testee";
state_type(event_based_actor* self, int x) : self(self), x(x) {
// nop
}
......@@ -190,10 +187,9 @@ CAF_TEST(states optionally take the self pointer as first argument) {
}
CAF_TEST(typed actors can use typed_actor_pointer as self pointer) {
struct state_type {
struct state_type : named_state {
using self_pointer = typed_adder_actor::pointer_view;
self_pointer self;
const char* name = "testee";
int value;
state_type(self_pointer self, int x) : self(self), value(x) {
// nop
......
......@@ -169,7 +169,7 @@ caf.mailbox-size{name="printer"} 3
caf.mailbox-size{name="parser"} 12)");
}
CAF_TEST(buckets for histograms are configurable via runtime settings){
CAF_TEST(buckets for histograms are configurable via runtime settings) {
auto bounds = [](auto&& buckets) {
std::vector<int64_t> result;
for (auto&& bucket : buckets)
......@@ -208,3 +208,19 @@ CAF_TEST(counter_instance is a shortcut for using the family manually) {
}
CAF_TEST_FIXTURE_SCOPE_END()
#define CHECK_CONTAINS(str) \
CAF_CHECK_NOT_EQUAL(collector.result.find(str), npos)
CAF_TEST(enabling actor metrics per config creates metric instances) {
actor_system_config cfg;
test_coordinator_fixture<>::init_config(cfg);
put(cfg.content, "caf.metrics-filters.actors.includes",
std::vector<std::string>{"caf.system.*"});
actor_system sys{cfg};
test_collector collector;
sys.metrics().collect(collector);
auto npos = std::string::npos;
CHECK_CONTAINS(R"(caf.actor.mailbox-size{name="caf.system.spawn-server"})");
CHECK_CONTAINS(R"(caf.actor.mailbox-size{name="caf.system.config-server"})");
}
......@@ -24,7 +24,7 @@
namespace caf::io {
struct connection_helper_state {
static const char* name;
static inline const char* name = "caf.system.connection-helper";
};
CAF_IO_EXPORT behavior
......
......@@ -211,7 +211,7 @@ prometheus_broker::~prometheus_broker() {
}
const char* prometheus_broker::name() const {
return "prometheus-broker";
return "caf.system.prometheus-broker";
}
bool prometheus_broker::has_process_metrics() noexcept {
......
......@@ -366,7 +366,7 @@ abstract_broker::resume(execution_unit* ctx, size_t mt) {
}
const char* abstract_broker::name() const {
return "broker";
return "user.broker";
}
void abstract_broker::init_broker() {
......
......@@ -97,7 +97,7 @@ void basp_broker::on_exit() {
}
const char* basp_broker::name() const {
return "basp-broker";
return "caf.system.basp-broker";
}
behavior basp_broker::make_behavior() {
......
......@@ -38,8 +38,6 @@ auto autoconnect_timeout = std::chrono::minutes(10);
} // namespace
const char* connection_helper_state::name = "connection_helper";
behavior
connection_helper(stateful_actor<connection_helper_state>* self, actor b) {
CAF_LOG_TRACE(CAF_ARG(b));
......
......@@ -305,8 +305,8 @@ void middleman::start() {
manager_ = make_middleman_actor(system(), basp);
// Launch metrics exporters.
using dict = config_value::dictionary;
if (auto ex = get_if<dict>(&system().config(), "metrics.export"))
if (auto prom = get_if<dict>(ex, "prometheus-http"))
if (auto prom = get_if<dict>(&system().config(),
"caf.middleman.prometheus-http"))
expose_prometheus_metrics(*prom);
}
......
......@@ -68,7 +68,7 @@ void middleman_actor_impl::on_exit() {
}
const char* middleman_actor_impl::name() const {
return "middleman_actor";
return "caf.system.middleman-actor";
}
auto middleman_actor_impl::make_behavior() -> behavior_type {
......
......@@ -947,9 +947,8 @@ T unbox(T* x) {
/// Implementation detail for `TESTEE` and `VARARGS_TESTEE`.
#define TESTEE_SCAFFOLD(tname) \
struct tname##_state : testee_state_base<tname##_state> { \
static const char* name; \
static inline const char* name = #tname; \
}; \
const char* tname##_state::name = #tname; \
using tname##_actor = stateful_actor<tname##_state>
/// Convenience macro for defining an actor named `tname`.
......
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