Commit 8972ef8a authored by Dominik Charousset's avatar Dominik Charousset

Add unit test for inbound_path::stats_t

parent 0ba4153a
......@@ -71,7 +71,10 @@ public:
/// whether the destructor sends `close` or `forced_close` messages.
error shutdown_reason;
/// Keep track of measurements for the last 10 batches.
/// Amount of credit we assign sources after receiving `open`.
static constexpr int initial_credit = 50;
/// Keep track of measurements for the last X batches.
static constexpr size_t stats_sampling_size = 16;
/// Stores statistics for measuring complexity of incoming batches.
......@@ -95,7 +98,7 @@ public:
size_t ring_iter;
/// Returns the maximum number of items this actor could handle for given
/// cycle length.
/// cycle length with a minimum of 1.
calculation_result calculate(std::chrono::nanoseconds cycle,
std::chrono::nanoseconds desired_complexity);
......
......@@ -25,10 +25,8 @@
namespace caf {
inbound_path::stats_t::stats_t() {
// we initially assume one item = 1us
// TODO: put constant in some header
measurement x{50, std::chrono::microseconds(50)};
inbound_path::stats_t::stats_t() : ring_iter(0) {
measurement x{0, std::chrono::nanoseconds(0)};
measurements.resize(stats_sampling_size, x);
}
......@@ -44,6 +42,8 @@ auto inbound_path::stats_t::calculate(std::chrono::nanoseconds c,
total_ns += static_cast<long>(x.calculation_time.count());
total_items += x.batch_size;
}
if (total_ns == 0)
return {1, 1};
// Instead of C * (N / t) we calculate N * (C / t) to avoid imprecisions due
// to double rounding for very small numbers.
auto ct = static_cast<double>(c.count()) / total_ns;
......@@ -74,6 +74,7 @@ inbound_path::~inbound_path() {
}
void inbound_path::handle(downstream_msg::batch& x) {
CAF_LOG_TRACE(CAF_ARG(x));
assigned_credit -= x.xs_size;
last_batch_id = x.id;
// TODO: add time-measurement functions to local actor and store taken
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/config.hpp"
#define CAF_SUITE inbound_path
#include "caf/test/unit_test.hpp"
#include <string>
#include "caf/inbound_path.hpp"
using namespace std;
using namespace caf;
// TODO: use std::literals::chrono_literals when switching to C++14
using ns = std::chrono::nanoseconds;
template <class... Ts>
void print(const char* format, Ts... xs) {
char buf[200];
snprintf(buf, 200, format, xs...);
CAF_MESSAGE(buf);
}
struct fixture {
inbound_path::stats_t x;
size_t sampling_size = inbound_path::stats_sampling_size;
fixture() {
CAF_CHECK_EQUAL(x.measurements.size(), sampling_size);
CAF_CHECK_EQUAL(sampling_size % 2, 0u);
}
void calculate(long total_items, long total_time) {
long c = 1000;
long d = 100;
long n = total_items;
long t = total_time;
long m = t > 0 ? std::max((c * n) / t, 1l) : 1l;
long b = t > 0 ? std::max((d * n) / t, 1l) : 1l;
print("with a cycle C = %ldns, desied complexity D = %ld,", c, d);
print("number of items N = %ld, and time delta t = %ld:", n, t);
print("- throughput M = max(C * N / t, 1) = max(%ld * %ld / %ld, 1) = %ld",
c, n, t, m);
print("- items/batch B = max(D * N / t, 1) = max(%ld * %ld / %ld, 1) = %ld",
d, n, t, b);
auto cr = x.calculate(ns(1000), ns(100));
CAF_CHECK_EQUAL(cr.items_per_batch, b);
CAF_CHECK_EQUAL(cr.max_throughput, m);
}
void store(long batch_size, long calculation_time_ns) {
inbound_path::stats_t::measurement m{batch_size, ns{calculation_time_ns}};
x.store(m);
}
};
CAF_TEST_FIXTURE_SCOPE(inbound_path_tests, fixture)
CAF_TEST(default_constructed) {
calculate(0, 0);
}
CAF_TEST(one_store) {
CAF_MESSAGE("store a measurement for 500ns with batch size of 50");
store(50, 500);
calculate(50, 500);
}
CAF_TEST(multiple_stores) {
CAF_MESSAGE("store a measurement: (50, 500ns), (60, 400ns), (40, 600ns)");
store(50, 500);
store(40, 600);
store(60, 400);
calculate(150, 1500);
}
CAF_TEST(overriding_stores) {
CAF_MESSAGE("fill measurements with (100, 1000ns)");
for (size_t i = 0; i < sampling_size; ++i)
store(100, 1000);
calculate(100, 1000);
CAF_MESSAGE("override first half of the measurements with (10, 1000ns)");
for (size_t i = 0; i < sampling_size / 2; ++i)
store(10, 1000);
calculate(55, 1000);
CAF_MESSAGE("override second half of the measurements with (10, 1000ns)");
for (size_t i = 0; i < sampling_size / 2; ++i)
store(10, 1000);
calculate(10, 1000);
}
CAF_TEST_FIXTURE_SCOPE_END()
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