Commit a07989aa authored by Dominik Charousset's avatar Dominik Charousset

Collect some base metrics per actor system

parent f81db0d3
...@@ -122,7 +122,6 @@ private: ...@@ -122,7 +122,6 @@ private:
actor_registry(actor_system& sys); actor_registry(actor_system& sys);
telemetry::int_gauge* running_;
mutable std::mutex running_mtx_; mutable std::mutex running_mtx_;
mutable std::condition_variable running_cv_; mutable std::condition_variable running_cv_;
......
...@@ -173,6 +173,21 @@ public: ...@@ -173,6 +173,21 @@ public:
virtual void demonitor(const node_id& node, const actor_addr& observer) = 0; virtual void demonitor(const node_id& node, const actor_addr& observer) = 0;
}; };
/// Metrics collected by the actor system by default.
/// @warning Do not modify these metrics in user code. Some may be used by the
/// system for synchronization.
struct base_metrics_t {
/// Counts the number of messages that where rejected because the target
/// mailbox was closed or did not exist.
telemetry::int_counter* rejected_messages;
/// Counts the total number of processed messages.
telemetry::int_counter* processed_messages;
/// Tracks the current number of running actors in the system.
telemetry::int_gauge* running_actors;
};
/// @warning The system stores a reference to `cfg`, which means the /// @warning The system stores a reference to `cfg`, which means the
/// config object must outlive the actor system. /// config object must outlive the actor system.
explicit actor_system(actor_system_config& cfg); explicit actor_system(actor_system_config& cfg);
...@@ -569,6 +584,14 @@ public: ...@@ -569,6 +584,14 @@ public:
profiler_->before_sending_scheduled(self, timeout, element); profiler_->before_sending_scheduled(self, timeout, element);
} }
base_metrics_t& base_metrics() noexcept {
return base_metrics_;
}
const base_metrics_t& base_metrics() const noexcept {
return base_metrics_;
}
tracing_data_factory* tracing_context() const noexcept { tracing_data_factory* tracing_context() const noexcept {
return tracing_context_; return tracing_context_;
} }
...@@ -608,6 +631,9 @@ private: ...@@ -608,6 +631,9 @@ private:
/// Manages all metrics collected by the system. /// Manages all metrics collected by the system.
telemetry::metric_registry metrics_; telemetry::metric_registry metrics_;
/// Stores all metrics that the actor system collects by default.
base_metrics_t base_metrics_;
/// Identifies this actor system in a distributed setting. /// Identifies this actor system in a distributed setting.
node_id node_; node_id node_;
......
...@@ -35,13 +35,14 @@ template <class Self, class SelfHandle, class Handle, class... Ts> ...@@ -35,13 +35,14 @@ template <class Self, class SelfHandle, class Handle, class... Ts>
void profiled_send(Self* self, SelfHandle&& src, const Handle& dst, void profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
message_id msg_id, std::vector<strong_actor_ptr> stages, message_id msg_id, std::vector<strong_actor_ptr> stages,
execution_unit* context, Ts&&... xs) { execution_unit* context, Ts&&... xs) {
CAF_IGNORE_UNUSED(self);
if (dst) { if (dst) {
auto element = make_mailbox_element(std::forward<SelfHandle>(src), msg_id, auto element = make_mailbox_element(std::forward<SelfHandle>(src), msg_id,
std::move(stages), std::move(stages),
std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING(self, *element); CAF_BEFORE_SENDING(self, *element);
dst->enqueue(std::move(element), context); dst->enqueue(std::move(element), context);
} else {
self->home_system().base_metrics().rejected_messages->inc();
} }
} }
...@@ -49,7 +50,6 @@ template <class Self, class SelfHandle, class Handle, class... Ts> ...@@ -49,7 +50,6 @@ template <class Self, class SelfHandle, class Handle, class... Ts>
void profiled_send(Self* self, SelfHandle&& src, const Handle& dst, void profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
actor_clock& clock, actor_clock::time_point timeout, actor_clock& clock, actor_clock::time_point timeout,
message_id msg_id, Ts&&... xs) { message_id msg_id, Ts&&... xs) {
CAF_IGNORE_UNUSED(self);
if (dst) { if (dst) {
if constexpr (std::is_same<Handle, group>::value) { if constexpr (std::is_same<Handle, group>::value) {
clock.schedule_message(timeout, dst, std::forward<SelfHandle>(src), clock.schedule_message(timeout, dst, std::forward<SelfHandle>(src),
...@@ -61,6 +61,8 @@ void profiled_send(Self* self, SelfHandle&& src, const Handle& dst, ...@@ -61,6 +61,8 @@ void profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(dst), clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(dst),
std::move(element)); std::move(element));
} }
} else {
self->home_system().base_metrics().rejected_messages->inc();
} }
} }
......
...@@ -170,7 +170,7 @@ public: ...@@ -170,7 +170,7 @@ public:
/// @returns `true` if `f` consumed at least one item. /// @returns `true` if `f` consumed at least one item.
template <class F> template <class F>
bool consume(F& f) noexcept(noexcept(f(std::declval<value_type&>()))) { bool consume(F& f) noexcept(noexcept(f(std::declval<value_type&>()))) {
return new_round(0, f).consumed_items; return new_round(0, f).consumed_items > 0;
} }
/// Run a new round with `quantum`, dispatching all tasks to `consumer`. /// Run a new round with `quantum`, dispatching all tasks to `consumer`.
...@@ -178,12 +178,12 @@ public: ...@@ -178,12 +178,12 @@ public:
new_round_result new_round(deficit_type quantum, F& consumer) noexcept( new_round_result new_round(deficit_type quantum, F& consumer) noexcept(
noexcept(consumer(std::declval<value_type&>()))) { noexcept(consumer(std::declval<value_type&>()))) {
if (list_.empty()) if (list_.empty())
return {false, false}; return {0, false};
deficit_ += quantum; deficit_ += quantum;
long consumed = 0;
auto ptr = next(); auto ptr = next();
if (ptr == nullptr) if (ptr == nullptr)
return {false, false}; return {0, false};
size_t consumed = 0;
do { do {
auto consumer_res = consumer(*ptr); auto consumer_res = consumer(*ptr);
switch (consumer_res) { switch (consumer_res) {
...@@ -194,7 +194,7 @@ public: ...@@ -194,7 +194,7 @@ public:
cache_.push_back(ptr.release()); cache_.push_back(ptr.release());
if (list_.empty()) { if (list_.empty()) {
deficit_ = 0; deficit_ = 0;
return {consumed != 0, false}; return {consumed, false};
} }
break; break;
case task_result::resume: case task_result::resume:
...@@ -202,7 +202,7 @@ public: ...@@ -202,7 +202,7 @@ public:
flush_cache(); flush_cache();
if (list_.empty()) { if (list_.empty()) {
deficit_ = 0; deficit_ = 0;
return {consumed != 0, false}; return {consumed, false};
} }
break; break;
default: default:
...@@ -210,11 +210,11 @@ public: ...@@ -210,11 +210,11 @@ public:
flush_cache(); flush_cache();
if (list_.empty()) if (list_.empty())
deficit_ = 0; deficit_ = 0;
return {consumed != 0, consumer_res == task_result::stop_all}; return {consumed, consumer_res == task_result::stop_all};
} }
ptr = next(); ptr = next();
} while (ptr != nullptr); } while (ptr != nullptr);
return {consumed != 0, false}; return {consumed, false};
} }
cache_type& cache() noexcept { cache_type& cache() noexcept {
......
...@@ -95,25 +95,27 @@ public: ...@@ -95,25 +95,27 @@ public:
/// @returns `true` if at least one item was consumed, `false` otherwise. /// @returns `true` if at least one item was consumed, `false` otherwise.
template <class F> template <class F>
new_round_result new_round(deficit_type quantum, F& consumer) { new_round_result new_round(deficit_type quantum, F& consumer) {
size_t consumed = 0;
if (!super::empty()) { if (!super::empty()) {
deficit_ += quantum; deficit_ += quantum;
auto ptr = next(); auto ptr = next();
if (ptr == nullptr) if (ptr == nullptr)
return {false, false}; return {0, false};
do { do {
++consumed;
switch (consumer(*ptr)) { switch (consumer(*ptr)) {
default: default:
break; break;
case task_result::stop: case task_result::stop:
return {true, false}; return {consumed, false};
case task_result::stop_all: case task_result::stop_all:
return {true, true}; return {consumed, true};
} }
ptr = next(); ptr = next();
} while (ptr != nullptr); } while (ptr != nullptr);
return {true, false}; return {consumed, false};
} }
return {false, false}; return {consumed, false};
} }
private: private:
......
...@@ -22,15 +22,16 @@ ...@@ -22,15 +22,16 @@
#include <type_traits> #include <type_traits>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/meta/type_name.hpp"
namespace caf::intrusive { namespace caf::intrusive {
/// Returns the state of a consumer from `new_round`. /// Returns the state of a consumer from `new_round`.
struct new_round_result { struct new_round_result {
/// Denotes whether the consumer accepted at least one element. /// Denotes whether the consumer accepted at least one element.
bool consumed_items : 1; size_t consumed_items;
/// Denotes whether the consumer returned `task_result::stop_all`. /// Denotes whether the consumer returned `task_result::stop_all`.
bool stop_all : 1; bool stop_all;
}; };
constexpr bool operator==(new_round_result x, new_round_result y) { constexpr bool operator==(new_round_result x, new_round_result y) {
...@@ -41,13 +42,9 @@ constexpr bool operator!=(new_round_result x, new_round_result y) { ...@@ -41,13 +42,9 @@ constexpr bool operator!=(new_round_result x, new_round_result y) {
return !(x == y); return !(x == y);
} }
constexpr new_round_result template <class Inspector>
make_new_round_result(bool consumed_items, bool stop_all = false) { typename Inspector::result_type inspect(Inspector& f, new_round_result& x) {
return {consumed_items, stop_all}; return f(meta::type_name("new_round_result"), x.consumed_items, x.stop_all);
}
constexpr new_round_result operator|(new_round_result x, new_round_result y) {
return {x.consumed_items || y.consumed_items, x.stop_all || y.stop_all};
} }
} // namespace caf::intrusive } // namespace caf::intrusive
...@@ -110,7 +110,7 @@ public: ...@@ -110,7 +110,7 @@ public:
/// @returns `true` if at least one item was consumed, `false` otherwise. /// @returns `true` if at least one item was consumed, `false` otherwise.
template <class F> template <class F>
new_round_result new_round(deficit_type quantum, F& f) { new_round_result new_round(deficit_type quantum, F& f) {
bool result = false; size_t consumed = 0;
bool stopped = false; bool stopped = false;
for (auto& kvp : qs_) { for (auto& kvp : qs_) {
if (policy_.enabled(kvp.second)) { if (policy_.enabled(kvp.second)) {
...@@ -118,7 +118,7 @@ public: ...@@ -118,7 +118,7 @@ public:
if (!stopped) { if (!stopped) {
new_round_helper<F> g{kvp.first, q, f}; new_round_helper<F> g{kvp.first, q, f};
auto res = q.new_round(policy_.quantum(q, quantum), g); auto res = q.new_round(policy_.quantum(q, quantum), g);
result = res.consumed_items; consumed += res.consumed_items;
if (res.stop_all) if (res.stop_all)
stopped = true; stopped = true;
} else { } else {
...@@ -129,7 +129,7 @@ public: ...@@ -129,7 +129,7 @@ public:
} }
} }
cleanup(); cleanup();
return {result, stopped}; return {consumed, stopped};
} }
/// Erases all keys previously marked via `erase_later`. /// Erases all keys previously marked via `erase_later`.
......
...@@ -186,7 +186,7 @@ private: ...@@ -186,7 +186,7 @@ private:
template <size_t I, class F> template <size_t I, class F>
detail::enable_if_t<I == num_queues, new_round_result> detail::enable_if_t<I == num_queues, new_round_result>
new_round_recursion(deficit_type, F&) noexcept { new_round_recursion(deficit_type, F&) noexcept {
return {false, false}; return {0, false};
} }
template <size_t I, class F> template <size_t I, class F>
...@@ -202,7 +202,8 @@ private: ...@@ -202,7 +202,8 @@ private:
inc_deficit_recursion<I + 1>(quantum); inc_deficit_recursion<I + 1>(quantum);
return res; return res;
} }
return res | new_round_recursion<I + 1>(quantum, f); auto sub = new_round_recursion<I + 1>(quantum, f);
return {res.consumed_items + sub.consumed_items, sub.stop_all};
} }
template <size_t I> template <size_t I>
......
...@@ -75,6 +75,7 @@ public: ...@@ -75,6 +75,7 @@ public:
} else { } else {
self->eq_impl(req_id.response_id(), self->ctrl(), self->context(), self->eq_impl(req_id.response_id(), self->ctrl(), self->context(),
make_error(sec::invalid_argument)); make_error(sec::invalid_argument));
self->home_system().base_metrics().rejected_messages->inc();
} }
using response_type using response_type
= response_type_t<typename Handle::signatures, = response_type_t<typename Handle::signatures,
......
...@@ -68,11 +68,13 @@ public: ...@@ -68,11 +68,13 @@ public:
"statically typed actors can only send() to other " "statically typed actors can only send() to other "
"statically typed actors; use anon_send() when " "statically typed actors; use anon_send() when "
"communicating to groups"); "communicating to groups");
auto self = dptr();
// TODO: consider whether it's feasible to track messages to groups // TODO: consider whether it's feasible to track messages to groups
if (dest) { if (dest) {
auto self = dptr();
dest->eq_impl(make_message_id(P), self->ctrl(), self->context(), dest->eq_impl(make_message_id(P), self->ctrl(), self->context(),
std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
} else {
self->home_system().base_metrics().rejected_messages->inc();
} }
} }
...@@ -137,11 +139,13 @@ public: ...@@ -137,11 +139,13 @@ public:
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 // TODO: consider whether it's feasible to track messages to groups
auto self = dptr();
if (dest) { if (dest) {
auto self = dptr();
auto& clock = self->system().clock(); auto& clock = self->system().clock();
clock.schedule_message(timeout, dest, self->ctrl(), clock.schedule_message(timeout, dest, self->ctrl(),
make_message(std::forward<Ts>(xs)...)); make_message(std::forward<Ts>(xs)...));
} else {
self->home_system().base_metrics().rejected_messages->inc();
} }
} }
......
...@@ -49,8 +49,7 @@ actor_registry::~actor_registry() { ...@@ -49,8 +49,7 @@ actor_registry::~actor_registry() {
} }
actor_registry::actor_registry(actor_system& sys) : system_(sys) { actor_registry::actor_registry(actor_system& sys) : system_(sys) {
running_ = sys.metrics().gauge_singleton( // nop
"caf", "running-actors", "Number of currently running actors.");
} }
strong_actor_ptr actor_registry::get_impl(actor_id key) const { strong_actor_ptr actor_registry::get_impl(actor_id key) const {
...@@ -97,19 +96,19 @@ void actor_registry::erase(actor_id key) { ...@@ -97,19 +96,19 @@ void actor_registry::erase(actor_id key) {
void actor_registry::inc_running() { void actor_registry::inc_running() {
# if CAF_LOG_LEVEL >= CAF_LOG_LEVEL_DEBUG # if CAF_LOG_LEVEL >= CAF_LOG_LEVEL_DEBUG
auto value = ++*running_; auto value = ++*system_.base_metrics().running_actors;
CAF_LOG_DEBUG(CAF_ARG(value)); CAF_LOG_DEBUG(CAF_ARG(value));
# else # else
++*running_; system_.base_metrics().running_actors->inc();
# endif # endif
} }
size_t actor_registry::running() const { size_t actor_registry::running() const {
return running_->value(); return static_cast<size_t>(system_.base_metrics().running_actors->value());
} }
void actor_registry::dec_running() { void actor_registry::dec_running() {
size_t new_val = --*running_; size_t new_val = --*system_.base_metrics().running_actors;
if (new_val <= 1) { if (new_val <= 1) {
std::unique_lock<std::mutex> guard(running_mtx_); std::unique_lock<std::mutex> guard(running_mtx_);
running_cv_.notify_all(); running_cv_.notify_all();
...@@ -121,8 +120,8 @@ void actor_registry::await_running_count_equal(size_t expected) const { ...@@ -121,8 +120,8 @@ void actor_registry::await_running_count_equal(size_t expected) const {
CAF_ASSERT(expected == 0 || expected == 1); CAF_ASSERT(expected == 0 || expected == 1);
CAF_LOG_TRACE(CAF_ARG(expected)); CAF_LOG_TRACE(CAF_ARG(expected));
std::unique_lock<std::mutex> guard{running_mtx_}; std::unique_lock<std::mutex> guard{running_mtx_};
while (running_->value() != static_cast<int64_t>(expected)) { while (running() != expected) {
CAF_LOG_DEBUG(CAF_ARG(running_->value())); CAF_LOG_DEBUG(CAF_ARG(running()));
running_cv_.wait(guard); running_cv_.wait(guard);
} }
} }
......
...@@ -216,10 +216,27 @@ actor_system::networking_module::~networking_module() { ...@@ -216,10 +216,27 @@ actor_system::networking_module::~networking_module() {
// nop // nop
} }
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",
"Number of rejected messages.", "1", true),
reg.counter_singleton("caf", "processed-messages",
"Number of processed messages.", "1", true),
reg.gauge_singleton("caf", "running-actors",
"Number of currently running actors."),
};
}
} // namespace
actor_system::actor_system(actor_system_config& cfg) actor_system::actor_system(actor_system_config& cfg)
: profiler_(cfg.profiler), : profiler_(cfg.profiler),
ids_(0), ids_(0),
metrics_(cfg), metrics_(cfg),
base_metrics_(make_base_metrics(metrics_)),
logger_(new caf::logger(*this), false), logger_(new caf::logger(*this), false),
registry_(*this), registry_(*this),
groups_(*this), groups_(*this),
......
...@@ -71,6 +71,7 @@ void blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) { ...@@ -71,6 +71,7 @@ void blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) {
// returns false if mailbox has been closed // returns false if mailbox has been closed
if (!mailbox().synchronized_push_back(mtx_, cv_, std::move(ptr))) { if (!mailbox().synchronized_push_back(mtx_, cv_, std::move(ptr))) {
CAF_LOG_REJECT_EVENT(); CAF_LOG_REJECT_EVENT();
home_system().base_metrics().rejected_messages->inc();
if (mid.is_request()) { if (mid.is_request()) {
detail::sync_request_bouncer srb{exit_reason()}; detail::sync_request_bouncer srb{exit_reason()};
srb(src, mid); srb(src, mid);
......
...@@ -184,6 +184,7 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) { ...@@ -184,6 +184,7 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
} }
case intrusive::inbox_result::queue_closed: { case intrusive::inbox_result::queue_closed: {
CAF_LOG_REJECT_EVENT(); CAF_LOG_REJECT_EVENT();
home_system().base_metrics().rejected_messages->inc();
if (mid.is_request()) { if (mid.is_request()) {
detail::sync_request_bouncer f{exit_reason()}; detail::sync_request_bouncer f{exit_reason()};
f(sender, mid); f(sender, mid);
...@@ -252,7 +253,7 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) { ...@@ -252,7 +253,7 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
get_normal_queue().flush_cache(); get_normal_queue().flush_cache();
get_urgent_queue().flush_cache(); get_urgent_queue().flush_cache();
detail::sync_request_bouncer bounce{fail_state}; detail::sync_request_bouncer bounce{fail_state};
while (mailbox_.queue().new_round(1000, bounce).consumed_items) while (mailbox_.queue().new_round(1000, bounce).consumed_items > 0)
; // nop ; // nop
} }
// Dispatch to parent's `cleanup` function. // Dispatch to parent's `cleanup` function.
...@@ -409,10 +410,14 @@ resumable::resume_result scheduled_actor::resume(execution_unit* ctx, ...@@ -409,10 +410,14 @@ resumable::resume_result scheduled_actor::resume(execution_unit* ctx,
CAF_LOG_DEBUG("start new DRR round"); CAF_LOG_DEBUG("start new DRR round");
// TODO: maybe replace '3' with configurable / adaptive value? // TODO: maybe replace '3' with configurable / adaptive value?
// Dispatch on the different message categories in our mailbox. // Dispatch on the different message categories in our mailbox.
if (!mailbox_.new_round(3, f).consumed_items) { auto consumed = mailbox_.new_round(3, f).consumed_items;
if (consumed == 0) {
reset_timeouts_if_needed(); reset_timeouts_if_needed();
if (mailbox().try_block()) if (mailbox().try_block())
return resumable::awaiting_message; return resumable::awaiting_message;
} else {
auto signed_val = static_cast<int64_t>(consumed);
home_system().base_metrics().processed_messages->inc(signed_val);
} }
// Check whether the visitor left the actor without behavior. // Check whether the visitor left the actor without behavior.
if (finalize()) { if (finalize()) {
......
...@@ -77,6 +77,10 @@ struct fixture { ...@@ -77,6 +77,10 @@ struct fixture {
} }
}; };
auto make_new_round_result(size_t consumed_items, bool stop_all) {
return new_round_result{consumed_items, stop_all};
}
} // namespace } // namespace
CAF_TEST_FIXTURE_SCOPE(drr_cached_queue_tests, fixture) CAF_TEST_FIXTURE_SCOPE(drr_cached_queue_tests, fixture)
...@@ -108,12 +112,12 @@ CAF_TEST(new_round) { ...@@ -108,12 +112,12 @@ CAF_TEST(new_round) {
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9); fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9);
// Allow f to consume 2, 4, and 6. // Allow f to consume 2, 4, and 6.
auto round_result = queue.new_round(3, f); auto round_result = queue.new_round(3, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true)); CAF_CHECK_EQUAL(round_result, make_new_round_result(3, false));
CAF_CHECK_EQUAL(fseq, "246"); CAF_CHECK_EQUAL(fseq, "246");
CAF_CHECK_EQUAL(queue.deficit(), 0); CAF_CHECK_EQUAL(queue.deficit(), 0);
// Allow g to consume 1, 3, 5, and 7. // Allow g to consume 1, 3, 5, and 7.
round_result = queue.new_round(4, g); round_result = queue.new_round(4, g);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true)); CAF_CHECK_EQUAL(round_result, make_new_round_result(4, false));
CAF_CHECK_EQUAL(gseq, "1357"); CAF_CHECK_EQUAL(gseq, "1357");
CAF_CHECK_EQUAL(queue.deficit(), 0); CAF_CHECK_EQUAL(queue.deficit(), 0);
} }
...@@ -128,21 +132,21 @@ CAF_TEST(skipping) { ...@@ -128,21 +132,21 @@ CAF_TEST(skipping) {
return task_result::resume; return task_result::resume;
}; };
CAF_MESSAGE("make a round on an empty queue"); CAF_MESSAGE("make a round on an empty queue");
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(false)); CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(0, false));
CAF_MESSAGE("make a round on a queue with only odd numbers (skip all)"); CAF_MESSAGE("make a round on a queue with only odd numbers (skip all)");
fill(queue, 1, 3, 5); fill(queue, 1, 3, 5);
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(false)); CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(0, false));
CAF_MESSAGE("make a round on a queue with an even number at the front"); CAF_MESSAGE("make a round on a queue with an even number at the front");
fill(queue, 2); fill(queue, 2);
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(true)); CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(1, false));
CAF_CHECK_EQUAL(seq, "2"); CAF_CHECK_EQUAL(seq, "2");
CAF_MESSAGE("make a round on a queue with an even number in between"); CAF_MESSAGE("make a round on a queue with an even number in between");
fill(queue, 7, 9, 4, 11, 13); fill(queue, 7, 9, 4, 11, 13);
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(true)); CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(1, false));
CAF_CHECK_EQUAL(seq, "24"); CAF_CHECK_EQUAL(seq, "24");
CAF_MESSAGE("make a round on a queue with an even number at the back"); CAF_MESSAGE("make a round on a queue with an even number at the back");
fill(queue, 15, 17, 6); fill(queue, 15, 17, 6);
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(true)); CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(1, false));
CAF_CHECK_EQUAL(seq, "246"); CAF_CHECK_EQUAL(seq, "246");
} }
...@@ -196,7 +200,7 @@ CAF_TEST(alternating_consumer) { ...@@ -196,7 +200,7 @@ CAF_TEST(alternating_consumer) {
// sequences and no odd value to read after 7 is available. // sequences and no odd value to read after 7 is available.
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9); fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9);
auto round_result = queue.new_round(1000, h); auto round_result = queue.new_round(1000, h);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true)); CAF_CHECK_EQUAL(round_result, make_new_round_result(8, false));
CAF_CHECK_EQUAL(seq, "21436587"); CAF_CHECK_EQUAL(seq, "21436587");
CAF_CHECK_EQUAL(queue.deficit(), 0); CAF_CHECK_EQUAL(queue.deficit(), 0);
CAF_CHECK_EQUAL(deep_to_string(queue.cache()), "[9]"); CAF_CHECK_EQUAL(deep_to_string(queue.cache()), "[9]");
......
...@@ -76,6 +76,10 @@ struct fixture { ...@@ -76,6 +76,10 @@ struct fixture {
} }
}; };
auto make_new_round_result(size_t consumed_items, bool stop_all) {
return new_round_result{consumed_items, stop_all};
}
} // namespace } // namespace
CAF_TEST_FIXTURE_SCOPE(drr_queue_tests, fixture) CAF_TEST_FIXTURE_SCOPE(drr_queue_tests, fixture)
...@@ -111,22 +115,22 @@ CAF_TEST(new_round) { ...@@ -111,22 +115,22 @@ CAF_TEST(new_round) {
}; };
// Allow f to consume 1, 2, and 3 with a leftover deficit of 1. // Allow f to consume 1, 2, and 3 with a leftover deficit of 1.
auto round_result = queue.new_round(7, f); auto round_result = queue.new_round(7, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true)); CAF_CHECK_EQUAL(round_result, make_new_round_result(3, false));
CAF_CHECK_EQUAL(seq, "123"); CAF_CHECK_EQUAL(seq, "123");
CAF_CHECK_EQUAL(queue.deficit(), 1); CAF_CHECK_EQUAL(queue.deficit(), 1);
// Allow f to consume 4 and 5 with a leftover deficit of 0. // Allow f to consume 4 and 5 with a leftover deficit of 0.
round_result = queue.new_round(8, f); round_result = queue.new_round(8, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true)); CAF_CHECK_EQUAL(round_result, make_new_round_result(2, false));
CAF_CHECK_EQUAL(seq, "12345"); CAF_CHECK_EQUAL(seq, "12345");
CAF_CHECK_EQUAL(queue.deficit(), 0); CAF_CHECK_EQUAL(queue.deficit(), 0);
// Allow f to consume 6 with a leftover deficit of 0 (queue is empty). // Allow f to consume 6 with a leftover deficit of 0 (queue is empty).
round_result = queue.new_round(1000, f); round_result = queue.new_round(1000, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true)); CAF_CHECK_EQUAL(round_result, make_new_round_result(1, false));
CAF_CHECK_EQUAL(seq, "123456"); CAF_CHECK_EQUAL(seq, "123456");
CAF_CHECK_EQUAL(queue.deficit(), 0); CAF_CHECK_EQUAL(queue.deficit(), 0);
// new_round on an empty queue does nothing. // new_round on an empty queue does nothing.
round_result = queue.new_round(1000, f); round_result = queue.new_round(1000, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(false)); CAF_CHECK_EQUAL(round_result, make_new_round_result(0, false));
CAF_CHECK_EQUAL(seq, "123456"); CAF_CHECK_EQUAL(seq, "123456");
CAF_CHECK_EQUAL(queue.deficit(), 0); CAF_CHECK_EQUAL(queue.deficit(), 0);
} }
......
...@@ -133,6 +133,10 @@ struct fixture { ...@@ -133,6 +133,10 @@ struct fixture {
} }
}; };
auto make_new_round_result(size_t consumed_items, bool stop_all) {
return new_round_result{consumed_items, stop_all};
}
} // namespace } // namespace
CAF_TEST_FIXTURE_SCOPE(wdrr_fixed_multiplexed_queue_tests, fixture) CAF_TEST_FIXTURE_SCOPE(wdrr_fixed_multiplexed_queue_tests, fixture)
...@@ -146,19 +150,19 @@ CAF_TEST(new_round) { ...@@ -146,19 +150,19 @@ CAF_TEST(new_round) {
// Allow f to consume 2 items per nested queue. // Allow f to consume 2 items per nested queue.
fetch_helper f; fetch_helper f;
auto round_result = queue.new_round(2, f); auto round_result = queue.new_round(2, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true)); CAF_CHECK_EQUAL(round_result, make_new_round_result(6, false));
CAF_CHECK_EQUAL(f.result, "0:3,0:6,1:1,1:4,2:2,2:5"); CAF_CHECK_EQUAL(f.result, "0:3,0:6,1:1,1:4,2:2,2:5");
CAF_REQUIRE_EQUAL(queue.empty(), false); CAF_REQUIRE_EQUAL(queue.empty(), false);
// Allow f to consume one more item from each queue. // Allow f to consume one more item from each queue.
f.result.clear(); f.result.clear();
round_result = queue.new_round(1, f); round_result = queue.new_round(1, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true)); CAF_CHECK_EQUAL(round_result, make_new_round_result(3, false));
CAF_CHECK_EQUAL(f.result, "0:9,1:7,2:8"); CAF_CHECK_EQUAL(f.result, "0:9,1:7,2:8");
CAF_REQUIRE_EQUAL(queue.empty(), false); CAF_REQUIRE_EQUAL(queue.empty(), false);
// Allow f to consume the remainder, i.e., 12. // Allow f to consume the remainder, i.e., 12.
f.result.clear(); f.result.clear();
round_result = queue.new_round(1000, f); round_result = queue.new_round(1000, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true)); CAF_CHECK_EQUAL(round_result, make_new_round_result(1, false));
CAF_CHECK_EQUAL(f.result, "0:12"); CAF_CHECK_EQUAL(f.result, "0:12");
CAF_REQUIRE_EQUAL(queue.empty(), true); CAF_REQUIRE_EQUAL(queue.empty(), true);
} }
......
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