Commit 8ec58ee2 authored by Dominik Charousset's avatar Dominik Charousset

Fix congestion feedback and batch size calculation

parent 3c42b882
...@@ -100,7 +100,7 @@ public: ...@@ -100,7 +100,7 @@ public:
desired); desired);
desired *= 2; desired *= 2;
auto stored = buffered(); auto stored = buffered();
return stored < desired ? desired - stored : 0u; return desired > stored ? desired - stored : 0u;
} }
size_t buffered() const noexcept override { size_t buffered() const noexcept override {
......
...@@ -64,7 +64,7 @@ constexpr auto buffer_capacity = int32_t{64 * 1024}; // 64 KB ...@@ -64,7 +64,7 @@ constexpr auto buffer_capacity = int32_t{64 * 1024}; // 64 KB
/// Frequency of computing the serialized size of incoming batches. Smaller /// Frequency of computing the serialized size of incoming batches. Smaller
/// values may increase accuracy, but also add computational overhead. /// values may increase accuracy, but also add computational overhead.
constexpr auto sampling_rate = int32_t{25}; constexpr auto sampling_rate = int32_t{100};
/// Frequency of re-calibrating batch sizes. For example, a calibration interval /// Frequency of re-calibrating batch sizes. For example, a calibration interval
/// of 10 and a sampling rate of 20 causes the actor to re-calibrate every 200 /// of 10 and a sampling rate of 20 causes the actor to re-calibrate every 200
......
...@@ -19,12 +19,16 @@ ...@@ -19,12 +19,16 @@
#pragma once #pragma once
#include "caf/credit_controller.hpp" #include "caf/credit_controller.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/serialized_size.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/stream.hpp"
namespace caf::detail { namespace caf::detail {
/// A credit controller that estimates the bytes required to store incoming /// A credit controller that estimates the bytes required to store incoming
/// batches and constrains credit based on upper bounds for memory usage. /// batches and constrains credit based on upper bounds for memory usage.
class size_based_credit_controller : public credit_controller { class CAF_CORE_EXPORT size_based_credit_controller : public credit_controller {
public: public:
// -- constants -------------------------------------------------------------- // -- constants --------------------------------------------------------------
...@@ -45,13 +49,35 @@ public: ...@@ -45,13 +49,35 @@ public:
// -- interface functions ---------------------------------------------------- // -- interface functions ----------------------------------------------------
void before_processing(downstream_msg::batch& batch) override;
calibration init() override; calibration init() override;
calibration calibrate() override; calibration calibrate() override;
private: // -- factory functions ------------------------------------------------------
template <class T>
static auto make(local_actor* self, stream<T>) {
class impl : public size_based_credit_controller {
using size_based_credit_controller::size_based_credit_controller;
void before_processing(downstream_msg::batch& x) override {
if (++this->sample_counter_ == this->sampling_rate_) {
this->sample_counter_ = 0;
this->inspector_.result = 0;
this->sampled_elements_ += x.xs_size;
for (auto& element : x.xs.get_as<std::vector<T>>(0)) {
auto res = this->inspector_(element);
static_cast<void>(res); // This inspector never produces an error.
}
this->sampled_total_size_
+= static_cast<int64_t>(this->inspector_.result);
}
}
};
return std::make_unique<impl>(self);
}
protected:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
local_actor* self_; local_actor* self_;
...@@ -69,6 +95,9 @@ private: ...@@ -69,6 +95,9 @@ private:
/// Stores how many bytes the sampled batches required when serialized. /// Stores how many bytes the sampled batches required when serialized.
int64_t sampled_total_size_ = 0; int64_t sampled_total_size_ = 0;
/// Computes how many bytes elements require on the wire.
serialized_size_inspector inspector_;
/// Stores whether this is the first run. /// Stores whether this is the first run.
bool initializing_ = true; bool initializing_ = true;
......
...@@ -61,10 +61,6 @@ public: ...@@ -61,10 +61,6 @@ public:
return driver_.acquire_credit(path, desired); return driver_.acquire_credit(path, desired);
} }
bool congested() const noexcept override {
return driver_.congested();
}
protected: protected:
void finalize(const error& reason) override { void finalize(const error& reason) override {
driver_.finalize(reason); driver_.finalize(reason);
......
...@@ -73,10 +73,6 @@ public: ...@@ -73,10 +73,6 @@ public:
CAF_LOG_ERROR("received unexpected batch type (dropped)"); CAF_LOG_ERROR("received unexpected batch type (dropped)");
} }
bool congested() const noexcept override {
return driver_.congested();
}
int32_t acquire_credit(inbound_path* path, int32_t desired) override { int32_t acquire_credit(inbound_path* path, int32_t desired) override {
return driver_.acquire_credit(path, desired); return driver_.acquire_credit(path, desired);
} }
......
...@@ -19,12 +19,13 @@ ...@@ -19,12 +19,13 @@
#pragma once #pragma once
#include "caf/credit_controller.hpp" #include "caf/credit_controller.hpp"
#include "caf/detail/core_export.hpp"
namespace caf::detail { namespace caf::detail {
/// A credit controller that estimates the bytes required to store incoming /// A credit controller that estimates the bytes required to store incoming
/// batches and constrains credit based on upper bounds for memory usage. /// batches and constrains credit based on upper bounds for memory usage.
class token_based_credit_controller : public credit_controller { class CAF_CORE_EXPORT token_based_credit_controller : public credit_controller {
public: public:
// -- constants -------------------------------------------------------------- // -- constants --------------------------------------------------------------
...@@ -51,6 +52,13 @@ public: ...@@ -51,6 +52,13 @@ public:
calibration calibrate() override; calibration calibrate() override;
// -- factory functions ------------------------------------------------------
template <class T>
static auto make(local_actor* self, stream<T>) {
return std::make_unique<token_based_credit_controller>(self);
}
private: private:
// -- see caf::defaults::stream::token_policy ------------------------------- // -- see caf::defaults::stream::token_policy -------------------------------
......
...@@ -25,10 +25,13 @@ ...@@ -25,10 +25,13 @@
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/credit_controller.hpp" #include "caf/credit_controller.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/size_based_credit_controller.hpp"
#include "caf/detail/token_based_credit_controller.hpp"
#include "caf/downstream_msg.hpp" #include "caf/downstream_msg.hpp"
#include "caf/logger.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
#include "caf/settings.hpp"
#include "caf/stream_aborter.hpp" #include "caf/stream_aborter.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_priority.hpp" #include "caf/stream_priority.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/telemetry/counter.hpp" #include "caf/telemetry/counter.hpp"
...@@ -63,16 +66,34 @@ public: ...@@ -63,16 +66,34 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
/// Constructs a path for given handle and stream ID. template <class T>
inbound_path(stream_manager_ptr mgr_ptr, stream_slots id, inbound_path(stream_manager* mgr, stream<T> in)
strong_actor_ptr ptr, type_id_t input_type); : inbound_path(mgr, type_id_v<T>) {
auto& cfg = config();
auto set_default = [this, in] {
controller_ = detail::size_based_credit_controller::make(self(), in);
};
if (auto str = get_if<std::string>(&cfg, "caf.stream.credit-policy")) {
if (*str == "token-based")
controller_ = detail::token_based_credit_controller::make(self(), in);
else if (*str == "size-based")
set_default();
else
CAF_LOG_WARNING("unrecognized credit policy:"
<< *str << "(falling back to 'size-based')");
} else {
set_default();
}
}
void init(strong_actor_ptr source_hdl, stream_slots id);
~inbound_path(); ~inbound_path();
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Points to the manager responsible for incoming traffic. /// Points to the manager responsible for incoming traffic.
stream_manager_ptr mgr; stream_manager* mgr;
/// Handle to the source. /// Handle to the source.
strong_actor_ptr hdl; strong_actor_ptr hdl;
...@@ -125,18 +146,20 @@ public: ...@@ -125,18 +146,20 @@ public:
/// Returns currently unassigned credit that we could assign to the source. /// Returns currently unassigned credit that we could assign to the source.
int32_t available_credit() const noexcept; int32_t available_credit() const noexcept;
/// Returns the system-wide configuration.
const settings& config() const noexcept;
// -- callbacks -------------------------------------------------------------- // -- callbacks --------------------------------------------------------------
/// Updates `last_batch_id` and `assigned_credit` before dispatching to the /// Updates `last_batch_id` and `assigned_credit` before dispatching to the
/// manager. /// manager.
void handle(downstream_msg::batch& x); void handle(downstream_msg::batch& x);
/// Dispatches any `downstream_msg` other than `batch` directly to the /// Forward the `close` message to the manager.
/// manager. void handle(downstream_msg::close& x);
template <class T>
void handle(T& x) { /// Forward the `forced_close` message to the manager.
mgr->handle(this, x); void handle(downstream_msg::forced_close& x);
}
/// Forces an ACK message after receiving no input for a considerable amount /// Forces an ACK message after receiving no input for a considerable amount
/// of time. /// of time.
...@@ -162,6 +185,9 @@ public: ...@@ -162,6 +185,9 @@ public:
static void static void
emit_irregular_shutdown(local_actor* self, stream_slots slots, emit_irregular_shutdown(local_actor* self, stream_slots slots,
const strong_actor_ptr& hdl, error reason); const strong_actor_ptr& hdl, error reason);
private:
inbound_path(stream_manager* mgr_ptr, type_id_t input_type);
}; };
/// @relates inbound_path /// @relates inbound_path
......
...@@ -532,9 +532,8 @@ public: ...@@ -532,9 +532,8 @@ public:
// -- inbound_path management ------------------------------------------------ // -- inbound_path management ------------------------------------------------
/// Creates a new path for incoming stream traffic from `sender`. /// Creates a new path for incoming stream traffic from `sender`.
virtual inbound_path* virtual bool add_inbound_path(type_id_t input_type,
make_inbound_path(stream_manager_ptr mgr, stream_slots slots, std::unique_ptr<inbound_path> path);
strong_actor_ptr sender, type_id_t rtti);
/// Silently closes incoming stream traffic on `slot`. /// Silently closes incoming stream traffic on `slot`.
virtual void erase_inbound_path_later(stream_slot slot); virtual void erase_inbound_path_later(stream_slot slot);
......
...@@ -19,10 +19,7 @@ ...@@ -19,10 +19,7 @@
#pragma once #pragma once
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/invalid_stream.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_slot.hpp"
#include "caf/type_id.hpp" #include "caf/type_id.hpp"
namespace caf { namespace caf {
......
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/downstream_msg.hpp" #include "caf/downstream_msg.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/inbound_path.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/stream.hpp" #include "caf/stream.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
...@@ -56,6 +57,8 @@ public: ...@@ -56,6 +57,8 @@ public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using inbound_path_ptr = std::unique_ptr<inbound_path>;
using inbound_paths_list = std::vector<inbound_path*>; using inbound_paths_list = std::vector<inbound_path*>;
/// Discrete point in time. /// Discrete point in time.
...@@ -147,6 +150,16 @@ public: ...@@ -147,6 +150,16 @@ public:
/// This function is called from the destructor of `inbound_path`. /// This function is called from the destructor of `inbound_path`.
virtual void deregister_input_path(inbound_path* x) noexcept; virtual void deregister_input_path(inbound_path* x) noexcept;
/// Creates an inbound path to the current sender without any type checking.
/// @pre `current_sender() != nullptr`
/// @pre `out().terminal() == false`
/// @private
template <class In>
stream_slot add_unchecked_inbound_path(stream<In> in) {
auto path = std::make_unique<inbound_path>(this, in);
return add_unchecked_inbound_path_impl(type_id_v<In>, std::move(path));
}
/// Removes an input path /// Removes an input path
virtual void remove_input_path(stream_slot slot, error reason, bool silent); virtual void remove_input_path(stream_slot slot, error reason, bool silent);
...@@ -188,7 +201,7 @@ public: ...@@ -188,7 +201,7 @@ public:
bool inbound_paths_idle() const noexcept; bool inbound_paths_idle() const noexcept;
/// Returns the parent actor. /// Returns the parent actor.
scheduled_actor* self() { scheduled_actor* self() noexcept {
return self_; return self_;
} }
...@@ -246,15 +259,6 @@ public: ...@@ -246,15 +259,6 @@ public:
std::move(handshake)); std::move(handshake));
} }
/// Creates an inbound path to the current sender without any type checking.
/// @pre `current_sender() != nullptr`
/// @pre `out().terminal() == false`
/// @private
template <class In>
stream_slot add_unchecked_inbound_path(stream<In>) {
return add_unchecked_inbound_path_impl(type_id_v<In>);
}
/// Adds a new outbound path to `rp.next()`. /// Adds a new outbound path to `rp.next()`.
/// @private /// @private
stream_slot stream_slot
...@@ -269,10 +273,6 @@ public: ...@@ -269,10 +273,6 @@ public:
/// @private /// @private
stream_slot add_unchecked_outbound_path_impl(message handshake); stream_slot add_unchecked_outbound_path_impl(message handshake);
/// Adds the current sender as an inbound path.
/// @pre Current message is an `open_stream_msg`.
stream_slot add_unchecked_inbound_path_impl(type_id_t rtti);
// -- time management -------------------------------------------------------- // -- time management --------------------------------------------------------
void tick(time_point now); void tick(time_point now);
...@@ -304,6 +304,11 @@ protected: ...@@ -304,6 +304,11 @@ protected:
/// implementation does nothing. /// implementation does nothing.
virtual void output_closed(error reason); virtual void output_closed(error reason);
/// Adds the current sender as an inbound path.
/// @pre Current message is an `open_stream_msg`.
stream_slot add_unchecked_inbound_path_impl(type_id_t input_type,
inbound_path_ptr path);
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Points to the parent actor. /// Points to the parent actor.
......
...@@ -62,8 +62,8 @@ public: ...@@ -62,8 +62,8 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Creates a new input path to the current sender. /// Creates a new input path to the current sender.
inbound_stream_slot<input_type> add_inbound_path(stream<input_type>) { inbound_stream_slot<input_type> add_inbound_path(stream<input_type> in) {
return {this->add_unchecked_inbound_path_impl(type_id_v<input_type>)}; return {this->add_unchecked_inbound_path(in)};
} }
private: private:
......
...@@ -60,12 +60,6 @@ public: ...@@ -60,12 +60,6 @@ public:
/// Processes a single batch. /// Processes a single batch.
virtual void process(std::vector<input_type>& batch) = 0; virtual void process(std::vector<input_type>& batch) = 0;
/// Can mark the sink as congested, e.g., when writing into a buffer that
/// fills up faster than it is drained.
virtual bool congested() const noexcept {
return false;
}
/// Acquires credit on an inbound path. The calculated credit to fill our /// Acquires credit on an inbound path. The calculated credit to fill our
/// queue for two cycles is `desired`, but the driver is allowed to return /// queue for two cycles is `desired`, but the driver is allowed to return
/// any non-negative value. /// any non-negative value.
......
...@@ -73,12 +73,6 @@ public: ...@@ -73,12 +73,6 @@ public:
// nop // nop
} }
/// Can mark the stage as congested. The default implementation signals a
/// congestion if the downstream manager has no capacity left in its buffer.
virtual bool congested() const noexcept {
return out_.capacity() == 0;
}
/// Acquires credit on an inbound path. The calculated credit to fill our /// Acquires credit on an inbound path. The calculated credit to fill our
/// queue for two cycles is `desired`, but the driver is allowed to return /// queue for two cycles is `desired`, but the driver is allowed to return
/// any non-negative value. /// any non-negative value.
......
...@@ -34,7 +34,7 @@ constexpr int32_t initial_sample_size = 10; ...@@ -34,7 +34,7 @@ constexpr int32_t initial_sample_size = 10;
namespace caf::detail { namespace caf::detail {
size_based_credit_controller::size_based_credit_controller(local_actor* ptr) size_based_credit_controller::size_based_credit_controller(local_actor* ptr)
: self_(ptr) { : self_(ptr), inspector_(ptr->system()) {
namespace fallback = defaults::stream::size_policy; namespace fallback = defaults::stream::size_policy;
// Initialize from the config parameters. // Initialize from the config parameters.
auto& cfg = ptr->system().config(); auto& cfg = ptr->system().config();
...@@ -59,18 +59,6 @@ size_based_credit_controller::~size_based_credit_controller() { ...@@ -59,18 +59,6 @@ size_based_credit_controller::~size_based_credit_controller() {
// nop // nop
} }
void size_based_credit_controller::before_processing(downstream_msg::batch& x) {
if (++sample_counter_ == sampling_rate_) {
sample_counter_ = 0;
sampled_elements_ += x.xs_size;
// FIXME: this wildly overestimates the actual size per element, because we
// include the size of the meta data per message. This also punishes
// small batches and we probably never reach a stable point even if
// incoming data always has the exact same size per element.
sampled_total_size_ += static_cast<int64_t>(serialized_size(x.xs));
}
}
credit_controller::calibration size_based_credit_controller::init() { credit_controller::calibration size_based_credit_controller::init() {
// Initially, we simply assume that the size of one element equals // Initially, we simply assume that the size of one element equals
// bytes-per-batch. // bytes-per-batch.
...@@ -90,7 +78,7 @@ credit_controller::calibration size_based_credit_controller::calibrate() { ...@@ -90,7 +78,7 @@ credit_controller::calibration size_based_credit_controller::calibrate() {
return static_cast<int32_t>(x); return static_cast<int32_t>(x);
}; };
if (!initializing_) { if (!initializing_) {
auto bpe = clamp_i32(sampled_total_size_ / calibration_interval_); auto bpe = clamp_i32(sampled_total_size_ / sampled_elements_);
bytes_per_element_ = static_cast<int32_t>( bytes_per_element_ = static_cast<int32_t>(
smoothing_factor_ * bpe // weighted current measurement smoothing_factor_ * bpe // weighted current measurement
+ (1.0 - smoothing_factor_) * bytes_per_element_ // past values + (1.0 - smoothing_factor_) * bytes_per_element_ // past values
...@@ -99,10 +87,11 @@ credit_controller::calibration size_based_credit_controller::calibrate() { ...@@ -99,10 +87,11 @@ credit_controller::calibration size_based_credit_controller::calibrate() {
// After our first run, we continue with the actual sampling rate. // After our first run, we continue with the actual sampling rate.
initializing_ = false; initializing_ = false;
sampling_rate_ = get_or(self_->config(), sampling_rate_ = get_or(self_->config(),
"caf.stream.size-policy.sample-rate", "caf.stream.size-based-policy.sampling-rate",
defaults::stream::size_policy::sampling_rate); defaults::stream::size_policy::sampling_rate);
bytes_per_element_ = clamp_i32(sampled_total_size_ / initial_sample_size); bytes_per_element_ = clamp_i32(sampled_total_size_ / sampled_elements_);
} }
sampled_elements_ = 0;
sampled_total_size_ = 0; sampled_total_size_ = 0;
return { return {
clamp_i32(buffer_capacity_ / bytes_per_element_), clamp_i32(buffer_capacity_ / bytes_per_element_),
......
...@@ -21,8 +21,6 @@ ...@@ -21,8 +21,6 @@
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/detail/meta_object.hpp" #include "caf/detail/meta_object.hpp"
#include "caf/detail/size_based_credit_controller.hpp"
#include "caf/detail/token_based_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"
...@@ -43,10 +41,12 @@ namespace caf { ...@@ -43,10 +41,12 @@ namespace caf {
// -- constructors, destructors, and assignment operators ---------------------- // -- constructors, destructors, and assignment operators ----------------------
inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id, inbound_path::inbound_path(stream_manager* ptr, type_id_t in_type) : mgr(ptr) {
strong_actor_ptr ptr, // Note: we can't include stream_manager.hpp in the header of inbound_path,
[[maybe_unused]] type_id_t in_type) // because that would cause a circular reference. Hence, we also can't use an
: mgr(std::move(mgr_ptr)), hdl(std::move(ptr)), slots(id) { // intrusive_ptr as member and instead call `ref/deref` manually.
CAF_ASSERT(mgr != nullptr);
mgr->ref();
auto self = mgr->self(); auto self = mgr->self();
auto [processed_elements, input_buffer_size] auto [processed_elements, input_buffer_size]
= self->inbound_stream_metrics(in_type); = self->inbound_stream_metrics(in_type);
...@@ -54,23 +54,18 @@ inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id, ...@@ -54,23 +54,18 @@ inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
mgr->register_input_path(this); mgr->register_input_path(this);
CAF_STREAM_LOG_DEBUG(self->name() CAF_STREAM_LOG_DEBUG(self->name()
<< "opens input stream with element type" << "opens input stream with element type"
<< detail::global_meta_object(in_type)->type_name << detail::global_meta_object(in_type)->type_name);
<< "at slot" << id.receiver << "from" << hdl);
auto setter = set_controller<detail::size_based_credit_controller>;
if (auto str = get_if<std::string>(&self->system().config(),
"caf.stream.credit-policy")) {
if (*str == "token-based")
setter = set_controller<detail::token_based_credit_controller>;
else if (*str != "size-based")
CAF_LOG_WARNING("unrecognized credit policy:"
<< *str << "(falling back to 'size-based')");
}
setter(controller_, self);
last_ack_time = self->now(); last_ack_time = self->now();
} }
inbound_path::~inbound_path() { inbound_path::~inbound_path() {
mgr->deregister_input_path(this); mgr->deregister_input_path(this);
mgr->deref();
}
void inbound_path::init(strong_actor_ptr source_hdl, stream_slots id) {
hdl = std::move(source_hdl);
slots = id;
} }
// -- properties --------------------------------------------------------------- // -- properties ---------------------------------------------------------------
...@@ -89,6 +84,10 @@ int32_t inbound_path::available_credit() const noexcept { ...@@ -89,6 +84,10 @@ int32_t inbound_path::available_credit() const noexcept {
return std::max(max_credit - assigned_credit, int32_t{0}); return std::max(max_credit - assigned_credit, int32_t{0});
} }
const settings& inbound_path::config() const noexcept {
return content(mgr->self()->config());
}
// -- callbacks ---------------------------------------------------------------- // -- callbacks ----------------------------------------------------------------
void inbound_path::handle(downstream_msg::batch& batch) { void inbound_path::handle(downstream_msg::batch& batch) {
...@@ -127,6 +126,14 @@ void inbound_path::tick(time_point now, duration_type max_batch_delay) { ...@@ -127,6 +126,14 @@ void inbound_path::tick(time_point now, duration_type max_batch_delay) {
} }
} }
void inbound_path::handle(downstream_msg::close& x) {
mgr->handle(this, x);
}
void inbound_path::handle(downstream_msg::forced_close& x) {
mgr->handle(this, x);
}
// -- messaging ---------------------------------------------------------------- // -- messaging ----------------------------------------------------------------
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) {
...@@ -134,9 +141,11 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) { ...@@ -134,9 +141,11 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
// Update state. // Update state.
auto [cmax, bsize, countdown] = controller_->init(); auto [cmax, bsize, countdown] = controller_->init();
max_credit = cmax; max_credit = cmax;
assigned_credit = mgr->acquire_credit(this, cmax);
desired_batch_size = bsize; desired_batch_size = bsize;
calibration_countdown = countdown; calibration_countdown = countdown;
if (auto available = available_credit(); available > 0)
if (auto acquired = mgr->acquire_credit(this, available); acquired > 0)
assigned_credit += acquired;
// 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);
......
...@@ -22,6 +22,7 @@ ...@@ -22,6 +22,7 @@
#include "caf/downstream_msg.hpp" #include "caf/downstream_msg.hpp"
#include "caf/inbound_path.hpp" #include "caf/inbound_path.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/stream_manager.hpp"
namespace caf::policy { namespace caf::policy {
......
...@@ -334,8 +334,8 @@ intrusive::task_result scheduled_actor::mailbox_visitor::operator()( ...@@ -334,8 +334,8 @@ intrusive::task_result scheduled_actor::mailbox_visitor::operator()(
processed_elements->inc(num_elements); processed_elements->inc(num_elements);
input_buffer_size->dec(num_elements); input_buffer_size->dec(num_elements);
} }
// Do *not* store a reference here since we potentially reset `inptr`. // Hold onto a strong reference since we might reset `inptr`.
auto mgr = inptr->mgr; auto mgr = stream_manager_ptr{inptr->mgr};
inptr->handle(content); inptr->handle(content);
// The sender slot can be 0. This is the case for forced_close or // The sender slot can be 0. This is the case for forced_close or
// forced_drop messages from stream aborters. // forced_drop messages from stream aborters.
...@@ -950,20 +950,16 @@ scheduled_actor::urgent_queue& scheduled_actor::get_urgent_queue() { ...@@ -950,20 +950,16 @@ scheduled_actor::urgent_queue& scheduled_actor::get_urgent_queue() {
return get<urgent_queue_index>(mailbox_.queue().queues()); return get<urgent_queue_index>(mailbox_.queue().queues());
} }
inbound_path* scheduled_actor::make_inbound_path(stream_manager_ptr mgr, bool scheduled_actor::add_inbound_path(type_id_t,
stream_slots slots, std::unique_ptr<inbound_path> path) {
strong_actor_ptr sender,
type_id_t input_type) {
static constexpr size_t queue_index = downstream_queue_index; static constexpr size_t queue_index = downstream_queue_index;
using policy_type = policy::downstream_messages::nested; using policy_type = policy::downstream_messages::nested;
auto& qs = get<queue_index>(mailbox_.queue().queues()).queues(); auto& qs = get<queue_index>(mailbox_.queue().queues()).queues();
auto res = qs.emplace(slots.receiver, policy_type{nullptr}); auto res = qs.emplace(path->slots.receiver, policy_type{nullptr});
if (!res.second) if (!res.second)
return nullptr; return false;
auto path = new inbound_path(std::move(mgr), slots, std::move(sender), res.first->second.policy().handler = std::move(path);
input_type); return true;
res.first->second.policy().handler.reset(path);
return path;
} }
void scheduled_actor::erase_inbound_path_later(stream_slot slot) { void scheduled_actor::erase_inbound_path_later(stream_slot slot) {
......
...@@ -158,7 +158,7 @@ void stream_manager::push() { ...@@ -158,7 +158,7 @@ void stream_manager::push() {
} }
bool stream_manager::congested() const noexcept { bool stream_manager::congested() const noexcept {
return false; return out().capacity() == 0;
} }
void stream_manager::deliver_handshake(response_promise& rp, stream_slot slot, void stream_manager::deliver_handshake(response_promise& rp, stream_slot slot,
...@@ -267,7 +267,8 @@ stream_manager::add_unchecked_outbound_path_impl(message handshake) { ...@@ -267,7 +267,8 @@ stream_manager::add_unchecked_outbound_path_impl(message handshake) {
} }
stream_slot stream_slot
stream_manager::add_unchecked_inbound_path_impl(type_id_t input_type) { stream_manager::add_unchecked_inbound_path_impl(type_id_t input_type,
inbound_path_ptr ptr) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto x = self_->current_mailbox_element(); auto x = self_->current_mailbox_element();
if (x == nullptr || !x->content().match_elements<open_stream_msg>()) { if (x == nullptr || !x->content().match_elements<open_stream_msg>()) {
...@@ -290,10 +291,11 @@ stream_manager::add_unchecked_inbound_path_impl(type_id_t input_type) { ...@@ -290,10 +291,11 @@ stream_manager::add_unchecked_inbound_path_impl(type_id_t input_type) {
} }
auto slot = assign_next_slot(); auto slot = assign_next_slot();
stream_slots path_id{osm.slot, slot}; stream_slots path_id{osm.slot, slot};
auto ptr = self_->make_inbound_path(this, path_id, std::move(osm.prev_stage), auto raw_ptr = ptr.get();
input_type); ptr->init(std::move(osm.prev_stage), path_id);
CAF_ASSERT(ptr != nullptr); if (!self_->add_inbound_path(input_type, std::move(ptr)))
ptr->emit_ack_open(self_, actor_cast<actor_addr>(osm.original_stage)); return invalid_stream_slot;
raw_ptr->emit_ack_open(self_, actor_cast<actor_addr>(osm.original_stage));
return slot; return slot;
} }
......
...@@ -200,10 +200,6 @@ public: ...@@ -200,10 +200,6 @@ public:
CAF_LOG_ERROR("received unexpected batch type (dropped)"); CAF_LOG_ERROR("received unexpected batch type (dropped)");
} }
bool congested() const noexcept override {
return out_.capacity() == 0;
}
fused_manager& out() noexcept override { fused_manager& out() noexcept override {
return out_; return out_;
} }
......
...@@ -314,18 +314,16 @@ public: ...@@ -314,18 +314,16 @@ public:
kvp.second->tick(now()); kvp.second->tick(now());
} }
inbound_path* make_inbound_path(stream_manager_ptr mgr, stream_slots slots, virtual bool add_inbound_path(type_id_t input_type,
strong_actor_ptr sender, std::unique_ptr<inbound_path> path) override {
type_id_t input_type) override {
using policy_type = policy::downstream_messages::nested; using policy_type = policy::downstream_messages::nested;
auto res = get<dmsg_id::value>(mbox.queues()) auto res = get<dmsg_id::value>(mbox.queues())
.queues().emplace(slots.receiver, policy_type{nullptr}); .queues()
.emplace(path->slots.receiver, policy_type{nullptr});
if (!res.second) if (!res.second)
return nullptr; return false;
auto path = new inbound_path(std::move(mgr), slots, std::move(sender), res.first->second.policy().handler = std::move(path);
input_type); return true;
res.first->second.policy().handler.reset(path);
return path;
} }
void erase_inbound_path_later(stream_slot slot) override { void erase_inbound_path_later(stream_slot slot) override {
......
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