Commit 41ec1379 authored by Dominik Charousset's avatar Dominik Charousset

Take (configurable) measurements in inbound path

Measure elapsed time for processing each batch using the actor clock.
Also, allow the clock to consider the measured qantity and make the test
actor clock configurable.
parent 175574fa
...@@ -49,9 +49,10 @@ public: ...@@ -49,9 +49,10 @@ public:
virtual time_point now() const noexcept; virtual time_point now() const noexcept;
/// Returns the difference between `t0` and `t1`, allowing the clock to /// Returns the difference between `t0` and `t1`, allowing the clock to
/// return any arbitrary value depending on the measurement that took place. /// return an arbitrary value depending on the measurement that took place
virtual duration_type difference(atom_value measurement, time_point t0, /// and the units measured.
time_point t1) const noexcept; virtual duration_type difference(atom_value measurement, long units,
time_point t0, time_point t1) const noexcept;
/// Schedules a `timeout_msg` for `self` at time point `t`, overriding any /// Schedules a `timeout_msg` for `self` at time point `t`, overriding any
/// previous receive timeout. /// previous receive timeout.
......
...@@ -30,6 +30,9 @@ public: ...@@ -30,6 +30,9 @@ public:
time_point now() const noexcept override; time_point now() const noexcept override;
duration_type difference(atom_value measurement, long units, time_point t0,
time_point t1) const noexcept override;
/// Tries to dispatch the next timeout or delayed message regardless of its /// Tries to dispatch the next timeout or delayed message regardless of its
/// timestamp. Returns `false` if `schedule().empty()`, otherwise `true`. /// timestamp. Returns `false` if `schedule().empty()`, otherwise `true`.
bool dispatch_once(); bool dispatch_once();
...@@ -40,6 +43,12 @@ public: ...@@ -40,6 +43,12 @@ public:
/// Advances the time by `x` and dispatches timeouts and delayed messages. /// Advances the time by `x` and dispatches timeouts and delayed messages.
void advance_time(duration_type x); void advance_time(duration_type x);
/// Configures the returned value for `difference`. For example, inserting
/// `('foo', 120ns)` causes the clock to return `units * 120ns` for any call
/// to `difference` with `measurement == 'foo'` regardless of the time points
/// passed to the function.
std::map<atom_value, duration_type> time_per_unit;
}; };
} // namespace detail } // namespace detail
......
...@@ -33,7 +33,7 @@ actor_clock::time_point actor_clock::now() const noexcept { ...@@ -33,7 +33,7 @@ actor_clock::time_point actor_clock::now() const noexcept {
} }
actor_clock::duration_type actor_clock::duration_type
actor_clock::difference(atom_value, time_point t0, actor_clock::difference(atom_value, long, time_point t0,
time_point t1) const noexcept { time_point t1) const noexcept {
return t1 - t0; return t1 - t0;
} }
......
...@@ -74,11 +74,15 @@ inbound_path::~inbound_path() { ...@@ -74,11 +74,15 @@ inbound_path::~inbound_path() {
void inbound_path::handle(downstream_msg::batch& x) { void inbound_path::handle(downstream_msg::batch& x) {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
assigned_credit -= x.xs_size; auto batch_size = x.xs_size;
assigned_credit -= batch_size;
last_batch_id = x.id; last_batch_id = x.id;
// TODO: add time-measurement functions to local actor and store taken auto& clock = mgr->self()->clock();
// measurement in stats auto t0 = clock.now();
mgr->handle(this, x); mgr->handle(this, x);
auto t1 = clock.now();
auto dt = clock.difference(atom("batch"), batch_size, t0, t1);
stats.store({batch_size, dt});
mgr->push(); mgr->push();
} }
......
...@@ -25,6 +25,15 @@ test_actor_clock::time_point test_actor_clock::now() const noexcept { ...@@ -25,6 +25,15 @@ test_actor_clock::time_point test_actor_clock::now() const noexcept {
return current_time; return current_time;
} }
test_actor_clock::duration_type
test_actor_clock::difference(atom_value measurement, long units, time_point t0,
time_point t1) const noexcept {
auto i = time_per_unit.find(measurement);
if (i != time_per_unit.end())
return units * i->second;
return t0 == t1 ? duration_type{1} : t1 - t0;
}
bool test_actor_clock::dispatch_once() { bool test_actor_clock::dispatch_once() {
if (schedule_.empty()) if (schedule_.empty())
return false; return false;
......
...@@ -143,7 +143,10 @@ TESTEE(broken_sink) { ...@@ -143,7 +143,10 @@ TESTEE(broken_sink) {
} }
struct fixture : test_coordinator_fixture<> { struct fixture : test_coordinator_fixture<> {
fixture() {
// Configure the clock to measure each batch item with 1us.
sched.clock().time_per_unit.emplace(atom("batch"), timespan{1000});
}
}; };
error fail_state(const actor& x) { error fail_state(const actor& x) {
......
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