Commit 8484a3b8 authored by Dominik Charousset's avatar Dominik Charousset

Redesign credit controller API

parent c634d7fb
...@@ -96,11 +96,14 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS} ...@@ -96,11 +96,14 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS}
src/detail/set_thread_name.cpp src/detail/set_thread_name.cpp
src/detail/shared_spinlock.cpp src/detail/shared_spinlock.cpp
src/detail/simple_actor_clock.cpp src/detail/simple_actor_clock.cpp
src/detail/size_based_credit_controller.cpp
src/detail/stringification_inspector.cpp src/detail/stringification_inspector.cpp
src/detail/sync_request_bouncer.cpp src/detail/sync_request_bouncer.cpp
src/detail/test_actor_clock.cpp src/detail/test_actor_clock.cpp
src/detail/thread_safe_actor_clock.cpp src/detail/thread_safe_actor_clock.cpp
src/detail/tick_emitter.cpp src/detail/tick_emitter.cpp
src/detail/token_based_credit_controller.cpp
src/detail/token_based_credit_controller.cpp
src/detail/type_id_list_builder.cpp src/detail/type_id_list_builder.cpp
src/detail/uri_impl.cpp src/detail/uri_impl.cpp
src/downstream_manager.cpp src/downstream_manager.cpp
...@@ -155,7 +158,6 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS} ...@@ -155,7 +158,6 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS}
src/sec_strings.cpp src/sec_strings.cpp
src/serializer.cpp src/serializer.cpp
src/settings.cpp src/settings.cpp
src/size_based_credit_controller.cpp
src/skip.cpp src/skip.cpp
src/stream_aborter.cpp src/stream_aborter.cpp
src/stream_manager.cpp src/stream_manager.cpp
...@@ -169,7 +171,6 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS} ...@@ -169,7 +171,6 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS}
src/telemetry/metric_family.cpp src/telemetry/metric_family.cpp
src/telemetry/metric_registry.cpp src/telemetry/metric_registry.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
......
...@@ -204,18 +204,12 @@ public: ...@@ -204,18 +204,12 @@ public:
// -- stream parameters ------------------------------------------------------ // -- stream parameters ------------------------------------------------------
/// @private
timespan stream_desired_batch_complexity;
/// @private /// @private
timespan stream_max_batch_delay; timespan stream_max_batch_delay;
/// @private /// @private
timespan stream_credit_round_interval; timespan stream_credit_round_interval;
/// @private
timespan stream_tick_duration() const noexcept;
// -- OpenSSL parameters ----------------------------------------------------- // -- OpenSSL parameters -----------------------------------------------------
std::string openssl_certificate; std::string openssl_certificate;
......
...@@ -295,7 +295,10 @@ private: ...@@ -295,7 +295,10 @@ private:
} }
auto new_size = buffered(); auto new_size = buffered();
CAF_ASSERT(old_size >= new_size); CAF_ASSERT(old_size >= new_size);
this->shipped_messages(old_size - new_size); auto shipped = old_size - new_size;
this->shipped_messages(shipped);
if (shipped > 0)
this->last_send_ = this->self()->now();
} }
state_map_type state_map_; state_map_type state_map_;
......
...@@ -32,64 +32,34 @@ public: ...@@ -32,64 +32,34 @@ public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
/// Wraps an assignment of the controller to its source. /// Wraps an assignment of the controller to its source.
struct CAF_CORE_EXPORT assignment { struct calibration {
/// Stores how much credit we assign to the source. /// Stores how much credit the path may emit at most.
int32_t credit; int32_t max_credit;
/// Stores how many elements we demand per batch. /// Stores how many elements we demand per batch.
int32_t batch_size; int32_t batch_size;
/// Stores how many batches the caller should wait before calling
/// `calibrate` again.
int32_t next_calibration;
}; };
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit credit_controller(scheduled_actor* self);
virtual ~credit_controller(); virtual ~credit_controller();
// -- properties -------------------------------------------------------------
scheduled_actor* self() noexcept {
return self_;
}
// -- pure virtual functions ------------------------------------------------- // -- pure virtual functions -------------------------------------------------
/// Called before processing the batch `x` in order to allow the controller /// Called before processing the batch `x` in order to allow the controller
/// to keep statistics on incoming batches. /// to keep statistics on incoming batches.
virtual void before_processing(downstream_msg::batch& x) = 0; virtual void before_processing(downstream_msg::batch& batch) = 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 an initial calibration for the path.
/// @returns The initial credit for the new sources. virtual calibration init() = 0;
virtual assignment compute_initial() = 0;
/// Assigs new credit to the source after a cycle ends.
/// @param cycle Duration of a cycle.
virtual assignment compute(timespan cycle) = 0;
// -- virtual functions ------------------------------------------------------
/// Computes a credit assignment to the source after crossing the /// Computes a credit assignment to the source after crossing the
/// low-threshold. May assign zero credit. /// low-threshold. May assign zero credit.
virtual assignment compute_bridge(); virtual calibration calibrate() = 0;
/// Returns the threshold for when we may give extra credit to a source
/// during a cycle.
/// @returns Zero or a negative value if the controller never grants bridge
/// credit, otherwise the threshold for calling `compute_bridge` to
/// generate additional credit.
virtual int32_t threshold() const noexcept;
private:
// -- member variables -------------------------------------------------------
/// Points to the parent system.
scheduled_actor* self_;
}; };
} // namespace caf } // namespace caf
...@@ -33,20 +33,60 @@ ...@@ -33,20 +33,60 @@
namespace caf::defaults::stream { namespace caf::defaults::stream {
constexpr auto desired_batch_complexity = timespan{50'000}; constexpr auto max_batch_delay = timespan{1'000'000};
constexpr auto max_batch_delay = timespan{5'000'000};
constexpr auto credit_round_interval = timespan{10'000'000}; /// Configures an algorithm for assigning credit and adjusting batch sizes.
constexpr auto credit_policy = string_view{"complexity"}; ///
/// The `size-based` controller (default) samples how many Bytes stream elements
/// occupy when serialized to CAF's binary wire format.
///
/// The `token-based` controller associates each stream element with one token.
/// Input buffer and batch sizes are then statically defined in terms of tokens.
/// This strategy makes no dynamic adjustment or sampling.
constexpr auto credit_policy = string_view{"size-based"};
[[deprecated("this parameter no longer has any effect")]] //
constexpr auto credit_round_interval
= max_batch_delay;
} // namespace caf::defaults::stream } // namespace caf::defaults::stream
namespace caf::defaults::stream::size_policy { namespace caf::defaults::stream::size_policy {
constexpr auto bytes_per_batch = int32_t{02 * 1024}; // 2 KB /// Desired size of a single batch in Bytes, when serialized into CAF's binary
/// wire format.
constexpr auto bytes_per_batch = int32_t{2 * 1024}; // 2 KB
/// Number of Bytes (over all received elements) an inbound path may buffer.
/// Actors use heuristics for calculating the estimated memory use, so actors
/// may still allocate more memory in practice.
constexpr auto buffer_capacity = int32_t{64 * 1024}; // 64 KB constexpr auto buffer_capacity = int32_t{64 * 1024}; // 64 KB
/// Frequency of computing the serialized size of incoming batches. Smaller
/// values may increase accuracy, but also add computational overhead.
constexpr auto sampling_rate = int32_t{25};
/// 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
/// batches.
constexpr auto calibration_interval = int32_t{20};
/// Value between 0 and 1 representing the degree of weighting decrease for
/// adjusting batch sizes. A higher factor discounts older observations faster.
constexpr auto smoothing_factor = float{0.6};
} // namespace caf::defaults::stream::size_policy } // namespace caf::defaults::stream::size_policy
namespace caf::defaults::stream::token_policy {
/// Number of elements in a single batch.
constexpr auto batch_size = int32_t{256}; // 2 KB for elements of size 8.
/// Maximum number of elements in the input buffer.
constexpr auto buffer_size = int32_t{4096}; // // 32 KB for elements of size 8.
} // namespace caf::defaults::stream::token_policy
namespace caf::defaults::scheduler { namespace caf::defaults::scheduler {
constexpr auto policy = string_view{"stealing"}; constexpr auto policy = string_view{"stealing"};
......
...@@ -26,17 +26,10 @@ namespace caf::detail { ...@@ -26,17 +26,10 @@ namespace caf::detail {
/// 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 size_based_credit_controller : public credit_controller {
public: public:
// -- member types -----------------------------------------------------------
using super = credit_controller;
// -- constants -------------------------------------------------------------- // -- constants --------------------------------------------------------------
/// Configures at what buffer level we grant bridge credit (0 to 1).
static constexpr float buffer_threshold = 0.75f;
/// Configures how many samples we require for recalculating buffer sizes. /// Configures how many samples we require for recalculating buffer sizes.
static constexpr int32_t min_samples = 10; static constexpr int32_t min_samples = 50;
/// Stores how many elements we buffer at most after the handshake. /// Stores how many elements we buffer at most after the handshake.
int32_t initial_buffer_size = 10; int32_t initial_buffer_size = 10;
...@@ -46,54 +39,50 @@ public: ...@@ -46,54 +39,50 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit size_based_credit_controller(scheduled_actor* self); explicit size_based_credit_controller(local_actor* self);
~size_based_credit_controller() override; ~size_based_credit_controller() override;
// -- overrides -------------------------------------------------------------- // -- interface functions ----------------------------------------------------
void before_processing(downstream_msg::batch& x) override; void before_processing(downstream_msg::batch& batch) override;
void after_processing(downstream_msg::batch& x) override; calibration init() override;
assignment compute_initial() override; calibration calibrate() override;
assignment compute(timespan cycle) override; private:
// -- member variables -------------------------------------------------------
assignment compute_bridge() override; local_actor* self_;
int32_t threshold() const noexcept override; /// Keeps track of when to sample a batch.
int32_t sample_counter_ = 0;
private: /// Stores the last computed (moving) average for the serialized size per
// -- member variables ------------------------------------------------------- /// element in the stream.
int32_t bytes_per_element_ = 0;
/// Total number of elements in all processed batches in the current cycle. /// Stores how many elements were sampled since last calling `calibrate`.
int64_t num_batches_ = 0; int32_t sampled_elements_ = 0;
/// Stores how many elements the buffer should hold at most. /// Stores how many bytes the sampled batches required when serialized.
int32_t buffer_size_ = initial_buffer_size; int64_t sampled_total_size_ = 0;
/// Stores how many elements each batch should contain. /// Stores whether this is the first run.
int32_t batch_size_ = initial_batch_size; bool initializing_ = true;
/// Configures how many bytes we store in total. // -- see caf::defaults::stream::size_policy --------------------------------
int32_t buffer_capacity_;
/// Configures how many bytes we transfer per batch.
int32_t bytes_per_batch_; int32_t bytes_per_batch_;
/// Stores how many elements we have sampled during the current cycle. int32_t buffer_capacity_;
int32_t sampled_elements_ = 0;
/// Stores approximately how many bytes the sampled elements require when int32_t sampling_rate_ = 1;
/// serialized.
int32_t sampled_total_size_ = 0;
/// Counter for keeping track of when to sample a batch. int32_t calibration_interval_;
int32_t sample_counter_ = 0;
/// Configured how many batches we skip for the size sampling. float smoothing_factor_;
int32_t sample_rate_ = 1;
}; };
} // namespace caf::detail } // namespace caf::detail
...@@ -20,36 +20,43 @@ ...@@ -20,36 +20,43 @@
#include "caf/credit_controller.hpp" #include "caf/credit_controller.hpp"
namespace caf { namespace caf::detail {
namespace detail {
/// Computes predictable credit in unit tests. /// A credit controller that estimates the bytes required to store incoming
class test_credit_controller : public credit_controller { /// batches and constrains credit based on upper bounds for memory usage.
class token_based_credit_controller : public credit_controller {
public: public:
// -- member types ----------------------------------------------------------- // -- constants --------------------------------------------------------------
using super = credit_controller; /// Configures how many samples we require for recalculating buffer sizes.
static constexpr int32_t min_samples = 50;
// -- constructors, destructors, and assignment operators -------------------- /// Stores how many elements we buffer at most after the handshake.
int32_t initial_buffer_size = 10;
/// Stores how many elements we allow per batch after the handshake.
int32_t initial_batch_size = 2;
using super::super; // -- constructors, destructors, and assignment operators --------------------
~test_credit_controller() override; explicit token_based_credit_controller(local_actor* self);
// -- overrides -------------------------------------------------------------- ~token_based_credit_controller() override;
void before_processing(downstream_msg::batch& x) override; // -- interface functions ----------------------------------------------------
void after_processing(downstream_msg::batch& x) override; void before_processing(downstream_msg::batch& batch) override;
assignment compute_initial() override; calibration init() override;
assignment compute(timespan cycle) override; calibration calibrate() override;
private: private:
/// Total number of elements in all processed batches in the current cycle. // -- see caf::defaults::stream::token_policy -------------------------------
int64_t num_elements_ = 0;
int32_t batch_size_;
int32_t buffer_size_;
}; };
} // namespace detail } // namespace caf::detail
} // namespace caf
...@@ -21,9 +21,11 @@ ...@@ -21,9 +21,11 @@
#include <memory> #include <memory>
#include <vector> #include <vector>
#include "caf/actor_clock.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/timespan.hpp"
namespace caf { namespace caf {
...@@ -48,6 +50,9 @@ public: ...@@ -48,6 +50,9 @@ public:
/// Unique pointer to an outbound path. /// Unique pointer to an outbound path.
using unique_path_ptr = std::unique_ptr<path_type>; using unique_path_ptr = std::unique_ptr<path_type>;
/// Discrete point in time, as reported by the actor clock.
using time_point = typename actor_clock::time_point;
/// Function object for iterating over all paths. /// Function object for iterating over all paths.
struct CAF_CORE_EXPORT path_visitor { struct CAF_CORE_EXPORT path_visitor {
virtual ~path_visitor(); virtual ~path_visitor();
...@@ -83,6 +88,11 @@ public: ...@@ -83,6 +88,11 @@ public:
/// stream and never has outbound paths. /// stream and never has outbound paths.
virtual bool terminal() const noexcept; virtual bool terminal() const noexcept;
// -- time management --------------------------------------------------------
/// Forces underful batches after reaching the maximum delay.
void tick(time_point now, timespan max_batch_delay);
// -- path management -------------------------------------------------------- // -- path management --------------------------------------------------------
/// Applies `f` to each path. /// Applies `f` to each path.
...@@ -241,6 +251,9 @@ protected: ...@@ -241,6 +251,9 @@ protected:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
stream_manager* parent_; stream_manager* parent_;
/// Stores the time stamp of our last batch.
time_point last_send_;
}; };
} // namespace caf } // namespace caf
...@@ -41,12 +41,36 @@ namespace caf { ...@@ -41,12 +41,36 @@ namespace caf {
/// State for a path to an upstream actor (source). /// State for a path to an upstream actor (source).
class CAF_CORE_EXPORT inbound_path { class CAF_CORE_EXPORT inbound_path {
public: public:
// -- member types -----------------------------------------------------------
/// Message type for propagating graceful shutdowns. /// Message type for propagating graceful shutdowns.
using regular_shutdown = upstream_msg::drop; using regular_shutdown = upstream_msg::drop;
/// Message type for propagating errors. /// Message type for propagating errors.
using irregular_shutdown = upstream_msg::forced_drop; using irregular_shutdown = upstream_msg::forced_drop;
/// Wraps optional actor metrics collected by this path.
struct metrics_t {
telemetry::int_counter* processed_elements;
telemetry::int_gauge* input_buffer_size;
};
/// Discrete point in time, as reported by the actor clock.
using time_point = typename actor_clock::time_point;
/// Time interval, as reported by the actor clock.
using duration_type = typename actor_clock::duration_type;
// -- constructors, destructors, and assignment operators --------------------
/// Constructs a path for given handle and stream ID.
inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
strong_actor_ptr ptr, type_id_t input_type);
~inbound_path();
// -- member variables -------------------------------------------------------
/// Points to the manager responsible for incoming traffic. /// Points to the manager responsible for incoming traffic.
stream_manager_ptr mgr; stream_manager_ptr mgr;
...@@ -56,18 +80,24 @@ public: ...@@ -56,18 +80,24 @@ public:
/// Stores slot IDs for sender (hdl) and receiver (self). /// Stores slot IDs for sender (hdl) and receiver (self).
stream_slots slots; stream_slots slots;
/// Optionally stores pointers to telemetry objects. /// Stores pointers to optional telemetry objects.
struct metrics_t { metrics_t metrics;
telemetry::int_counter* processed_elements;
telemetry::int_gauge* input_buffer_size;
} metrics;
/// Stores the last computed desired batch size. /// Stores the last computed desired batch size. Adjusted at run-time by the
/// controller.
int32_t desired_batch_size = 0; int32_t desired_batch_size = 0;
/// Amount of credit we have signaled upstream. /// Amount of credit we have signaled upstream.
int32_t assigned_credit = 0; int32_t assigned_credit = 0;
/// Maximum amount of credit that the path may signal upstream. Adjusted at
/// run-time by the controller.
int32_t max_credit = 0;
/// Decremented whenever receiving a batch. Triggers a re-calibration by the
/// controller when reaching zero.
int32_t calibration_countdown = 10;
/// Priority of incoming batches from this source. /// Priority of incoming batches from this source.
stream_priority prio = stream_priority::normal; stream_priority prio = stream_priority::normal;
...@@ -80,17 +110,22 @@ public: ...@@ -80,17 +110,22 @@ public:
/// Controller for assigning credit to the source. /// Controller for assigning credit to the source.
std::unique_ptr<credit_controller> controller_; std::unique_ptr<credit_controller> controller_;
/// Stores the time point of the last credit decision for this source. /// Stores when the last ACK was emitted.
actor_clock::time_point last_credit_decision; time_point last_ack_time;
/// Stores the time point of the last credit decision for this source. // -- properties -------------------------------------------------------------
actor_clock::time_point next_credit_decision;
/// Constructs a path for given handle and stream ID. /// Returns whether the path received no input since last emitting
inbound_path(stream_manager_ptr mgr_ptr, stream_slots id, /// `ack_batch`, i.e., `last_acked_batch_id == last_batch_id`.
strong_actor_ptr ptr, type_id_t input_type); bool up_to_date() const noexcept;
~inbound_path(); /// Returns a pointer to the parent actor.
scheduled_actor* self() const noexcept;
/// Returns currently unassigned credit that we could assign to the source.
int32_t available_credit() const noexcept;
// -- 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.
...@@ -103,23 +138,19 @@ public: ...@@ -103,23 +138,19 @@ public:
mgr->handle(this, x); mgr->handle(this, x);
} }
/// Forces an ACK message after receiving no input for a considerable amount
/// of time.
void tick(time_point now, duration_type max_batch_delay);
// -- messaging --------------------------------------------------------------
/// Emits an `upstream_msg::ack_batch`. /// Emits an `upstream_msg::ack_batch`.
void emit_ack_open(local_actor* self, actor_addr rebind_from); void emit_ack_open(local_actor* self, actor_addr rebind_from);
/// Sends an `upstream_msg::ack_batch` for granting new credit. Credit is /// Sends an `upstream_msg::ack_batch` for granting new credit.
/// calculated from sampled batch durations, the cycle duration and the
/// desired batch complexity.
/// @param self Points to the parent actor, i.e., sender of the message. /// @param self Points to the parent actor, i.e., sender of the message.
/// @param queued_items Accumulated size of all batches that are currently /// @param new_credit Amount of new credit to assign to the source.
/// waiting in the mailbox. void emit_ack_batch(local_actor* self, int32_t new_credit);
/// @param now Current timestamp.
/// @param cycle Time between credit rounds.
void emit_ack_batch(local_actor* self, int32_t queued_items,
actor_clock::time_point now, timespan cycle);
/// Returns whether the path received no input since last emitting
/// `ack_batch`, i.e., `last_acked_batch_id == last_batch_id`.
bool up_to_date();
/// Sends an `upstream_msg::drop` on this path. /// Sends an `upstream_msg::drop` on this path.
void emit_regular_shutdown(local_actor* self); void emit_regular_shutdown(local_actor* self);
...@@ -131,9 +162,6 @@ public: ...@@ -131,9 +162,6 @@ 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);
/// Returns a pointer to the parent actor.
scheduled_actor* self();
}; };
/// @relates inbound_path /// @relates inbound_path
......
...@@ -32,7 +32,6 @@ ...@@ -32,7 +32,6 @@
#include "caf/actor_traits.hpp" #include "caf/actor_traits.hpp"
#include "caf/detail/behavior_stack.hpp" #include "caf/detail/behavior_stack.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/tick_emitter.hpp"
#include "caf/detail/unordered_flat_map.hpp" #include "caf/detail/unordered_flat_map.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
...@@ -681,6 +680,10 @@ public: ...@@ -681,6 +680,10 @@ public:
|| !pending_stream_managers_.empty(); || !pending_stream_managers_.empty();
} }
auto max_batch_delay() const noexcept {
return max_batch_delay_;
}
/// @endcond /// @endcond
protected: protected:
...@@ -723,15 +726,13 @@ protected: ...@@ -723,15 +726,13 @@ protected:
/// yet received an ACK. /// yet received an ACK.
stream_manager_map pending_stream_managers_; stream_manager_map pending_stream_managers_;
/// Controls batch and credit timeouts. /// Stores how long the actor should try to accumulate more items in order to
detail::tick_emitter stream_ticks_; /// send a full stream batch.
timespan max_batch_delay_;
/// Number of ticks per batch delay. /// Number of ticks per batch delay.
size_t max_batch_delay_ticks_; size_t max_batch_delay_ticks_;
/// Number of ticks of each credit round.
size_t credit_round_ticks_;
/// Pointer to a private thread object associated with a detached actor. /// Pointer to a private thread object associated with a detached actor.
detail::private_thread* private_thread_; detail::private_thread* private_thread_;
......
...@@ -25,12 +25,14 @@ ...@@ -25,12 +25,14 @@
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/actor_clock.hpp"
#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/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"
#include "caf/timespan.hpp"
#include "caf/upstream_msg.hpp" #include "caf/upstream_msg.hpp"
namespace caf { namespace caf {
...@@ -56,6 +58,11 @@ public: ...@@ -56,6 +58,11 @@ public:
using inbound_paths_list = std::vector<inbound_path*>; using inbound_paths_list = std::vector<inbound_path*>;
/// Discrete point in time.
using time_point = typename actor_clock::time_point;
// -- constructors, destructors, and assignment operators --------------------
stream_manager(scheduled_actor* selfptr, stream_manager(scheduled_actor* selfptr,
stream_priority prio = stream_priority::normal); stream_priority prio = stream_priority::normal);
...@@ -83,10 +90,6 @@ public: ...@@ -83,10 +90,6 @@ public:
/// buffers of in- and outbound paths. /// buffers of in- and outbound paths.
virtual void shutdown(); virtual void shutdown();
/// Tries to advance the stream by generating more credit or by sending
/// batches.
void advance();
/// Pushes new data to downstream actors by sending batches. The amount of /// Pushes new data to downstream actors by sending batches. The amount of
/// pushed data is limited by the available credit. /// pushed data is limited by the available credit.
virtual void push(); virtual void push();
...@@ -107,7 +110,7 @@ public: ...@@ -107,7 +110,7 @@ public:
/// messages. /// messages.
virtual bool generate_messages(); virtual bool generate_messages();
// -- pure virtual member functions ------------------------------------------ // -- interface functions ----------------------------------------------------
/// Returns the manager for downstream communication. /// Returns the manager for downstream communication.
virtual downstream_manager& out() = 0; virtual downstream_manager& out() = 0;
...@@ -127,6 +130,11 @@ public: ...@@ -127,6 +130,11 @@ public:
/// Advances time. /// Advances time.
virtual void cycle_timeout(size_t cycle_nr); virtual void cycle_timeout(size_t cycle_nr);
/// Acquires credit on an inbound path. The calculated credit to fill our
/// queue for two cycles is `desired`, but the manager is allowed to return
/// any non-negative value.
virtual int32_t acquire_credit(inbound_path* path, int32_t desired);
// -- input path management -------------------------------------------------- // -- input path management --------------------------------------------------
/// Informs the manager that a new input path opens. /// Informs the manager that a new input path opens.
...@@ -184,10 +192,7 @@ public: ...@@ -184,10 +192,7 @@ public:
return self_; return self_;
} }
/// Acquires credit on an inbound path. The calculated credit to fill our // -- output path management -------------------------------------------------
/// queue for two cycles is `desired`, but the manager is allowed to return
/// any non-negative value.
virtual int32_t acquire_credit(inbound_path* path, int32_t desired);
/// Creates an outbound path to the current sender without any type checking. /// Creates an outbound path to the current sender without any type checking.
/// @pre `out().terminal() == false` /// @pre `out().terminal() == false`
...@@ -268,6 +273,10 @@ public: ...@@ -268,6 +273,10 @@ public:
/// @pre Current message is an `open_stream_msg`. /// @pre Current message is an `open_stream_msg`.
stream_slot add_unchecked_inbound_path_impl(type_id_t rtti); stream_slot add_unchecked_inbound_path_impl(type_id_t rtti);
// -- time management --------------------------------------------------------
void tick(time_point now);
protected: protected:
// -- modifiers for self ----------------------------------------------------- // -- modifiers for self -----------------------------------------------------
...@@ -312,6 +321,10 @@ protected: ...@@ -312,6 +321,10 @@ protected:
/// Stores individual flags, for continuous streaming or when shutting down. /// Stores individual flags, for continuous streaming or when shutting down.
int flags_; int flags_;
/// Stores the maximum amount of time outbound paths should buffer elements
/// before sending underful batches.
timespan max_batch_delay_;
private: private:
void setf(int flag) noexcept { void setf(int flag) noexcept {
auto x = flags_; auto x = flags_;
......
...@@ -62,7 +62,7 @@ public: ...@@ -62,7 +62,7 @@ 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(const stream<input_type>&) { inbound_stream_slot<input_type> add_inbound_path(stream<input_type>) {
return {this->add_unchecked_inbound_path_impl(type_id_v<input_type>)}; return {this->add_unchecked_inbound_path_impl(type_id_v<input_type>)};
} }
......
...@@ -59,7 +59,7 @@ public: ...@@ -59,7 +59,7 @@ public:
} }
DownstreamManager& out() override { DownstreamManager& out() override {
return left_super::out(); return this->out_;
} }
}; };
......
...@@ -57,9 +57,8 @@ actor_system_config::actor_system_config() ...@@ -57,9 +57,8 @@ actor_system_config::actor_system_config()
config_file_path(default_config_file), config_file_path(default_config_file),
slave_mode_fun(nullptr) { slave_mode_fun(nullptr) {
// (1) hard-coded defaults // (1) hard-coded defaults
stream_desired_batch_complexity = defaults::stream::desired_batch_complexity;
stream_max_batch_delay = defaults::stream::max_batch_delay; stream_max_batch_delay = defaults::stream::max_batch_delay;
stream_credit_round_interval = defaults::stream::credit_round_interval; stream_credit_round_interval = 2 * stream_max_batch_delay;
// fill our options vector for creating config file and CLI parsers // fill our options vector for creating config file and CLI parsers
using std::string; using std::string;
using string_list = std::vector<string>; using string_list = std::vector<string>;
...@@ -70,14 +69,19 @@ actor_system_config::actor_system_config() ...@@ -70,14 +69,19 @@ actor_system_config::actor_system_config()
.add<string>(config_file_path, "config-file", .add<string>(config_file_path, "config-file",
"set config file (default: caf-application.conf)"); "set config file (default: caf-application.conf)");
opt_group{custom_options_, "caf.stream"} opt_group{custom_options_, "caf.stream"}
.add<timespan>(stream_desired_batch_complexity, "desired-batch-complexity",
"processing time per batch")
.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",
"time between emitting credit")
.add<string>("credit-policy", .add<string>("credit-policy",
"selects an algorithm for credit computation"); "selects an implementation for credit computation");
opt_group{custom_options_, "caf.stream.size-based-policy"}
.add<int32_t>("bytes-per-batch", "desired batch size in bytes")
.add<int32_t>("buffer-capacity", "maximum input buffer size in bytes")
.add<int32_t>("sampling-rate", "frequency of collecting batch sizes")
.add<int32_t>("calibration-interval", "frequency of re-calibrations")
.add<float>("smoothing-factor", "factor for discounting older samples");
opt_group{custom_options_, "caf.stream.token-based-policy"}
.add<int32_t>("batch-size", "number of elements per batch")
.add<int32_t>("buffer-size", "max. number of elements in the input buffer");
opt_group{custom_options_, "caf.scheduler"} opt_group{custom_options_, "caf.scheduler"}
.add<string>("policy", "'stealing' (default) or 'sharing'") .add<string>("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")
...@@ -117,12 +121,8 @@ settings actor_system_config::dump_content() const { ...@@ -117,12 +121,8 @@ settings actor_system_config::dump_content() const {
auto& caf_group = result["caf"].as_dictionary(); auto& caf_group = result["caf"].as_dictionary();
// -- streaming parameters // -- streaming parameters
auto& stream_group = caf_group["stream"].as_dictionary(); auto& stream_group = caf_group["stream"].as_dictionary();
put_missing(stream_group, "desired-batch-complexity",
defaults::stream::desired_batch_complexity);
put_missing(stream_group, "max-batch-delay", put_missing(stream_group, "max-batch-delay",
defaults::stream::max_batch_delay); defaults::stream::max_batch_delay);
put_missing(stream_group, "credit-round-interval",
defaults::stream::credit_round_interval);
put_missing(stream_group, "credit-policy", defaults::stream::credit_policy); put_missing(stream_group, "credit-policy", defaults::stream::credit_policy);
put_missing(stream_group, "size-policy.buffer-capacity", put_missing(stream_group, "size-policy.buffer-capacity",
defaults::stream::size_policy::buffer_capacity); defaults::stream::size_policy::buffer_capacity);
...@@ -354,12 +354,6 @@ actor_system_config& actor_system_config::set_impl(string_view name, ...@@ -354,12 +354,6 @@ actor_system_config& actor_system_config::set_impl(string_view name,
return *this; return *this;
} }
timespan actor_system_config::stream_tick_duration() const noexcept {
auto ns_count = caf::detail::gcd(stream_credit_round_interval.count(),
stream_max_batch_delay.count());
return timespan{ns_count};
}
std::string actor_system_config::render(const error& x) { std::string actor_system_config::render(const error& x) {
return to_string(x); return to_string(x);
} }
......
...@@ -20,20 +20,8 @@ ...@@ -20,20 +20,8 @@
namespace caf { namespace caf {
credit_controller::credit_controller(scheduled_actor* self) : self_(self) {
// nop
}
credit_controller::~credit_controller() { credit_controller::~credit_controller() {
// nop // nop
} }
credit_controller::assignment credit_controller::compute_bridge() {
return {0, 0};
}
int32_t credit_controller::threshold() const noexcept {
return -1;
}
} // namespace caf } // namespace caf
...@@ -24,77 +24,91 @@ ...@@ -24,77 +24,91 @@
#include "caf/detail/serialized_size.hpp" #include "caf/detail/serialized_size.hpp"
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
// Safe us some typing and very ugly formatting. namespace {
#define impl size_based_credit_controller
namespace caf::detail { // Sample the first 10 batches when starting up.
constexpr int32_t initial_sample_size = 10;
impl::impl(scheduled_actor* self) : super(self) { } // namespace
auto& cfg = self->system().config();
buffer_capacity_ = get_or(cfg, "caf.stream.size-policy.buffer-capacity",
defaults::stream::size_policy::buffer_capacity);
bytes_per_batch_ = get_or(cfg, "caf.stream.size-policy.bytes-per-batch",
defaults::stream::size_policy::bytes_per_batch);
}
impl::~impl() { namespace caf::detail {
// nop
}
void impl::before_processing(downstream_msg::batch& x) { size_based_credit_controller::size_based_credit_controller(local_actor* ptr)
if (++sample_counter_ == sample_rate_) { : self_(ptr) {
sampled_elements_ += x.xs_size; namespace fallback = defaults::stream::size_policy;
sampled_total_size_ += serialized_size(self()->system(), x.xs); // Initialize from the config parameters.
sample_counter_ = 0; auto& cfg = ptr->system().config();
if (auto section = get_if<settings>(&cfg, "caf.stream.size-based-policy")) {
bytes_per_batch_ = get_or(*section, "bytes-per-batch",
fallback::bytes_per_batch);
buffer_capacity_ = get_or(*section, "buffer-capacity",
fallback::buffer_capacity);
calibration_interval_ = get_or(*section, "calibration-interval",
fallback::calibration_interval);
smoothing_factor_ = get_or(*section, "smoothing-factor",
fallback::smoothing_factor);
} else {
buffer_capacity_ = fallback::buffer_capacity;
bytes_per_batch_ = fallback::bytes_per_batch;
calibration_interval_ = fallback::calibration_interval;
smoothing_factor_ = fallback::smoothing_factor;
} }
++num_batches_;
} }
void impl::after_processing(downstream_msg::batch&) { size_based_credit_controller::~size_based_credit_controller() {
// nop // nop
} }
credit_controller::assignment impl::compute_initial() { void size_based_credit_controller::before_processing(downstream_msg::batch& x) {
return {buffer_size_, batch_size_}; if (++sample_counter_ == sampling_rate_) {
} sample_counter_ = 0;
sampled_elements_ += x.xs_size;
credit_controller::assignment impl::compute(timespan) { // FIXME: this wildly overestimates the actual size per element, because we
if (sampled_elements_ >= min_samples) { // include the size of the meta data per message. This also punishes
// Helper for truncating a 64-bit integer to a 32-bit integer with a // small batches and we probably never reach a stable point even if
// minimum value of 1. // incoming data always has the exact same size per element.
auto clamp_i32 = [](int64_t x) -> int32_t { sampled_total_size_ += static_cast<int64_t>(serialized_size(x.xs));
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);
};
// Calculate ideal batch size by size.
auto bytes_per_element = clamp_i32(sampled_total_size_ / sampled_elements_);
batch_size_ = clamp_i32(bytes_per_batch_ / bytes_per_element);
buffer_size_ = clamp_i32(buffer_capacity_ / bytes_per_element);
// Reset bookkeeping state.
sampled_elements_ = 0;
sampled_total_size_ = 0;
// Adjust the sample rate to reach min_samples in the next cycle.
sample_rate_ = clamp_i32(num_batches_ / min_samples);
if (sample_counter_ >= sample_rate_)
sample_counter_ = 0;
num_batches_ = 0;
} }
return {buffer_size_, batch_size_};
} }
credit_controller::assignment impl::compute_bridge() { credit_controller::calibration size_based_credit_controller::init() {
CAF_ASSERT(batch_size_ > 0); // Initially, we simply assume that the size of one element equals
CAF_ASSERT(buffer_size_ > batch_size_); // bytes-per-batch.
return {buffer_size_, batch_size_}; return {buffer_capacity_ / bytes_per_batch_, 1, initial_sample_size};
} }
int32_t impl::threshold() const noexcept { credit_controller::calibration size_based_credit_controller::calibrate() {
return buffer_size_-1; CAF_ASSERT(sample_counter_ == 0);
// return static_cast<int32_t>(buffer_size_ * buffer_threshold); // Helper for truncating a 64-bit integer to a 32-bit integer with a minimum
// value of 1.
auto clamp_i32 = [](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);
};
if (!initializing_) {
auto bpe = clamp_i32(sampled_total_size_ / calibration_interval_);
bytes_per_element_ = static_cast<int32_t>(
smoothing_factor_ * bpe // weighted current measurement
+ (1.0 - smoothing_factor_) * bytes_per_element_ // past values
);
} else {
// After our first run, we continue with the actual sampling rate.
initializing_ = false;
sampling_rate_ = get_or(self_->config(),
"caf.stream.size-policy.sample-rate",
defaults::stream::size_policy::sampling_rate);
bytes_per_element_ = clamp_i32(sampled_total_size_ / initial_sample_size);
}
sampled_total_size_ = 0;
return {
clamp_i32(buffer_capacity_ / bytes_per_element_),
clamp_i32(bytes_per_batch_ / bytes_per_element_),
sampling_rate_ * calibration_interval_,
};
} }
} // namespace caf::detail } // namespace caf::detail
...@@ -16,60 +16,45 @@ ...@@ -16,60 +16,45 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/detail/test_credit_controller.hpp" #include "caf/detail/token_based_credit_controller.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/scheduled_actor.hpp" #include "caf/config_value.hpp"
#include "caf/defaults.hpp"
namespace caf { #include "caf/detail/serialized_size.hpp"
namespace detail { #include "caf/local_actor.hpp"
#include "caf/settings.hpp"
test_credit_controller::~test_credit_controller() {
// nop namespace caf::detail {
token_based_credit_controller::token_based_credit_controller(local_actor* ptr) {
namespace fallback = defaults::stream::token_policy;
// Initialize from the config parameters.
auto& cfg = ptr->system().config();
if (auto section = get_if<settings>(&cfg, "caf.stream.token-based-policy")) {
batch_size_ = get_or(*section, "batch-size", fallback::batch_size);
buffer_size_ = get_or(*section, "buffer-size", fallback::buffer_size);
} else {
batch_size_ = fallback::batch_size;
buffer_size_ = fallback::buffer_size;
}
} }
void test_credit_controller::before_processing(downstream_msg::batch& x) { token_based_credit_controller::~token_based_credit_controller() {
num_elements_ += x.xs_size; // nop
} }
void test_credit_controller::after_processing(downstream_msg::batch&) { void token_based_credit_controller::before_processing(downstream_msg::batch&) {
// nop // nop
} }
credit_controller::assignment test_credit_controller::compute_initial() { credit_controller::calibration token_based_credit_controller::init() {
return {50, 50}; return calibrate();
} }
credit_controller::assignment test_credit_controller::compute(timespan cycle) { credit_controller::calibration token_based_credit_controller::calibrate() {
auto& cfg = self()->system().config(); return {buffer_size_, batch_size_, std::numeric_limits<int32_t>::max()};
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::detail
} // namespace caf
...@@ -41,7 +41,7 @@ downstream_manager::downstream_manager::path_predicate::~path_predicate() { ...@@ -41,7 +41,7 @@ downstream_manager::downstream_manager::path_predicate::~path_predicate() {
downstream_manager::downstream_manager(stream_manager* parent) downstream_manager::downstream_manager(stream_manager* parent)
: parent_(parent) { : parent_(parent) {
// nop last_send_ = parent->self()->now();
} }
downstream_manager::~downstream_manager() { downstream_manager::~downstream_manager() {
...@@ -62,6 +62,13 @@ bool downstream_manager::terminal() const noexcept { ...@@ -62,6 +62,13 @@ bool downstream_manager::terminal() const noexcept {
return true; return true;
} }
// -- time management ----------------------------------------------------------
void downstream_manager::tick(time_point now, timespan max_batch_delay) {
if (now >= last_send_ + max_batch_delay && buffered() > 0)
force_emit_batches();
}
// -- path management ---------------------------------------------------------- // -- path management ----------------------------------------------------------
std::vector<stream_slot> downstream_manager::path_slots() { std::vector<stream_slot> downstream_manager::path_slots() {
......
...@@ -22,51 +22,27 @@ ...@@ -22,51 +22,27 @@
#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/size_based_credit_controller.hpp"
#include "caf/detail/test_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"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/settings.hpp" #include "caf/settings.hpp"
namespace caf {
namespace { namespace {
constexpr bool force_ack = true; template <class T>
void set_controller(std::unique_ptr<caf::credit_controller>& ptr,
void emit_ack_batch(inbound_path& path, credit_controller::assignment x, caf::local_actor* self) {
bool force_ack_msg = false) { ptr = std::make_unique<T>(self);
CAF_ASSERT(x.batch_size > 0);
auto& out = path.mgr->out();
path.desired_batch_size = x.batch_size;
int32_t new_credit = 0;
auto used = static_cast<int32_t>(out.buffered()) + path.assigned_credit;
auto guard = detail::make_scope_guard([&] {
if (!force_ack_msg || path.up_to_date())
return;
unsafe_send_as(
path.self(), path.hdl,
make<upstream_msg::ack_batch>(path.slots.invert(), path.self()->address(),
0, x.batch_size, path.last_batch_id));
path.last_acked_batch_id = path.last_batch_id;
});
if (x.credit <= used)
return;
new_credit = path.mgr->acquire_credit(&path, x.credit - used);
if (new_credit < 1)
return;
guard.disable();
unsafe_send_as(path.self(), path.hdl,
make<upstream_msg::ack_batch>(
path.slots.invert(), path.self()->address(), new_credit,
x.batch_size, path.last_batch_id));
path.last_acked_batch_id = path.last_batch_id;
path.assigned_credit += new_credit;
} }
} // namespace } // namespace
namespace caf {
// -- constructors, destructors, and assignment operators ----------------------
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, strong_actor_ptr ptr,
[[maybe_unused]] type_id_t in_type) [[maybe_unused]] type_id_t in_type)
...@@ -80,55 +56,87 @@ inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id, ...@@ -80,55 +56,87 @@ inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
<< "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); << "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(), if (auto str = get_if<std::string>(&self->system().config(),
"caf.stream.credit-policy")) { "caf.stream.credit-policy")) {
if (*str == "testing") if (*str == "token-based")
controller_.reset(new detail::test_credit_controller(self)); setter = set_controller<detail::token_based_credit_controller>;
else if (*str == "size") else if (*str != "size-based")
controller_.reset(new detail::size_based_credit_controller(self)); CAF_LOG_WARNING("unrecognized credit policy:"
} else { << *str << "(falling back to 'size-based')");
controller_.reset(new detail::size_based_credit_controller(self));
} }
setter(controller_, self);
last_ack_time = self->now();
} }
inbound_path::~inbound_path() { inbound_path::~inbound_path() {
mgr->deregister_input_path(this); mgr->deregister_input_path(this);
} }
void inbound_path::handle(downstream_msg::batch& x) { // -- properties ---------------------------------------------------------------
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
auto batch_size = x.xs_size; bool inbound_path::up_to_date() const noexcept {
last_batch_id = x.id; return last_acked_batch_id == last_batch_id;
CAF_STREAM_LOG_DEBUG(mgr->self()->name() << "handles batch of size" }
<< batch_size << "on slot" << slots.receiver << "with"
<< assigned_credit << "assigned credit"); scheduled_actor* inbound_path::self() const noexcept {
if (assigned_credit <= batch_size) { return mgr->self();
assigned_credit = 0; }
// Do not log a message when "running out of credit" for the first batch
// that can easily consume the initial credit in one shot. int32_t inbound_path::available_credit() const noexcept {
CAF_STREAM_LOG_DEBUG_IF(next_credit_decision.time_since_epoch().count() > 0, // The max_credit may decrease as a result of re-calibration, in which case
mgr->self()->name() // the source can have more than the maximum amount for a brief period.
<< "ran out of credit at slot" << slots.receiver); return std::max(max_credit - assigned_credit, int32_t{0});
} else { }
assigned_credit -= batch_size;
CAF_ASSERT(assigned_credit >= 0); // -- callbacks ----------------------------------------------------------------
void inbound_path::handle(downstream_msg::batch& batch) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(batch));
// Handle batch.
auto batch_size = batch.xs_size;
last_batch_id = batch.id;
CAF_STREAM_LOG_DEBUG(self()->name() << "handles batch of size" << batch_size
<< "on slot" << slots.receiver << "with"
<< assigned_credit << "assigned credit");
assigned_credit -= batch_size;
CAF_ASSERT(assigned_credit >= 0);
controller_->before_processing(batch);
mgr->handle(this, batch);
// Update settings as necessary.
if (--calibration_countdown == 0) {
auto [cmax, bsize, countdown] = controller_->calibrate();
max_credit = cmax;
desired_batch_size = bsize;
calibration_countdown = countdown;
} }
auto threshold = controller_->threshold(); // Send ACK whenever we can assign credit for another batch to the source.
if (threshold >= 0 && assigned_credit <= threshold) if (auto available = available_credit(); available >= desired_batch_size)
caf::emit_ack_batch(*this, controller_->compute_bridge()); if (auto acquired = mgr->acquire_credit(this, available); acquired > 0)
controller_->before_processing(x); emit_ack_batch(self(), acquired);
mgr->handle(this, x); // FIXME: move this up to the actor
controller_->after_processing(x);
mgr->push(); mgr->push();
} }
void inbound_path::tick(time_point now, duration_type max_batch_delay) {
if (now >= last_ack_time + (2 * max_batch_delay)) {
int32_t new_credit = 0;
if (auto available = available_credit(); available > 0)
new_credit = mgr->acquire_credit(this, available);
emit_ack_batch(self(), new_credit);
}
}
// -- 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) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(rebind_from)); CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(rebind_from));
// Update state. // Update state.
auto initial = controller_->compute_initial(); auto [cmax, bsize, countdown] = controller_->init();
assigned_credit = mgr->acquire_credit(this, initial.credit); max_credit = cmax;
CAF_ASSERT(assigned_credit >= 0); assigned_credit = mgr->acquire_credit(this, cmax);
desired_batch_size = std::min(initial.batch_size, assigned_credit); desired_batch_size = bsize;
calibration_countdown = countdown;
// 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);
...@@ -137,19 +145,21 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) { ...@@ -137,19 +145,21 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
make<upstream_msg::ack_open>( make<upstream_msg::ack_open>(
slots.invert(), self->address(), std::move(rebind_from), slots.invert(), self->address(), std::move(rebind_from),
self->ctrl(), assigned_credit, desired_batch_size)); self->ctrl(), assigned_credit, desired_batch_size));
last_credit_decision = self->clock().now(); last_ack_time = self->now();
} }
void inbound_path::emit_ack_batch(local_actor*, int32_t, void inbound_path::emit_ack_batch(local_actor* self, int32_t new_credit) {
actor_clock::time_point now, timespan cycle) { CAF_LOG_TRACE(CAF_ARG(new_credit));
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(cycle)); CAF_ASSERT(desired_batch_size > 0);
last_credit_decision = now; if (last_acked_batch_id == last_batch_id && new_credit == 0)
next_credit_decision = now + cycle; return;
caf::emit_ack_batch(*this, controller_->compute(cycle), force_ack); unsafe_send_as(self, hdl,
} make<upstream_msg::ack_batch>(slots.invert(), self->address(),
new_credit, desired_batch_size,
bool inbound_path::up_to_date() { last_batch_id));
return last_acked_batch_id == last_batch_id; last_acked_batch_id = last_batch_id;
assigned_credit += new_credit;
last_ack_time = self->now();
} }
void inbound_path::emit_regular_shutdown(local_actor* self) { void inbound_path::emit_regular_shutdown(local_actor* self) {
...@@ -180,8 +190,4 @@ void inbound_path::emit_irregular_shutdown(local_actor* self, ...@@ -180,8 +190,4 @@ void inbound_path::emit_irregular_shutdown(local_actor* self,
std::move(reason))); std::move(reason)));
} }
scheduled_actor* inbound_path::self() {
return mgr->self();
}
} // namespace caf } // namespace caf
...@@ -94,6 +94,10 @@ void local_actor::setup_metrics() { ...@@ -94,6 +94,10 @@ void local_actor::setup_metrics() {
metrics_ = make_instance_metrics(this); metrics_ = make_instance_metrics(this);
} }
auto local_actor::now() const noexcept -> clock_type::time_point {
return clock().now();
}
void local_actor::request_response_timeout(timespan timeout, message_id mid) { void local_actor::request_response_timeout(timespan timeout, message_id mid) {
CAF_LOG_TRACE(CAF_ARG(timeout) << CAF_ARG(mid)); CAF_LOG_TRACE(CAF_ARG(timeout) << CAF_ARG(mid));
if (timeout == infinite) if (timeout == infinite)
......
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#include "caf/actor_ostream.hpp" #include "caf/actor_ostream.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp" #include "caf/detail/default_invoke_result_visitor.hpp"
#include "caf/detail/meta_object.hpp" #include "caf/detail/meta_object.hpp"
#include "caf/detail/private_thread.hpp" #include "caf/detail/private_thread.hpp"
...@@ -136,19 +137,8 @@ scheduled_actor::scheduled_actor(actor_config& cfg) ...@@ -136,19 +137,8 @@ scheduled_actor::scheduled_actor(actor_config& cfg)
#endif // CAF_ENABLE_EXCEPTIONS #endif // CAF_ENABLE_EXCEPTIONS
{ {
auto& sys_cfg = home_system().config(); auto& sys_cfg = home_system().config();
auto interval = sys_cfg.stream_tick_duration(); max_batch_delay_ = get_or(sys_cfg, "caf.stream.max_batch_delay",
CAF_ASSERT(interval.count() > 0); defaults::stream::max_batch_delay);
stream_ticks_.interval(interval);
CAF_ASSERT(sys_cfg.stream_max_batch_delay.count() > 0);
auto div = [](timespan x, timespan y) {
return static_cast<size_t>(x.count() / y.count());
};
max_batch_delay_ticks_ = div(sys_cfg.stream_max_batch_delay, interval);
CAF_ASSERT(max_batch_delay_ticks_ > 0);
credit_round_ticks_ = div(sys_cfg.stream_credit_round_interval, interval);
CAF_ASSERT(credit_round_ticks_ > 0);
CAF_LOG_DEBUG(CAF_ARG(interval) << CAF_ARG(max_batch_delay_ticks_)
<< CAF_ARG(credit_round_ticks_));
} }
scheduled_actor::~scheduled_actor() { scheduled_actor::~scheduled_actor() {
...@@ -913,8 +903,6 @@ bool scheduled_actor::finalize() { ...@@ -913,8 +903,6 @@ bool scheduled_actor::finalize() {
i = stream_managers_.erase(i); i = stream_managers_.erase(i);
else else
++i; ++i;
if (stream_managers_.empty())
stream_ticks_.stop();
} }
} }
// An actor is considered alive as long as it has a behavior and didn't set // An actor is considered alive as long as it has a behavior and didn't set
...@@ -1075,8 +1063,6 @@ stream_slot scheduled_actor::next_slot() { ...@@ -1075,8 +1063,6 @@ stream_slot scheduled_actor::next_slot() {
void scheduled_actor::assign_slot(stream_slot x, stream_manager_ptr mgr) { void scheduled_actor::assign_slot(stream_slot x, stream_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
if (stream_managers_.empty())
stream_ticks_.start(clock().now());
CAF_ASSERT(stream_managers_.count(x) == 0); CAF_ASSERT(stream_managers_.count(x) == 0);
stream_managers_.emplace(x, std::move(mgr)); stream_managers_.emplace(x, std::move(mgr));
} }
...@@ -1106,15 +1092,12 @@ scheduled_actor::assign_next_pending_slot_to(stream_manager_ptr mgr) { ...@@ -1106,15 +1092,12 @@ scheduled_actor::assign_next_pending_slot_to(stream_manager_ptr mgr) {
bool scheduled_actor::add_stream_manager(stream_slot id, bool scheduled_actor::add_stream_manager(stream_slot id,
stream_manager_ptr mgr) { stream_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG(id)); CAF_LOG_TRACE(CAF_ARG(id));
if (stream_managers_.empty())
stream_ticks_.start(clock().now());
return stream_managers_.emplace(id, std::move(mgr)).second; return stream_managers_.emplace(id, std::move(mgr)).second;
} }
void scheduled_actor::erase_stream_manager(stream_slot id) { void scheduled_actor::erase_stream_manager(stream_slot id) {
CAF_LOG_TRACE(CAF_ARG(id)); CAF_LOG_TRACE(CAF_ARG(id));
if (stream_managers_.erase(id) != 0 && stream_managers_.empty()) stream_managers_.erase(id);
stream_ticks_.stop();
CAF_LOG_DEBUG(CAF_ARG2("stream_managers_.size", stream_managers_.size())); CAF_LOG_DEBUG(CAF_ARG2("stream_managers_.size", stream_managers_.size()));
} }
...@@ -1133,8 +1116,6 @@ void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) { ...@@ -1133,8 +1116,6 @@ void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) {
i = stream_managers_.erase(i); i = stream_managers_.erase(i);
else else
++i; ++i;
if (stream_managers_.empty())
stream_ticks_.stop();
} }
{ // Lifetime scope of second iterator pair. { // Lifetime scope of second iterator pair.
auto i = pending_stream_managers_.begin(); auto i = pending_stream_managers_.begin();
...@@ -1193,41 +1174,21 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) { ...@@ -1193,41 +1174,21 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
actor_clock::time_point actor_clock::time_point
scheduled_actor::advance_streams(actor_clock::time_point now) { scheduled_actor::advance_streams(actor_clock::time_point now) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE(CAF_ARG(now));
if (!stream_ticks_.started()) { if (stream_managers_.empty())
CAF_LOG_DEBUG("tick emitter not started yet");
return actor_clock::time_point::max(); return actor_clock::time_point::max();
} std::vector<stream_manager*> managers;
/// Advance time for driving forced batches and credit. managers.reserve(stream_managers_.size());
auto bitmask = stream_ticks_.timeouts(now, {max_batch_delay_ticks_, for (auto& kvp : stream_managers_)
credit_round_ticks_}); managers.emplace_back(kvp.second.get());
// Force batches on all output paths. std::sort(managers.begin(), managers.end());
if ((bitmask & 0x01) != 0 && !stream_managers_.empty()) { auto e = std::unique(managers.begin(), managers.end());
std::vector<stream_manager*> managers; for (auto i = managers.begin(); i != e; ++i)
managers.reserve(stream_managers_.size()); (*i)->tick(now);
for (auto& kvp : stream_managers_) auto idle = [](const stream_manager* mgr) { return mgr->idle(); };
managers.emplace_back(kvp.second.get()); if (std::all_of(managers.begin(), e, idle))
std::sort(managers.begin(), managers.end()); return actor_clock::time_point::max();
auto e = std::unique(managers.begin(), managers.end()); return now + max_batch_delay_;
for (auto i = managers.begin(); i != e; ++i)
(*i)->out().force_emit_batches();
}
// Fill up credit on each input path.
if ((bitmask & 0x02) != 0) {
CAF_LOG_DEBUG("new credit round");
auto cycle = stream_ticks_.interval();
cycle *= static_cast<decltype(cycle)::rep>(credit_round_ticks_);
auto& qs = get_downstream_queue().queues();
for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get();
if (inptr != nullptr) {
auto tts = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(this, tts, now, cycle);
}
}
}
return stream_ticks_.next_timeout(now, {max_batch_delay_ticks_,
credit_round_ticks_});
} }
} // namespace caf } // namespace caf
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/defaults.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/inbound_path.hpp" #include "caf/inbound_path.hpp"
...@@ -37,7 +38,9 @@ namespace caf { ...@@ -37,7 +38,9 @@ namespace caf {
stream_manager::stream_manager(scheduled_actor* selfptr, stream_priority prio) stream_manager::stream_manager(scheduled_actor* selfptr, stream_priority prio)
: self_(selfptr), pending_handshakes_(0), priority_(prio), flags_(0) { : self_(selfptr), pending_handshakes_(0), priority_(prio), flags_(0) {
// nop auto& cfg = selfptr->config();
max_batch_delay_ = get_or(cfg, "caf.stream.max-batch-delay",
defaults::stream::max_batch_delay);
} }
stream_manager::~stream_manager() { stream_manager::~stream_manager() {
...@@ -147,28 +150,6 @@ void stream_manager::shutdown() { ...@@ -147,28 +150,6 @@ void stream_manager::shutdown() {
ipath->emit_regular_shutdown(self_); ipath->emit_regular_shutdown(self_);
} }
void stream_manager::advance() {
CAF_LOG_TRACE("");
// Try to emit more credit.
if (!inbound_paths_.empty()) {
auto now = self_->clock().now();
auto& cfg = self_->system().config();
auto interval = cfg.stream_credit_round_interval;
auto& qs = self_->get_downstream_queue().queues();
// Iterate all queues for inbound traffic.
for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get();
// Ignore inbound paths of other managers.
if (inptr->mgr.get() == this) {
auto tts = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(self_, tts, now, interval);
}
}
}
// Try to generate more batches.
push();
}
void stream_manager::push() { void stream_manager::push() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
do { do {
...@@ -316,6 +297,12 @@ stream_manager::add_unchecked_inbound_path_impl(type_id_t input_type) { ...@@ -316,6 +297,12 @@ stream_manager::add_unchecked_inbound_path_impl(type_id_t input_type) {
return slot; return slot;
} }
void stream_manager::tick(time_point now) {
for (auto path : inbound_paths_)
path->tick(now, max_batch_delay_);
out().tick(now, max_batch_delay_);
}
stream_slot stream_manager::assign_next_slot() { stream_slot stream_manager::assign_next_slot() {
return self_->assign_next_slot_to(this); return self_->assign_next_slot_to(this);
} }
......
...@@ -150,7 +150,7 @@ using fixture = test_coordinator_fixture<>; ...@@ -150,7 +150,7 @@ using fixture = test_coordinator_fixture<>;
CAF_TEST_FIXTURE_SCOPE(local_streaming_tests, fixture) CAF_TEST_FIXTURE_SCOPE(local_streaming_tests, fixture)
CAF_TEST(depth_3_pipeline_with_fork) { CAF_TEST(depth_3_pipeline_with_fork) {
auto src = sys.spawn(file_reader, 50u); auto src = sys.spawn(file_reader, 60u);
auto stg = sys.spawn(stream_multiplexer); auto stg = sys.spawn(stream_multiplexer);
auto snk1 = sys.spawn(sum_up); auto snk1 = sys.spawn(sum_up);
auto snk2 = sys.spawn(sum_up); auto snk2 = sys.spawn(sum_up);
...@@ -168,14 +168,14 @@ CAF_TEST(depth_3_pipeline_with_fork) { ...@@ -168,14 +168,14 @@ CAF_TEST(depth_3_pipeline_with_fork) {
run(); run();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u); CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u); CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk1).state.x, 1275); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk1).state.x, sum(60));
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk2).state.x, 1275); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk2).state.x, sum(60));
self->send_exit(stg, exit_reason::kill); self->send_exit(stg, exit_reason::kill);
} }
CAF_TEST(depth_3_pipeline_with_join) { CAF_TEST(depth_3_pipeline_with_join) {
auto src1 = sys.spawn(file_reader, 50u); auto src1 = sys.spawn(file_reader, 60u);
auto src2 = sys.spawn(file_reader, 50u); auto src2 = sys.spawn(file_reader, 60u);
auto stg = sys.spawn(stream_multiplexer); auto stg = sys.spawn(stream_multiplexer);
auto snk = sys.spawn(sum_up); auto snk = sys.spawn(sum_up);
auto& st = deref<stream_multiplexer_actor>(stg).state; auto& st = deref<stream_multiplexer_actor>(stg).state;
...@@ -192,46 +192,7 @@ CAF_TEST(depth_3_pipeline_with_join) { ...@@ -192,46 +192,7 @@ CAF_TEST(depth_3_pipeline_with_join) {
run(); run();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u); CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u); CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 2550); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, sum(60) * 2);
self->send_exit(stg, exit_reason::kill);
}
CAF_TEST(closing_downstreams_before_end_of_stream) {
auto src = sys.spawn(file_reader, 10000u);
auto stg = sys.spawn(stream_multiplexer);
auto snk1 = sys.spawn(sum_up);
auto snk2 = sys.spawn(sum_up);
auto& st = deref<stream_multiplexer_actor>(stg).state;
CAF_MESSAGE("connect sinks to the stage (fork)");
self->send(snk1, join_atom_v, stg);
self->send(snk2, join_atom_v, stg);
consume_messages();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_MESSAGE("connect source to the stage (fork)");
self->send(stg * src, "numbers.txt");
consume_messages();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 1u);
CAF_MESSAGE("do a single round of credit");
trigger_timeouts();
consume_messages();
CAF_MESSAGE("make sure the stream isn't done yet");
CAF_REQUIRE(!deref<file_reader_actor>(src).state.buf.empty());
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 1u);
CAF_MESSAGE("get the next not-yet-buffered integer");
auto next_pending = deref<file_reader_actor>(src).state.buf.front();
CAF_REQUIRE_GREATER(next_pending, 0);
auto sink1_result = sum(next_pending - 1);
CAF_MESSAGE("gracefully close sink 1, next pending: " << next_pending);
self->send(stg, close_atom_v, 0);
expect((close_atom, int32_t), from(self).to(stg));
CAF_MESSAGE("ship remaining elements");
run();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u);
CAF_CHECK_LESS(deref<sum_up_actor>(snk1).state.x, sink1_result);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk2).state.x, sum(10000));
self->send_exit(stg, exit_reason::kill); self->send_exit(stg, exit_reason::kill);
} }
......
...@@ -45,7 +45,6 @@ ...@@ -45,7 +45,6 @@
#include "caf/detail/stream_sink_impl.hpp" #include "caf/detail/stream_sink_impl.hpp"
#include "caf/detail/stream_source_impl.hpp" #include "caf/detail/stream_source_impl.hpp"
#include "caf/detail/stream_stage_impl.hpp" #include "caf/detail/stream_stage_impl.hpp"
#include "caf/detail/tick_emitter.hpp"
#include "caf/downstream_manager.hpp" #include "caf/downstream_manager.hpp"
#include "caf/downstream_msg.hpp" #include "caf/downstream_msg.hpp"
#include "caf/inbound_path.hpp" #include "caf/inbound_path.hpp"
...@@ -149,9 +148,6 @@ public: ...@@ -149,9 +148,6 @@ public:
/// Defines the container for storing message handlers. /// Defines the container for storing message handlers.
using behavior_type = behavior; using behavior_type = behavior;
/// The type of a single tick.
using clock_type = detail::tick_emitter::clock_type;
/// The type of a single tick. /// The type of a single tick.
using time_point = clock_type::time_point; using time_point = clock_type::time_point;
...@@ -163,21 +159,12 @@ public: ...@@ -163,21 +159,12 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
entity(actor_config& cfg, const char* cstr_name, time_point* global_time, entity(actor_config& cfg, const char* cstr_name, time_point* global_time)
duration_type credit_interval, duration_type force_batches_interval) : super(cfg),
: super(cfg), mbox(unit, unit, unit, unit, unit),
mbox(unit, unit, unit, unit, unit), name_(cstr_name),
name_(cstr_name), global_time_(global_time) {
global_time_(global_time), CAF_ASSERT(global_time_ != nullptr);
tick_emitter_(global_time == nullptr ? clock_type::now()
: *global_time) {
auto cycle = detail::gcd(credit_interval.count(),
force_batches_interval.count());
ticks_per_force_batches_interval =
static_cast<size_t>(force_batches_interval.count() / cycle);
ticks_per_credit_interval =
static_cast<size_t>(credit_interval.count() / cycle);
tick_emitter_.interval(duration_type{cycle});
} }
void enqueue(mailbox_element_ptr what, execution_unit*) override { void enqueue(mailbox_element_ptr what, execution_unit*) override {
...@@ -256,13 +243,13 @@ public: ...@@ -256,13 +243,13 @@ public:
public: public:
using super = stream_stage_driver<int32_t, downstream_manager>; using super = stream_stage_driver<int32_t, downstream_manager>;
driver(downstream_manager& out, vector<int32_t>* log) driver(downstream_manager& out, vector<int32_t>* log, const char* name)
: super(out), : super(out), log_(log), name(name) {
log_(log) {
// nop // nop
} }
void process(downstream<int>& out, vector<int>& batch) override { void process(downstream<int>& out, vector<int>& batch) override {
CAF_MESSAGE(name << " forwards " << batch.size() << " elements");
log_->insert(log_->end(), batch.begin(), batch.end()); log_->insert(log_->end(), batch.begin(), batch.end());
out.append(batch.begin(), batch.end()); out.append(batch.begin(), batch.end());
} }
...@@ -273,8 +260,9 @@ public: ...@@ -273,8 +260,9 @@ public:
private: private:
vector<int>* log_; vector<int>* log_;
const char* name;
}; };
forwarder = detail::make_stream_stage<driver>(this, &data); forwarder = detail::make_stream_stage<driver>(this, &data, name_);
auto res = forwarder->add_outbound_path(ref.ctrl()); auto res = forwarder->add_outbound_path(ref.ctrl());
CAF_MESSAGE(name_ << " starts forwarding to " << ref.name() CAF_MESSAGE(name_ << " starts forwarding to " << ref.name()
<< " on slot " << res.value()); << " on slot " << res.value());
...@@ -321,25 +309,9 @@ public: ...@@ -321,25 +309,9 @@ public:
scheduled_actor::handle_upstream_msg(slots, sender, x); scheduled_actor::handle_upstream_msg(slots, sender, x);
} }
void advance_time() { void tick() {
auto cycle = std::chrono::milliseconds(100); for (auto& kvp : stream_managers())
auto f = [&](tick_type x) { kvp.second->tick(now());
if (x % ticks_per_force_batches_interval == 0) {
// Force batches on all output paths.
for (auto& kvp : stream_managers())
kvp.second->out().force_emit_batches();
}
if (x % ticks_per_credit_interval == 0) {
// Fill credit on each input path up to 30.
auto& qs = get<dmsg_id::value>(mbox.queues()).queues();
for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get();
auto tts = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(this, tts, now(), cycle);
}
}
};
tick_emitter_.update(now(), f);
} }
inbound_path* make_inbound_path(stream_manager_ptr mgr, stream_slots slots, inbound_path* make_inbound_path(stream_manager_ptr mgr, stream_slots slots,
...@@ -375,7 +347,7 @@ public: ...@@ -375,7 +347,7 @@ public:
} }
time_point now() { time_point now() {
return global_time_ == nullptr ? clock_type::now() : *global_time_; return *global_time_;
} }
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
...@@ -388,7 +360,6 @@ public: ...@@ -388,7 +360,6 @@ public:
tick_type ticks_per_force_batches_interval; tick_type ticks_per_force_batches_interval;
tick_type ticks_per_credit_interval; tick_type ticks_per_credit_interval;
time_point* global_time_; time_point* global_time_;
detail::tick_emitter tick_emitter_;
}; };
struct msg_visitor { struct msg_visitor {
...@@ -447,6 +418,7 @@ struct msg_visitor { ...@@ -447,6 +418,7 @@ struct msg_visitor {
auto& dm = x.content().get_mutable_as<downstream_msg>(0); auto& dm = x.content().get_mutable_as<downstream_msg>(0);
auto f = detail::make_overload( auto f = detail::make_overload(
[&](downstream_msg::batch& y) { [&](downstream_msg::batch& y) {
TRACE(self->name(), batch, CAF_ARG(dm.slots), CAF_ARG(y.xs_size));
inptr->handle(y); inptr->handle(y);
if (inptr->mgr->done()) { if (inptr->mgr->done()) {
CAF_MESSAGE(self->name() CAF_MESSAGE(self->name()
...@@ -492,15 +464,7 @@ struct msg_visitor { ...@@ -492,15 +464,7 @@ struct msg_visitor {
struct fixture { struct fixture {
using scheduler_type = scheduler::test_coordinator; using scheduler_type = scheduler::test_coordinator;
struct timing_config { timespan max_batch_delay = defaults::stream::max_batch_delay;
timespan credit_interval = std::chrono::milliseconds(100);
timespan force_batches_interval = std::chrono::milliseconds(50);
timespan step = force_batches_interval;
};
timing_config tc;
actor_system_config cfg; actor_system_config cfg;
actor_system sys; actor_system sys;
...@@ -513,13 +477,11 @@ struct fixture { ...@@ -513,13 +477,11 @@ struct fixture {
entity& bob; entity& bob;
entity& carl; entity& carl;
static actor spawn(actor_system& sys, actor_id id, const char* name, static actor spawn(actor_system& sys, actor_id id, const char* name) {
timing_config& tc) {
actor_config conf; actor_config conf;
auto& clock = dynamic_cast<scheduler_type&>(sys.scheduler()).clock(); auto& clock = dynamic_cast<scheduler_type&>(sys.scheduler()).clock();
auto global_time = &clock.current_time; auto global_time = &clock.current_time;
return make_actor<entity>(id, node_id{}, &sys, conf, name, global_time, return make_actor<entity>(id, node_id{}, &sys, conf, name, global_time);
tc.credit_interval, tc.force_batches_interval);
} }
static entity& fetch(const actor& hdl) { static entity& fetch(const actor& hdl) {
...@@ -531,18 +493,21 @@ struct fixture { ...@@ -531,18 +493,21 @@ struct fixture {
caf::test::engine::argv())) caf::test::engine::argv()))
CAF_FAIL("parsing the config failed: " << to_string(err)); CAF_FAIL("parsing the config failed: " << to_string(err));
cfg.set("caf.scheduler.policy", "testing"); cfg.set("caf.scheduler.policy", "testing");
cfg.set("caf.stream.credit-policy", "token-based");
cfg.set("caf.stream.token-based-policy.batch-size", 50);
cfg.set("caf.stream.token-based-policy.buffer-size", 200);
return cfg; return cfg;
} }
fixture() fixture()
: sys(init_config(cfg)), : sys(init_config(cfg)),
sched(dynamic_cast<scheduler_type&>(sys.scheduler())), sched(dynamic_cast<scheduler_type&>(sys.scheduler())),
alice_hdl(spawn(sys, 0, "alice", tc)), alice_hdl(spawn(sys, 0, "alice")),
bob_hdl(spawn(sys, 1, "bob", tc)), bob_hdl(spawn(sys, 1, "bob")),
carl_hdl(spawn(sys, 2, "carl", tc)), carl_hdl(spawn(sys, 2, "carl")),
alice(fetch(alice_hdl)), alice(fetch(alice_hdl)),
bob(fetch(bob_hdl)), bob(fetch(bob_hdl)),
carl(fetch(carl_hdl)) { carl(fetch(carl_hdl)) {
// nop // nop
} }
...@@ -568,10 +533,10 @@ struct fixture { ...@@ -568,10 +533,10 @@ struct fixture {
template <class... Ts> template <class... Ts>
void next_cycle(Ts&... xs) { void next_cycle(Ts&... xs) {
entity* es[] = {&xs...}; entity* es[] = {&xs...};
CAF_MESSAGE("advance clock by " << tc.credit_interval.count() << "ns"); CAF_MESSAGE("advance clock by " << max_batch_delay);
sched.clock().current_time += tc.credit_interval; sched.clock().current_time += max_batch_delay;
for (auto e : es) for (auto e : es)
e->advance_time(); e->tick();
} }
template <class F, class... Ts> template <class F, class... Ts>
...@@ -583,10 +548,10 @@ struct fixture { ...@@ -583,10 +548,10 @@ struct fixture {
while (!std::all_of(std::begin(fs), std::end(fs), mailbox_empty)) while (!std::all_of(std::begin(fs), std::end(fs), mailbox_empty))
for (auto& f : fs) for (auto& f : fs)
f.self->mbox.new_round(1, f); f.self->mbox.new_round(1, f);
CAF_MESSAGE("advance clock by " << tc.step.count() << "ns"); CAF_MESSAGE("advance clock by " << max_batch_delay);
sched.clock().current_time += tc.step; sched.clock().current_time += max_batch_delay;
for (auto e : es) for (auto e : es)
e->advance_time(); e->tick();
} }
while (!pred()); while (!pred());
} }
...@@ -614,16 +579,12 @@ CAF_TEST_FIXTURE_SCOPE(native_streaming_classes_tests, fixture) ...@@ -614,16 +579,12 @@ CAF_TEST_FIXTURE_SCOPE(native_streaming_classes_tests, fixture)
CAF_TEST(depth_2_pipeline_30_items) { CAF_TEST(depth_2_pipeline_30_items) {
alice.start_streaming(bob, 30); alice.start_streaming(bob, 30);
loop(alice, bob); loop_until([&] { return done_streaming(); }, alice, bob);
next_cycle(alice, bob); // emit first ack_batch
loop(alice, bob);
next_cycle(alice, bob); // to emit final ack_batch
loop(alice, bob);
CAF_CHECK_EQUAL(bob.data, make_iota(0, 30)); CAF_CHECK_EQUAL(bob.data, make_iota(0, 30));
} }
CAF_TEST(depth_2_pipeline_2000_items) { CAF_TEST(depth_2_pipeline_500_items) {
constexpr size_t num_messages = 2000; constexpr size_t num_messages = 500;
alice.start_streaming(bob, num_messages); alice.start_streaming(bob, num_messages);
loop_until([&] { return done_streaming(); }, alice, bob); loop_until([&] { return done_streaming(); }, alice, bob);
CAF_CHECK_EQUAL(bob.data, make_iota(0, num_messages)); CAF_CHECK_EQUAL(bob.data, make_iota(0, num_messages));
...@@ -632,19 +593,13 @@ CAF_TEST(depth_2_pipeline_2000_items) { ...@@ -632,19 +593,13 @@ CAF_TEST(depth_2_pipeline_2000_items) {
CAF_TEST(depth_3_pipeline_30_items) { CAF_TEST(depth_3_pipeline_30_items) {
bob.forward_to(carl); bob.forward_to(carl);
alice.start_streaming(bob, 30); alice.start_streaming(bob, 30);
loop(alice, bob, carl); loop_until([&] { return done_streaming(); }, alice, bob, carl);
next_cycle(alice, bob, carl); // emit first ack_batch
loop(alice, bob, carl);
next_cycle(alice, bob, carl);
loop(alice, bob, carl);
next_cycle(alice, bob, carl); // emit final ack_batch
loop(alice, bob, carl);
CAF_CHECK_EQUAL(bob.data, make_iota(0, 30)); CAF_CHECK_EQUAL(bob.data, make_iota(0, 30));
CAF_CHECK_EQUAL(carl.data, make_iota(0, 30)); CAF_CHECK_EQUAL(carl.data, make_iota(0, 30));
} }
CAF_TEST(depth_3_pipeline_2000_items) { CAF_TEST(depth_3_pipeline_500_items) {
constexpr size_t num_messages = 2000; constexpr size_t num_messages = 500;
bob.forward_to(carl); bob.forward_to(carl);
alice.start_streaming(bob, num_messages); alice.start_streaming(bob, num_messages);
CAF_MESSAGE("loop over alice and bob until bob is congested"); CAF_MESSAGE("loop over alice and bob until bob is congested");
......
...@@ -40,33 +40,8 @@ TESTEE_SETUP(); ...@@ -40,33 +40,8 @@ TESTEE_SETUP();
using buf = std::deque<int>; using buf = std::deque<int>;
std::function<void(buf&)> init(size_t buf_size) {
return [=](buf& xs) {
xs.resize(buf_size);
std::iota(xs.begin(), xs.end(), 1);
};
}
void push_from_buf(buf& xs, downstream<int>& out, size_t num) {
CAF_MESSAGE("push " << num << " messages downstream");
auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i)
out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
}
std::function<bool(const buf&)> is_done(scheduled_actor* self) {
return [=](const buf& xs) {
if (xs.empty()) {
CAF_MESSAGE(self->name() << " exhausted its buffer");
return true;
}
return false;
};
}
template <class T, class Self> template <class T, class Self>
std::function<void(T&, const error&)> fin(Self* self) { auto fin(Self* self) {
return [=](T&, const error& err) { return [=](T&, const error& err) {
self->state.fin_called += 1; self->state.fin_called += 1;
if (err == none) { if (err == none) {
...@@ -102,18 +77,38 @@ TESTEE_STATE(file_reader) { ...@@ -102,18 +77,38 @@ TESTEE_STATE(file_reader) {
}; };
VARARGS_TESTEE(file_reader, size_t buf_size) { VARARGS_TESTEE(file_reader, size_t buf_size) {
auto init = [](size_t buf_size) {
return [=](buf& xs) {
xs.resize(buf_size);
std::iota(xs.begin(), xs.end(), 1);
};
};
auto push_from_buf = [](buf& xs, downstream<int>& out, size_t num) {
CAF_MESSAGE("push " << num << " messages downstream");
auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i)
out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
};
auto is_done = [self](const buf& xs) {
if (xs.empty()) {
CAF_MESSAGE(self->name() << " exhausted its buffer");
return true;
}
return false;
};
return { return {
[=](string& fname) -> result<stream<int>> { [=](string& fname) -> result<stream<int>> {
CAF_CHECK_EQUAL(fname, "numbers.txt"); CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true); CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return attach_stream_source(self, init(buf_size), push_from_buf, return attach_stream_source(self, init(buf_size), push_from_buf, is_done,
is_done(self), fin<buf>(self)); fin<buf>(self));
}, },
[=](string& fname, actor next) { [=](string& fname, actor next) {
CAF_CHECK_EQUAL(fname, "numbers.txt"); CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true); CAF_CHECK_EQUAL(self->mailbox().empty(), true);
attach_stream_source(self, next, init(buf_size), push_from_buf, attach_stream_source(self, next, init(buf_size), push_from_buf, is_done,
is_done(self), fin<buf>(self)); fin<buf>(self));
}, },
}; };
} }
...@@ -259,11 +254,7 @@ CAF_TEST(depth_2_pipeline_50_items) { ...@@ -259,11 +254,7 @@ CAF_TEST(depth_2_pipeline_50_items) {
expect((upstream_msg::ack_open), from(snk).to(src)); expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (a single batch)"); CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk)); expect((downstream_msg::batch), from(src).to(snk));
tick();
expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(src).to(src));
expect((upstream_msg::ack_batch), from(snk).to(src)); expect((upstream_msg::ack_batch), from(snk).to(src));
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk)); expect((downstream_msg::close), from(src).to(snk));
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 1275); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 1275);
CAF_MESSAGE("verify that each actor called its finalizer once"); CAF_MESSAGE("verify that each actor called its finalizer once");
...@@ -278,15 +269,11 @@ CAF_TEST(depth_2_pipeline_setup2_50_items) { ...@@ -278,15 +269,11 @@ CAF_TEST(depth_2_pipeline_setup2_50_items) {
CAF_MESSAGE("initiate stream handshake"); CAF_MESSAGE("initiate stream handshake");
self->send(src, "numbers.txt", snk); self->send(src, "numbers.txt", snk);
expect((string, actor), from(self).to(src).with("numbers.txt", snk)); expect((string, actor), from(self).to(src).with("numbers.txt", snk));
expect((open_stream_msg), from(strong_actor_ptr{nullptr}).to(snk)); expect((open_stream_msg), to(snk));
expect((upstream_msg::ack_open), from(snk).to(src)); expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (a single batch)"); CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk)); expect((downstream_msg::batch), from(src).to(snk));
tick();
expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(src).to(src));
expect((upstream_msg::ack_batch), from(snk).to(src)); expect((upstream_msg::ack_batch), from(snk).to(src));
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk)); expect((downstream_msg::close), from(src).to(snk));
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 1275); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 1275);
CAF_MESSAGE("verify that each actor called its finalizer once"); CAF_MESSAGE("verify that each actor called its finalizer once");
...@@ -311,9 +298,6 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) { ...@@ -311,9 +298,6 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) {
expect((upstream_msg::ack_open), from(snk).to(src)); expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (a single batch)"); CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk)); expect((downstream_msg::batch), from(src).to(snk));
tick();
expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(src).to(src));
expect((upstream_msg::ack_batch), from(snk).to(src)); expect((upstream_msg::ack_batch), from(snk).to(src));
CAF_MESSAGE("expect close message from src and then result from snk"); CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk)); expect((downstream_msg::close), from(src).to(snk));
...@@ -381,6 +365,9 @@ CAF_TEST(depth_2_pipeline_error_at_source) { ...@@ -381,6 +365,9 @@ CAF_TEST(depth_2_pipeline_error_at_source) {
CAF_MESSAGE("start data transmission (and abort source)"); CAF_MESSAGE("start data transmission (and abort source)");
hard_kill(src); hard_kill(src);
expect((downstream_msg::batch), from(src).to(snk)); expect((downstream_msg::batch), from(src).to(snk));
expect((downstream_msg::batch), from(src).to(snk));
expect((downstream_msg::batch), from(src).to(snk));
expect((downstream_msg::batch), from(src).to(snk));
expect((downstream_msg::forced_close), from(_).to(snk)); expect((downstream_msg::forced_close), from(_).to(snk));
CAF_MESSAGE("verify that the sink called its finalizer once"); CAF_MESSAGE("verify that the sink called its finalizer once");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
......
...@@ -684,7 +684,9 @@ public: ...@@ -684,7 +684,9 @@ public:
cfg.set("caf.middleman.manual-multiplexing", true); cfg.set("caf.middleman.manual-multiplexing", true);
cfg.set("caf.middleman.workers", size_t{0}); cfg.set("caf.middleman.workers", size_t{0});
} }
cfg.set("caf.stream.credit-policy", "testing"); cfg.set("caf.stream.credit-policy", "token-based");
cfg.set("caf.stream.token-based-policy.batch-size", 50);
cfg.set("caf.stream.token-based-policy.buffer-size", 200);
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