Commit c634d7fb authored by Dominik Charousset's avatar Dominik Charousset

Remove the complexity-based credit controller

The idea of a self-adjusting buffer seemed promising. However, computing
buffer space based on processing times easily leads to very large
buffers when entire batches only take microseconds to process. We could
reproduce excessive buffering in many cases, where applications would
use several hundred MB when using this controller.
parent 3ccf2d6e
......@@ -64,7 +64,6 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS}
src/binary_deserializer.cpp
src/binary_serializer.cpp
src/blocking_actor.cpp
src/complexity_based_credit_controller.cpp
src/config_option.cpp
src/config_option_adder.cpp
src/config_option_set.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 "caf/credit_controller.hpp"
namespace caf {
namespace detail {
/// Computes credit for an attached source based on measuring the complexity of
/// incoming batches.
class complexity_based_credit_controller : public credit_controller {
public:
// -- member types -----------------------------------------------------------
using super = credit_controller;
// -- constants --------------------------------------------------------------
/// Stores how many elements we buffer at most after the handshake.
int32_t initial_buffer_size = 50;
/// Stores how many elements we allow per batch after the handshake.
int32_t initial_batch_size = 10;
// -- constructors, destructors, and assignment operators --------------------
explicit complexity_based_credit_controller(scheduled_actor* self);
~complexity_based_credit_controller() override;
// -- overrides --------------------------------------------------------------
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:
// -- 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_;
};
} // 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/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::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 {initial_buffer_size, initial_batch_size};
}
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;
// Give enough credit to last 2 cycles.
result.credit = 2 * 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 caf::detail
......@@ -20,7 +20,6 @@
#include "caf/actor_system_config.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/complexity_based_credit_controller.hpp"
#include "caf/detail/meta_object.hpp"
#include "caf/detail/size_based_credit_controller.hpp"
#include "caf/detail/test_credit_controller.hpp"
......@@ -87,10 +86,8 @@ inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
controller_.reset(new detail::test_credit_controller(self));
else if (*str == "size")
controller_.reset(new detail::size_based_credit_controller(self));
else
controller_.reset(new detail::complexity_based_credit_controller(self));
} else {
controller_.reset(new detail::complexity_based_credit_controller(self));
controller_.reset(new detail::size_based_credit_controller(self));
}
}
......
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