Commit 1feb1109 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Integrate desired batch sizes

parent e60feb3d
......@@ -35,7 +35,7 @@ public:
long credit() const override {
// We receive messages until we have exhausted all downstream credit and
// have filled our buffer to its minimum size.
return this->min_credit() + this->min_buffer_size();
return this->min_credit();
}
void emit_batches() override {
......@@ -55,6 +55,12 @@ public:
}
}
}
long desired_batch_size() const override {
// TODO: this is an O(n) computation, consider storing the result in a
// member variable for
return super::min_desired_batch_size();
}
};
} // namespace caf
......
......@@ -44,7 +44,7 @@ public:
long credit() const override {
// We receive messages until we have exhausted all downstream credit and
// have filled our buffer to its minimum size.
return this->min_credit() + this->min_buffer_size();
return this->min_credit();
}
void emit_batches() override {
......
......@@ -34,12 +34,6 @@ public:
// nop
}
using super::min_batch_size;
long min_batch_size() const override {
return 1;
}
long min_buffer_size() const override {
return 5;
}
......
......@@ -194,21 +194,19 @@ public:
}
long credit() const override {
// TODO: only return credit of the main stream?
return std::accumulate(
begin(), end(), std::numeric_limits<long>::max(),
[](long x, pointer y) { return std::min(x, y->credit()); });
}
long buffered() const override {
// TODO: only return how many items are buffered at the main stream?
return std::accumulate(begin(), end(), 0l, [](long x, const_pointer y) {
return x + y->buffered();
});
}
long min_batch_size() const override {
return main_stream().min_batch_size();
}
long min_buffer_size() const override {
return main_stream().min_buffer_size();
}
......@@ -217,10 +215,6 @@ public:
return main_stream().max_batch_delay();
}
void min_batch_size(long x) override {
main_stream().min_batch_size(x);
}
void max_batch_delay(duration x) override {
main_stream().max_batch_delay(x);
}
......
......@@ -54,18 +54,6 @@ public:
path_ptr find(const stream_id& sid, const actor_addr& x) override;
long high_watermark() const override;
long min_credit_assignment() const override;
long max_credit() const override;
void high_watermark(long x) override;
void min_credit_assignment(long x) override;
void max_credit(long x) override;
void assign_credit(long downstream_capacity) override;
long initial_credit(long downstream_capacity, path_type* x) override;
......
......@@ -26,7 +26,7 @@ namespace caf {
/// Type-erased policy for dispatching data to sinks.
class invalid_stream_scatterer : public stream_scatterer {
public:
invalid_stream_scatterer() = default;
invalid_stream_scatterer(local_actor* self = nullptr);
~invalid_stream_scatterer() override;
......@@ -43,6 +43,8 @@ public:
bool remove_path(const stream_id& sid, const actor_addr& x,
error reason, bool silent) override;
bool paths_clean() const override;
void close() override;
void abort(error reason) override;
......@@ -65,14 +67,12 @@ public:
long buffered() const override;
long min_batch_size() const override;
long desired_batch_size() const override;
long min_buffer_size() const override;
duration max_batch_delay() const override;
void min_batch_size(long x) override;
void max_batch_delay(duration x) override;
};
......
......@@ -43,7 +43,7 @@ public:
long credit() const override {
// We receive messages until we have exhausted all downstream credit and
// have filled our buffer to its minimum size.
return this->total_credit() + this->min_buffer_size();
return this->total_credit();
}
void emit_batches() override {
......@@ -64,6 +64,12 @@ public:
}
}
}
long desired_batch_size() const override {
// TODO: this is an O(n) computation, consider storing the result in a
// member variable for
return super::total_desired_batch_size();
}
};
} // namespace caf
......
......@@ -94,26 +94,6 @@ public:
/// `sid`, otherwise `nullptr`.
virtual path_ptr path_at(size_t index) = 0;
/// Returns the point at which an actor stops sending out demand immediately
/// (waiting for the available credit to first drop below the watermark).
virtual long high_watermark() const = 0;
/// Returns the minimum amount of credit required to send a `demand` message.
virtual long min_credit_assignment() const = 0;
/// Returns the maximum credit assigned to a single upstream actors.
virtual long max_credit() const = 0;
/// Sets the point at which an actor stops sending out demand immediately
/// (waiting for the available credit to first drop below the watermark).
virtual void high_watermark(long x) = 0;
/// Sets the minimum amount of credit required to send a `demand` message.
virtual void min_credit_assignment(long x) = 0;
/// Sets the maximum credit assigned to a single upstream actors.
virtual void max_credit(long x) = 0;
/// Assigns new credit to all sources.
virtual void assign_credit(long downstream_capacity) = 0;
......
......@@ -53,24 +53,9 @@ public:
void abort(error reason) override;
long high_watermark() const override;
long min_credit_assignment() const override;
long max_credit() const override;
void high_watermark(long x) override;
void min_credit_assignment(long x) override;
void max_credit(long x) override;
protected:
void emit_credits();
long high_watermark_;
long min_credit_assignment_;
long max_credit_;
std::vector<assignment_pair> assignment_vec_;
/// Listeners for the final result.
......
......@@ -103,9 +103,8 @@ public:
/// Returns the size of the output buffer.
virtual long buffered() const = 0;
/// Minimum amount of messages required to emit a batch. A value of 0
/// disables batch accumulation.
virtual long min_batch_size() const = 0;
/// Returns the downstream-requested size for a single batch.
virtual long desired_batch_size() const = 0;
/// Minimum amount of messages we wish to store at the actor in order to emit
/// new batches immediately when receiving new downstream demand. Usually
......@@ -116,10 +115,6 @@ public:
/// reached.
virtual duration max_batch_delay() const = 0;
/// Minimum amount of messages required to emit a batch. A value of 0
/// disables batch delays.
virtual void min_batch_size(long x) = 0;
/// Forces to actor to emit a batch even if the minimum batch size was not
/// reached.
virtual void max_batch_delay(duration x) = 0;
......
......@@ -143,18 +143,13 @@ public:
bool paths_clean() const override;
long min_batch_size() const override;
long min_buffer_size() const override;
duration max_batch_delay() const override;
void min_batch_size(long x) override;
void max_batch_delay(duration x) override;
protected:
long min_batch_size_;
duration max_batch_delay_;
};
......
......@@ -57,13 +57,26 @@ public:
}
bool generate_messages() override {
// produce new elements
auto capacity = out_.credit() - out_.buffered();
if (capacity <= 0)
// Produce new elements. If we have less elements buffered (bsize) than fit
// into a single batch (dbs = desired batch size) then we fill up the
// buffer up to that point. Otherwise, we fill up the buffer up to its
// capacity since we are currently waiting for downstream demand and can
// use the delay to store batches upfront.
size_t size_hint;
auto bsize = out_.buffered();
auto dbs = out_.desired_batch_size();
if (bsize < dbs) {
size_hint= static_cast<size_t>(dbs - bsize);
} else {
auto bmax = out_.min_buffer_size();
if (bsize < bmax)
size_hint = static_cast<size_t>(bmax - bsize);
else
return false;
}
downstream<typename DownstreamPolicy::value_type> ds{out_.buf()};
fun_(state_, ds, static_cast<size_t>(capacity));
return true;
fun_(state_, ds, size_hint);
return out_.buffered() != bsize;
}
state_type& state() {
......@@ -78,8 +91,12 @@ protected:
void downstream_demand(outbound_path* path, long) override {
CAF_LOG_TRACE(CAF_ARG(path));
if (!at_end()) {
auto dbs = out_.desired_batch_size();
generate_messages();
while (out_.buffered() >= dbs && out_.credit() >= dbs) {
push();
generate_messages();
}
} else if (out_.buffered() > 0) {
push();
} else {
......
......@@ -19,64 +19,23 @@
#ifndef CAF_TERMINAL_STREAM_SCATTERER_HPP
#define CAF_TERMINAL_STREAM_SCATTERER_HPP
#include "caf/stream_scatterer.hpp"
#include "caf/invalid_stream_scatterer.hpp"
namespace caf {
/// Special-purpose scatterer for sinks that terminate a stream. A terminal
/// stream scatterer generates credit without downstream actors.
class terminal_stream_scatterer : public stream_scatterer {
class terminal_stream_scatterer : public invalid_stream_scatterer {
public:
using super = invalid_stream_scatterer;
terminal_stream_scatterer(local_actor* = nullptr);
~terminal_stream_scatterer() override;
path_ptr add_path(const stream_id& sid, strong_actor_ptr origin,
strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages,
message_id handshake_mid, message handshake_data,
stream_priority prio, bool redeployable) override;
path_ptr confirm_path(const stream_id& sid, const actor_addr& from,
strong_actor_ptr to, long initial_demand,
long desired_batch_size, bool redeployable) override;
bool remove_path(const stream_id& sid, const actor_addr& x,
error reason, bool silent) override;
bool paths_clean() const override;
void close() override;
void abort(error reason) override;
long num_paths() const override;
bool closed() const override;
bool continuous() const override;
void continuous(bool value) override;
path_type* path_at(size_t index) override;
void emit_batches() override;
path_type* find(const stream_id& sid, const actor_addr& x) override;
long credit() const override;
long buffered() const override;
long min_batch_size() const override;
long min_buffer_size() const override;
duration max_batch_delay() const override;
void min_batch_size(long x) override;
void max_batch_delay(duration x) override;
long desired_batch_size() const override;
};
} // namespace caf
......
......@@ -73,30 +73,6 @@ stream_gatherer::path_type* invalid_stream_gatherer::find(const stream_id&,
return nullptr;
}
long invalid_stream_gatherer::high_watermark() const {
return 0;
}
long invalid_stream_gatherer::min_credit_assignment() const {
return 0;
}
long invalid_stream_gatherer::max_credit() const {
return 0;
}
void invalid_stream_gatherer::high_watermark(long) {
// nop
}
void invalid_stream_gatherer::min_credit_assignment(long) {
// nop
}
void invalid_stream_gatherer::max_credit(long) {
// nop
}
void invalid_stream_gatherer::assign_credit(long) {
// nop
}
......
......@@ -22,6 +22,10 @@
namespace caf {
invalid_stream_scatterer::invalid_stream_scatterer(local_actor*) {
// nop
}
invalid_stream_scatterer::~invalid_stream_scatterer() {
// nop
}
......@@ -48,6 +52,10 @@ bool invalid_stream_scatterer::remove_path(const stream_id&, const actor_addr&,
return false;
}
bool invalid_stream_scatterer::paths_clean() const {
return true;
}
void invalid_stream_scatterer::close() {
// nop
}
......@@ -93,7 +101,7 @@ long invalid_stream_scatterer::buffered() const {
return 0;
}
long invalid_stream_scatterer::min_batch_size() const {
long invalid_stream_scatterer::desired_batch_size() const {
return 0;
}
......@@ -105,10 +113,6 @@ duration invalid_stream_scatterer::max_batch_delay() const {
return infinite;
}
void invalid_stream_scatterer::min_batch_size(long) {
// nop
}
void invalid_stream_scatterer::max_batch_delay(duration) {
// nop
}
......
......@@ -31,8 +31,13 @@ random_gatherer::~random_gatherer() {
void random_gatherer::assign_credit(long available) {
CAF_LOG_TRACE(CAF_ARG(available));
if (available <= 0 || assignment_vec_.empty())
return;
auto max_credit = available / static_cast<long>(assignment_vec_.size());
if (max_credit == 0)
return;
for (auto& kvp : assignment_vec_) {
auto x = std::min(available, max_credit() - kvp.first->assigned_credit);
auto x = std::min(available, max_credit - kvp.first->assigned_credit);
available -= x;
kvp.second = x;
}
......@@ -40,7 +45,7 @@ void random_gatherer::assign_credit(long available) {
}
long random_gatherer::initial_credit(long available, path_type*) {
return std::min(available, max_credit());
return available;
}
/*
......
......@@ -21,10 +21,7 @@
namespace caf {
stream_gatherer_impl::stream_gatherer_impl(local_actor* selfptr)
: super(selfptr),
high_watermark_(40),
min_credit_assignment_(1),
max_credit_(50) {
: super(selfptr) {
// nop
}
......@@ -96,30 +93,6 @@ void stream_gatherer_impl::abort(error reason) {
listeners_.clear();
}
long stream_gatherer_impl::high_watermark() const {
return high_watermark_;
}
long stream_gatherer_impl::min_credit_assignment() const {
return min_credit_assignment_;
}
long stream_gatherer_impl::max_credit() const {
return max_credit_;
}
void stream_gatherer_impl::high_watermark(long x) {
high_watermark_ = x;
}
void stream_gatherer_impl::min_credit_assignment(long x) {
min_credit_assignment_ = x;
}
void stream_gatherer_impl::max_credit(long x) {
max_credit_ = x;
}
void stream_gatherer_impl::emit_credits() {
for (auto& kvp : assignment_vec_)
if (kvp.second > 0)
......
......@@ -25,7 +25,6 @@ namespace caf {
stream_scatterer_impl::stream_scatterer_impl(local_actor* selfptr)
: super(selfptr),
min_batch_size_(1),
max_batch_delay_(infinite) {
// nop
}
......@@ -120,10 +119,6 @@ long stream_scatterer_impl::max_desired_batch_size() const {
return max_desired_batch_size(paths_);
}
long stream_scatterer_impl::min_batch_size() const {
return min_batch_size_;
}
long stream_scatterer_impl::min_buffer_size() const {
return 50; // TODO: at least place the default in a header
}
......@@ -132,10 +127,6 @@ duration stream_scatterer_impl::max_batch_delay() const {
return max_batch_delay_;
}
void stream_scatterer_impl::min_batch_size(long x) {
min_batch_size_ = x;
}
void stream_scatterer_impl::max_batch_delay(duration x) {
max_batch_delay_ = std::move(x);
}
......
......@@ -22,7 +22,8 @@
namespace caf {
terminal_stream_scatterer::terminal_stream_scatterer(local_actor*) {
terminal_stream_scatterer::terminal_stream_scatterer(local_actor* ptr)
: super(ptr) {
// nop
}
......@@ -30,97 +31,14 @@ terminal_stream_scatterer::~terminal_stream_scatterer() {
// nop
}
stream_scatterer::path_ptr
terminal_stream_scatterer::add_path(const stream_id&, strong_actor_ptr,
strong_actor_ptr,
mailbox_element::forwarding_stack,
message_id, message, stream_priority,
bool) {
CAF_LOG_ERROR("terminal_stream_scatterer::add_path called");
return nullptr;
}
stream_scatterer::path_ptr
terminal_stream_scatterer::confirm_path(const stream_id&, const actor_addr&,
strong_actor_ptr, long, long, bool) {
CAF_LOG_ERROR("terminal_stream_scatterer::confirm_path called");
return nullptr;
}
bool terminal_stream_scatterer::remove_path(const stream_id&, const actor_addr&,
error, bool) {
CAF_LOG_ERROR("terminal_stream_scatterer::remove_path called");
return false;
}
bool terminal_stream_scatterer::paths_clean() const {
return true;
}
void terminal_stream_scatterer::close() {
// nop
}
void terminal_stream_scatterer::abort(error) {
// nop
}
long terminal_stream_scatterer::num_paths() const {
return 0;
}
bool terminal_stream_scatterer::closed() const {
return true;
}
bool terminal_stream_scatterer::continuous() const {
return false;
}
void terminal_stream_scatterer::continuous(bool) {
// nop
}
stream_scatterer::path_type* terminal_stream_scatterer::path_at(size_t) {
return nullptr;
}
void terminal_stream_scatterer::emit_batches() {
// nop
}
stream_scatterer::path_type*
terminal_stream_scatterer::find(const stream_id&, const actor_addr&) {
return nullptr;
}
long terminal_stream_scatterer::credit() const {
// TODO: do something more advanced, yes?
return 50;
}
long terminal_stream_scatterer::buffered() const {
return 0;
}
long terminal_stream_scatterer::min_batch_size() const {
return 0;
}
long terminal_stream_scatterer::min_buffer_size() const {
return 0;
}
duration terminal_stream_scatterer::max_batch_delay() const {
return infinite;
}
void terminal_stream_scatterer::min_batch_size(long) {
// nop
}
void terminal_stream_scatterer::max_batch_delay(duration) {
// nop
long terminal_stream_scatterer::desired_batch_size() const {
// TODO: do something more advanced, yes?
return 50;
}
} // namespace caf
......@@ -95,8 +95,6 @@ behavior stream_splitter(stateful_actor<stream_splitter_state>* self) {
// Force the splitter to collect credit until reaching 3 in order
// to receive only full batches from upstream (simplifies testing).
// Restrict maximum credit per path to 5 (simplifies testing).
self->state.stage->in().min_credit_assignment(3);
self->state.stage->in().max_credit(5);
self->streams().emplace(id, self->state.stage);
return {
[=](join_atom, filter_type filter) -> stream<element_type> {
......
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