Commit 49f80aa9 authored by Dominik Charousset's avatar Dominik Charousset

Make credit computation configurable

parent c05c1e0e
...@@ -46,10 +46,12 @@ set(LIBCAF_CORE_SRCS ...@@ -46,10 +46,12 @@ set(LIBCAF_CORE_SRCS
src/binary_deserializer.cpp src/binary_deserializer.cpp
src/binary_serializer.cpp src/binary_serializer.cpp
src/blocking_actor.cpp src/blocking_actor.cpp
src/complexity_based_credit_controller.cpp
src/config_option.cpp src/config_option.cpp
src/config_option_adder.cpp src/config_option_adder.cpp
src/config_option_set.cpp src/config_option_set.cpp
src/config_value.cpp src/config_value.cpp
src/credit_controller.cpp
src/decorator/sequencer.cpp src/decorator/sequencer.cpp
src/default_attachable.cpp src/default_attachable.cpp
src/defaults.cpp src/defaults.cpp
...@@ -144,6 +146,7 @@ set(LIBCAF_CORE_SRCS ...@@ -144,6 +146,7 @@ set(LIBCAF_CORE_SRCS
src/string_algorithms.cpp src/string_algorithms.cpp
src/string_view.cpp src/string_view.cpp
src/term.cpp src/term.cpp
src/test_credit_controller.cpp
src/thread_hook.cpp src/thread_hook.cpp
src/timestamp.cpp src/timestamp.cpp
src/tracing_data.cpp src/tracing_data.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <cstdint>
#include "caf/downstream_msg.hpp"
#include "caf/fwd.hpp"
namespace caf {
/// Computes credit for an attached source.
class credit_controller {
public:
// -- member types -----------------------------------------------------------
/// Wraps an assignment of the controller to its source.
struct assignment {
/// Store how much credit we assign to the source.
int32_t credit;
/// Store how many elements we demand per batch.
int32_t batch_size;
};
// -- constructors, destructors, and assignment operators --------------------
explicit credit_controller(scheduled_actor* self);
virtual ~credit_controller();
// -- properties -------------------------------------------------------------
scheduled_actor* self() noexcept {
return self_;
}
// -- pure virtual functions -------------------------------------------------
/// Called before processing the batch `x` in order to allow the controller
/// to keep statistics on incoming batches.
virtual void before_processing(downstream_msg::batch& x) = 0;
/// Called after processing the batch `x` in order to allow the controller to
/// keep statistics on incoming batches.
/// @note The consumer may alter the batch while processing it, for example
/// by moving each element of the batch downstream.
virtual void after_processing(downstream_msg::batch& x) = 0;
/// Assigns initial credit during the stream handshake.
/// @returns The initial credit for the new sources.
virtual assignment compute_initial() = 0;
/// Computes a credit assignment to the source after a cycle ends.
virtual assignment compute(timespan cycle) = 0;
// -- virtual functions ------------------------------------------------------
/// Computes a credit assignment to the source after crossing the
/// low-threshold. May assign zero credit.
virtual assignment compute_bridge();
/// Returns the threshold for when we may give extra credit to a source
/// during a cycle.
/// @returns A negative value if the controller never grants bridge credit,
/// otherwise the threshold for calling `compute_bridge` to generate
/// additional credit.
virtual int32_t low_threshold() const noexcept;
private:
// -- member variables -------------------------------------------------------
/// Points to the parent system.
scheduled_actor* self_;
};
} // namespace caf
...@@ -36,6 +36,7 @@ namespace stream { ...@@ -36,6 +36,7 @@ namespace stream {
extern const timespan desired_batch_complexity; extern const timespan desired_batch_complexity;
extern const timespan max_batch_delay; extern const timespan max_batch_delay;
extern const timespan credit_round_interval; extern const timespan credit_round_interval;
extern const atom_value credit_policy;
} // namespace streaming } // namespace streaming
......
...@@ -5,7 +5,7 @@ ...@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework * * | |___ / ___ \| _| Framework *
* \____/_/ \_|_| * * \____/_/ \_|_| *
* * * *
* Copyright 2011-2018 Dominik Charousset * * Copyright 2011-2019 Dominik Charousset *
* * * *
* Distributed under the terms and conditions of the BSD 3-Clause License or * * 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 * * (at your option) under the terms and conditions of the Boost Software *
...@@ -16,75 +16,53 @@ ...@@ -16,75 +16,53 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/config.hpp" #pragma once
#define CAF_SUITE inbound_path #include "caf/credit_controller.hpp"
#include "caf/test/unit_test.hpp"
namespace caf {
#include <string> namespace detail {
#include "caf/inbound_path.hpp" /// Computes credit for an attached source based on measuring the complexity of
/// incoming batches.
using namespace std; class complexity_based_credit_controller : public credit_controller {
using namespace caf; public:
// -- member types -----------------------------------------------------------
namespace {
using super = credit_controller;
#define PRINT(format, ...) \
{ \ // -- constructors, destructors, and assignment operators --------------------
char buf[200]; \
snprintf(buf, 200, format, __VA_ARGS__); \ explicit complexity_based_credit_controller(scheduled_actor* self);
CAF_MESSAGE(buf); \
}
struct fixture {
inbound_path::stats_t x;
void calculate(int32_t total_items, int32_t total_time) {
int32_t c = 1000;
int32_t d = 100;
int32_t n = total_items;
int32_t t = total_time;
int32_t m = t > 0 ? std::max((c * n) / t, 1) : 1;
int32_t b = t > 0 ? std::max((d * n) / t, 1) : 1;
PRINT("with a cycle C = %dns, desied complexity D = %d,", c, d);
PRINT("number of items N = %d, and time delta t = %d:", n, t);
PRINT("- throughput M = max(C * N / t, 1) = max(%d * %d / %d, 1) = %d",
c, n, t, m);
PRINT("- items/batch B = max(D * N / t, 1) = max(%d * %d / %d, 1) = %d",
d, n, t, b);
auto cr = x.calculate(timespan(c), timespan(d));
CAF_CHECK_EQUAL(cr.items_per_batch, b);
CAF_CHECK_EQUAL(cr.max_throughput, m);
}
void store(int32_t batch_size, int32_t calculation_time_ns) {
inbound_path::stats_t::measurement m{batch_size,
timespan{calculation_time_ns}};
x.store(m);
}
};
} // namespace ~complexity_based_credit_controller() override;
CAF_TEST_FIXTURE_SCOPE(inbound_path_tests, fixture) // -- implementation of virtual functions ------------------------------------
CAF_TEST(default_constructed) { void before_processing(downstream_msg::batch& x) override;
calculate(0, 0);
}
CAF_TEST(one_store) { void after_processing(downstream_msg::batch& x) override;
CAF_MESSAGE("store a measurement for 500ns with batch size of 50");
store(50, 500);
calculate(50, 500);
}
CAF_TEST(multiple_stores) { assignment compute_initial() override;
CAF_MESSAGE("store measurements: (50, 500ns), (60, 400ns), (40, 600ns)");
store(50, 500); assignment compute(timespan cycle) override;
store(40, 600);
store(60, 400); private:
calculate(150, 1500); // -- member variables -------------------------------------------------------
}
/// Total number of elements in all processed batches in the current cycle.
int64_t num_elements_ = 0;
/// Elapsed time for processing all elements of all batches in the current
/// cycle.
timespan processing_time_;
/// Timestamp of the last call to `before_processing`.
timestamp processing_begin_;
/// Stores the desired per-batch complexity.
timespan complexity_;
};
CAF_TEST_FIXTURE_SCOPE_END() } // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/credit_controller.hpp"
namespace caf {
namespace detail {
/// Computes predictable credit in unit tests.
class test_credit_controller : public credit_controller {
public:
// -- member types -----------------------------------------------------------
using super = credit_controller;
// -- constructors, destructors, and assignment operators --------------------
using super::super;
~test_credit_controller() override;
// -- implementation of virtual functions ------------------------------------
void before_processing(downstream_msg::batch& x) override;
void after_processing(downstream_msg::batch& x) override;
assignment compute_initial() override;
assignment compute(timespan cycle) override;
private:
/// Total number of elements in all processed batches in the current cycle.
int64_t num_elements_ = 0;
};
} // namespace detail
} // namespace caf
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/actor_clock.hpp" #include "caf/actor_clock.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/credit_controller.hpp"
#include "caf/downstream_msg.hpp" #include "caf/downstream_msg.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
#include "caf/rtti_pair.hpp" #include "caf/rtti_pair.hpp"
...@@ -68,49 +69,8 @@ public: ...@@ -68,49 +69,8 @@ public:
/// ID of the last received batch. /// ID of the last received batch.
int64_t last_batch_id; int64_t last_batch_id;
/// Amount of credit we assign sources after receiving `open`. /// Controller for assigning credit to the source.
static constexpr int initial_credit = 50; std::unique_ptr<credit_controller> controller_;
/// Stores statistics for measuring complexity of incoming batches.
struct stats_t {
/// Wraps a time measurement for a single processed batch.
struct measurement {
/// Number of items in the batch.
int32_t batch_size;
/// Elapsed time for processing all elements of the batch.
timespan calculation_time;
};
/// Wraps the resulf of `stats_t::calculate()`.
struct calculation_result {
/// Number of items per credit cycle.
int32_t max_throughput;
/// Number of items per batch to reach the desired batch complexity.
int32_t items_per_batch;
};
/// Total number of elements in all processed batches.
int64_t num_elements;
/// Elapsed time for processing all elements of all batches.
timespan processing_time;
stats_t();
/// Returns the maximum number of items this actor could handle for given
/// cycle length with a minimum of 1.
calculation_result calculate(timespan cycle, timespan desired_complexity);
/// Adds a measurement to this statistic.
void store(measurement x);
/// Resets this statistic.
void reset();
};
/// Summarizes how many elements we processed during the last cycle and how
/// much time we spent processing those elements.
stats_t stats;
/// Stores the time point of the last credit decision for this source. /// Stores the time point of the last credit decision for this source.
actor_clock::time_point last_credit_decision; actor_clock::time_point last_credit_decision;
...@@ -146,10 +106,8 @@ public: ...@@ -146,10 +106,8 @@ public:
/// waiting in the mailbox. /// waiting in the mailbox.
/// @param now Current timestamp. /// @param now Current timestamp.
/// @param cycle Time between credit rounds. /// @param cycle Time between credit rounds.
/// @param desired_batch_complexity Desired processing time per batch.
void emit_ack_batch(local_actor* self, int32_t queued_items, void emit_ack_batch(local_actor* self, int32_t queued_items,
actor_clock::time_point now, timespan cycle, actor_clock::time_point now, timespan cycle);
timespan desired_batch_complexity);
/// Returns whether the path received no input since last emitting /// Returns whether the path received no input since last emitting
/// `ack_batch`, i.e., `last_acked_batch_id == last_batch_id`. /// `ack_batch`, i.e., `last_acked_batch_id == last_batch_id`.
...@@ -167,6 +125,10 @@ public: ...@@ -167,6 +125,10 @@ public:
error reason); error reason);
private: private:
scheduled_actor* self();
actor_system& system();
actor_clock& clock(); actor_clock& clock();
}; };
...@@ -178,4 +140,3 @@ typename Inspector::return_type inspect(Inspector& f, inbound_path& x) { ...@@ -178,4 +140,3 @@ typename Inspector::return_type inspect(Inspector& f, inbound_path& x) {
} }
} // namespace caf } // namespace caf
...@@ -82,7 +82,9 @@ actor_system_config::actor_system_config() ...@@ -82,7 +82,9 @@ actor_system_config::actor_system_config()
.add<timespan>(stream_max_batch_delay, "max-batch-delay", .add<timespan>(stream_max_batch_delay, "max-batch-delay",
"maximum delay for partial batches") "maximum delay for partial batches")
.add<timespan>(stream_credit_round_interval, "credit-round-interval", .add<timespan>(stream_credit_round_interval, "credit-round-interval",
"time between emitting credit"); "time between emitting credit")
.add<atom_value>("credit-policy",
"selects an algorithm for credit computation");
opt_group{custom_options_, "scheduler"} opt_group{custom_options_, "scheduler"}
.add<atom_value>("policy", "'stealing' (default) or 'sharing'") .add<atom_value>("policy", "'stealing' (default) or 'sharing'")
.add<size_t>("max-threads", "maximum number of worker threads") .add<size_t>("max-threads", "maximum number of worker threads")
...@@ -158,6 +160,7 @@ settings actor_system_config::dump_content() const { ...@@ -158,6 +160,7 @@ settings actor_system_config::dump_content() const {
defaults::stream::max_batch_delay); defaults::stream::max_batch_delay);
put_missing(stream_group, "credit-round-interval", put_missing(stream_group, "credit-round-interval",
defaults::stream::credit_round_interval); defaults::stream::credit_round_interval);
put_missing(stream_group, "credit-policy", defaults::stream::credit_policy);
// -- scheduler parameters // -- scheduler parameters
auto& scheduler_group = result["scheduler"].as_dictionary(); auto& scheduler_group = result["scheduler"].as_dictionary();
put_missing(scheduler_group, "policy", defaults::scheduler::policy); put_missing(scheduler_group, "policy", defaults::scheduler::policy);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/detail/complexity_based_credit_controller.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/scheduled_actor.hpp"
// Safe us some typing and very ugly formatting.
#define impl complexity_based_credit_controller
namespace caf {
namespace detail {
impl::impl(scheduled_actor* self) : super(self) {
auto& cfg = self->system().config();
complexity_ = cfg.stream_desired_batch_complexity;
}
impl::~impl() {
// nop
}
void impl::before_processing(downstream_msg::batch& x) {
num_elements_ += x.xs_size;
processing_begin_ = make_timestamp();
}
void impl::after_processing(downstream_msg::batch&) {
processing_time_ += make_timestamp() - processing_begin_;
}
credit_controller::assignment impl::compute_initial() {
return {50, 10};
}
credit_controller::assignment impl::compute(timespan cycle) {
// Max throughput = C * (N / t), where C = cycle length, N = measured items,
// and t = measured time. Desired batch size is the same formula with D
// (desired complexity) instead of C. We compute our values in 64-bit for
// more precision before truncating to a 32-bit integer type at the end.
int64_t total_ns = processing_time_.count();
if (total_ns == 0)
return {1, 1};
// Helper for truncating a 64-bit integer to a 32-bit integer with a minimum
// value of 1.
auto clamp = [](int64_t x) -> int32_t {
static constexpr auto upper_bound = std::numeric_limits<int32_t>::max();
if (x > upper_bound)
return upper_bound;
if (x <= 0)
return 1;
return static_cast<int32_t>(x);
};
// Instead of C * (N / t) we calculate (C * N) / t to avoid double conversion
// and rounding errors.
assignment result;
result.credit = clamp((cycle.count() * num_elements_) / total_ns);
result.batch_size = clamp((complexity_.count() * num_elements_) / total_ns);
// Reset state and return.
num_elements_ = 0;
processing_time_ = timespan{0};
return result;
}
} // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/credit_controller.hpp"
namespace caf {
credit_controller::credit_controller(scheduled_actor* self) : self_(self) {
// nop
}
credit_controller::~credit_controller() {
// nop
}
credit_controller::assignment credit_controller::compute_bridge() {
return {0, 0};
}
int32_t credit_controller::low_threshold() const noexcept {
return -1;
}
} // namespace caf
...@@ -49,6 +49,7 @@ namespace stream { ...@@ -49,6 +49,7 @@ namespace stream {
const timespan desired_batch_complexity = us(50); const timespan desired_batch_complexity = us(50);
const timespan max_batch_delay = ms(5); const timespan max_batch_delay = ms(5);
const timespan credit_round_interval = ms(10); const timespan credit_round_interval = ms(10);
const atom_value credit_policy = atom("complexity");
} // namespace stream } // namespace stream
......
...@@ -18,59 +18,23 @@ ...@@ -18,59 +18,23 @@
#include "caf/inbound_path.hpp" #include "caf/inbound_path.hpp"
#include "caf/send.hpp" #include "caf/actor_system_config.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/complexity_based_credit_controller.hpp"
#include "caf/detail/test_credit_controller.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/no_stages.hpp" #include "caf/no_stages.hpp"
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/send.hpp"
#include "caf/settings.hpp"
namespace caf { namespace caf {
inbound_path::stats_t::stats_t() : num_elements(0), processing_time(0) {
// nop
}
auto inbound_path::stats_t::calculate(timespan c, timespan d)
-> calculation_result {
// Max throughput = C * (N / t), where C = cycle length, N = measured items,
// and t = measured time. Desired batch size is the same formula with D
// instead of C.
// We compute our values in 64-bit for more precision before truncating to a
// 32-bit integer type at the end.
int64_t total_ns = processing_time.count();
if (total_ns == 0)
return {1, 1};
/// Helper for truncating a 64-bit integer to a 32-bit integer with a minimum
/// value of 1.
auto clamp = [](int64_t x) -> int32_t {
static constexpr auto upper_bound = std::numeric_limits<int32_t>::max();
if (x > upper_bound)
return upper_bound;
if (x <= 0)
return 1;
return static_cast<int32_t>(x);
};
// Instead of C * (N / t) we calculate (C * N) / t to avoid double conversion
// and rounding errors.
return {clamp((c.count() * num_elements) / total_ns),
clamp((d.count() * num_elements) / total_ns)};
}
void inbound_path::stats_t::store(measurement x) {
num_elements += x.batch_size;
processing_time += x.calculation_time;
}
void inbound_path::stats_t::reset() {
num_elements = 0;
processing_time = timespan{0};
}
inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id, inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
strong_actor_ptr ptr, rtti_pair in_type) strong_actor_ptr ptr, rtti_pair in_type)
: mgr(std::move(mgr_ptr)), : mgr(std::move(mgr_ptr)),
hdl(std::move(ptr)), hdl(std::move(ptr)),
slots(id), slots(id),
desired_batch_size(initial_credit),
assigned_credit(0), assigned_credit(0),
prio(stream_priority::normal), prio(stream_priority::normal),
last_acked_batch_id(0), last_acked_batch_id(0),
...@@ -81,6 +45,13 @@ inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id, ...@@ -81,6 +45,13 @@ inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
<< "opens input stream with element type" << "opens input stream with element type"
<< mgr->self()->system().types().portable_name(in_type) << mgr->self()->system().types().portable_name(in_type)
<< "at slot" << id.receiver << "from" << hdl); << "at slot" << id.receiver << "from" << hdl);
switch (atom_uint(get_or(system().config(), "stream.credit-policy",
defaults::stream::credit_policy))) {
case atom_uint("testing"):
controller_.reset(new detail::test_credit_controller(self()));
default:
controller_.reset(new detail::complexity_based_credit_controller(self()));
}
} }
inbound_path::~inbound_path() { inbound_path::~inbound_path() {
...@@ -89,10 +60,8 @@ inbound_path::~inbound_path() { ...@@ -89,10 +60,8 @@ 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(slots) << CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
auto& clk = clock();
auto batch_size = x.xs_size; auto batch_size = x.xs_size;
last_batch_id = x.id; last_batch_id = x.id;
auto t0 = clk.now();
CAF_STREAM_LOG_DEBUG(mgr->self()->name() << "handles batch of size" CAF_STREAM_LOG_DEBUG(mgr->self()->name() << "handles batch of size"
<< batch_size << "on slot" << slots.receiver << "with" << batch_size << "on slot" << slots.receiver << "with"
<< assigned_credit << "assigned credit"); << assigned_credit << "assigned credit");
...@@ -109,32 +78,33 @@ void inbound_path::handle(downstream_msg::batch& x) { ...@@ -109,32 +78,33 @@ void inbound_path::handle(downstream_msg::batch& x) {
assigned_credit -= batch_size; assigned_credit -= batch_size;
CAF_ASSERT(assigned_credit >= 0); CAF_ASSERT(assigned_credit >= 0);
} }
controller_->before_processing(x);
mgr->handle(this, x); mgr->handle(this, x);
auto t1 = clk.now(); controller_->after_processing(x);
auto dt = clk.difference(atom("batch"), batch_size, t0, t1);
stats.store({batch_size, dt});
mgr->push(); mgr->push();
} }
void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) { void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(rebind_from)); CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(rebind_from));
// Update state. // Update state.
assigned_credit = mgr->acquire_credit(this, initial_credit); auto initial = controller_->compute_initial();
assigned_credit = mgr->acquire_credit(this, initial.credit);
CAF_ASSERT(assigned_credit >= 0); CAF_ASSERT(assigned_credit >= 0);
desired_batch_size = std::min(initial.batch_size, assigned_credit);
// Make sure we receive errors from this point on. // Make sure we receive errors from this point on.
stream_aborter::add(hdl, self->address(), slots.receiver, stream_aborter::add(hdl, self->address(), slots.receiver,
stream_aborter::source_aborter); stream_aborter::source_aborter);
// Send message. // Send message.
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<upstream_msg::ack_open>( make<upstream_msg::ack_open>(slots.invert(), self->address(),
slots.invert(), self->address(), std::move(rebind_from), std::move(rebind_from),
self->ctrl(), assigned_credit, desired_batch_size)); self->ctrl(), assigned_credit,
desired_batch_size));
last_credit_decision = clock().now(); last_credit_decision = clock().now();
} }
void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items, void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
actor_clock::time_point now, timespan cycle, actor_clock::time_point now, timespan cycle) {
timespan complexity) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(queued_items) << CAF_ARG(cycle) CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(queued_items) << CAF_ARG(cycle)
<< CAF_ARG(complexity)); << CAF_ARG(complexity));
CAF_IGNORE_UNUSED(queued_items); CAF_IGNORE_UNUSED(queued_items);
...@@ -143,10 +113,9 @@ void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items, ...@@ -143,10 +113,9 @@ void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
next_credit_decision = now + cycle; next_credit_decision = now + cycle;
// Hand out enough credit to fill our queue for 2 cycles but never exceed // Hand out enough credit to fill our queue for 2 cycles but never exceed
// the downstream capacity. // the downstream capacity.
auto x = stats.calculate(cycle, complexity);
auto stats_guard = detail::make_scope_guard([&] { stats.reset(); });
auto& out = mgr->out(); auto& out = mgr->out();
auto max_capacity = std::min(x.max_throughput * 2, out.max_capacity()); auto x = controller_->compute(cycle);
auto max_capacity = std::min(x.credit * 2, out.max_capacity());
CAF_ASSERT(max_capacity > 0); CAF_ASSERT(max_capacity > 0);
// Protect against overflow on `assigned_credit`. // Protect against overflow on `assigned_credit`.
auto max_new_credit = std::numeric_limits<int32_t>::max() - assigned_credit; auto max_new_credit = std::numeric_limits<int32_t>::max() - assigned_credit;
...@@ -170,12 +139,12 @@ void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items, ...@@ -170,12 +139,12 @@ void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
<< CAF_ARG(assigned_credit)); << CAF_ARG(assigned_credit));
if (credit == 0 && up_to_date()) if (credit == 0 && up_to_date())
return; return;
CAF_LOG_DEBUG(CAF_ARG(assigned_credit) << CAF_ARG(max_capacity) CAF_LOG_DEBUG(CAF_ARG(assigned_credit)
<< CAF_ARG(queued_items) << CAF_ARG(credit) << CAF_ARG(max_capacity) << CAF_ARG(queued_items)
<< CAF_ARG(desired_batch_size)); << CAF_ARG(credit) << CAF_ARG(x.batch_size));
assigned_credit += credit; assigned_credit += credit;
CAF_ASSERT(assigned_credit >= 0); CAF_ASSERT(assigned_credit >= 0);
desired_batch_size = static_cast<int32_t>(x.items_per_batch); desired_batch_size = x.batch_size;
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<upstream_msg::ack_batch>(slots.invert(), self->address(), make<upstream_msg::ack_batch>(slots.invert(), self->address(),
static_cast<int32_t>(credit), static_cast<int32_t>(credit),
...@@ -216,6 +185,14 @@ void inbound_path::emit_irregular_shutdown(local_actor* self, ...@@ -216,6 +185,14 @@ void inbound_path::emit_irregular_shutdown(local_actor* self,
std::move(reason))); std::move(reason)));
} }
scheduled_actor* inbound_path::self() {
return mgr->self();
}
actor_system& inbound_path::system() {
return mgr->self()->system();
}
actor_clock& inbound_path::clock() { actor_clock& inbound_path::clock() {
return mgr->self()->clock(); return mgr->self()->clock();
} }
......
...@@ -1158,12 +1158,11 @@ scheduled_actor::advance_streams(actor_clock::time_point now) { ...@@ -1158,12 +1158,11 @@ scheduled_actor::advance_streams(actor_clock::time_point now) {
CAF_LOG_DEBUG("new credit round"); CAF_LOG_DEBUG("new credit round");
auto cycle = stream_ticks_.interval(); auto cycle = stream_ticks_.interval();
cycle *= static_cast<decltype(cycle)::rep>(credit_round_ticks_); cycle *= static_cast<decltype(cycle)::rep>(credit_round_ticks_);
auto bc = home_system().config().stream_desired_batch_complexity;
auto& qs = get_downstream_queue().queues(); auto& qs = get_downstream_queue().queues();
for (auto& kvp : qs) { for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get(); auto inptr = kvp.second.policy().handler.get();
auto bs = static_cast<int32_t>(kvp.second.total_task_size()); auto tts = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(this, bs, now, cycle, bc); inptr->emit_ack_batch(this, tts, now, cycle);
} }
} }
return stream_ticks_.next_timeout(now, {max_batch_delay_ticks_, return stream_ticks_.next_timeout(now, {max_batch_delay_ticks_,
......
...@@ -155,7 +155,6 @@ void stream_manager::advance() { ...@@ -155,7 +155,6 @@ void stream_manager::advance() {
if (!inbound_paths_.empty()) { if (!inbound_paths_.empty()) {
auto now = self_->clock().now(); auto now = self_->clock().now();
auto& cfg = self_->system().config(); auto& cfg = self_->system().config();
auto bc = cfg.stream_desired_batch_complexity;
auto interval = cfg.stream_credit_round_interval; auto interval = cfg.stream_credit_round_interval;
auto& qs = self_->get_downstream_queue().queues(); auto& qs = self_->get_downstream_queue().queues();
// Iterate all queues for inbound traffic. // Iterate all queues for inbound traffic.
...@@ -163,8 +162,8 @@ void stream_manager::advance() { ...@@ -163,8 +162,8 @@ void stream_manager::advance() {
auto inptr = kvp.second.policy().handler.get(); auto inptr = kvp.second.policy().handler.get();
// Ignore inbound paths of other managers. // Ignore inbound paths of other managers.
if (inptr->mgr.get() == this) { if (inptr->mgr.get() == this) {
auto bs = static_cast<int32_t>(kvp.second.total_task_size()); auto tts = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(self_, bs, now, interval, bc); inptr->emit_ack_batch(self_, tts, now, interval);
} }
} }
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/detail/test_credit_controller.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/scheduled_actor.hpp"
namespace caf {
namespace detail {
test_credit_controller::~test_credit_controller() {
// nop
}
void test_credit_controller::before_processing(downstream_msg::batch& x) {
num_elements_ += x.xs_size;
}
void test_credit_controller::after_processing(downstream_msg::batch&) {
// nop
}
credit_controller::assignment test_credit_controller::compute_initial() {
return {50, 50};
}
credit_controller::assignment test_credit_controller::compute(timespan cycle) {
auto& cfg = self()->system().config();
auto complexity = cfg.stream_desired_batch_complexity;
// Max throughput = C * (N / t), where C = cycle length, N = measured items,
// and t = measured time. Desired batch size is the same formula with D
// (desired complexity) instead of C. We compute our values in 64-bit for
// more precision before truncating to a 32-bit integer type at the end.
int64_t total_ns = num_elements_ * 1000; // calculate with 1us per element
if (total_ns == 0)
return {1, 1};
// Helper for truncating a 64-bit integer to a 32-bit integer with a minimum
// value of 1.
auto clamp = [](int64_t x) -> int32_t {
static constexpr auto upper_bound = std::numeric_limits<int32_t>::max();
if (x > upper_bound)
return upper_bound;
if (x <= 0)
return 1;
return static_cast<int32_t>(x);
};
// Instead of C * (N / t) we calculate (C * N) / t to avoid double conversion
// and rounding errors.
assignment result;
result.credit = clamp((cycle.count() * num_elements_) / total_ns);
result.batch_size = clamp((complexity.count() * num_elements_) / total_ns);
// Reset state and return.
num_elements_ = 0;
return result;
}
} // namespace detail
} // namespace caf
...@@ -329,7 +329,6 @@ public: ...@@ -329,7 +329,6 @@ public:
void advance_time() { void advance_time() {
auto cycle = std::chrono::milliseconds(100); auto cycle = std::chrono::milliseconds(100);
auto desired_batch_complexity = std::chrono::microseconds(50);
auto f = [&](tick_type x) { auto f = [&](tick_type x) {
if (x % ticks_per_force_batches_interval == 0) { if (x % ticks_per_force_batches_interval == 0) {
// Force batches on all output paths. // Force batches on all output paths.
...@@ -341,9 +340,8 @@ public: ...@@ -341,9 +340,8 @@ public:
auto& qs = get<dmsg_id::value>(mbox.queues()).queues(); auto& qs = get<dmsg_id::value>(mbox.queues()).queues();
for (auto& kvp : qs) { for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get(); auto inptr = kvp.second.policy().handler.get();
auto bs = static_cast<int32_t>(kvp.second.total_task_size()); auto tts = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(this, bs, now(), cycle, inptr->emit_ack_batch(this, tts, now(), cycle);
desired_batch_complexity);
} }
} }
}; };
......
...@@ -693,6 +693,7 @@ public: ...@@ -693,6 +693,7 @@ public:
cfg.set("middleman.network-backend", caf::atom("testing")); cfg.set("middleman.network-backend", caf::atom("testing"));
cfg.set("middleman.manual-multiplexing", true); cfg.set("middleman.manual-multiplexing", true);
cfg.set("middleman.workers", size_t{0}); cfg.set("middleman.workers", size_t{0});
cfg.set("stream.credit-policy", caf::atom("testing"));
return cfg; return cfg;
} }
......
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