Commit a5817717 authored by Dominik Charousset's avatar Dominik Charousset

Tweak credit computation on downstream paths

parent fddb27fa
......@@ -65,50 +65,50 @@ public:
/// Returns the total available credit for all sinks in `xs` in O(n).
template <class PathContainer>
static size_t total_credit(const PathContainer& xs) {
static long total_credit(const PathContainer& xs) {
return fold(xs, 0u,
[](size_t x, const typename PathContainer::value_type& y) {
[=](long x, const typename PathContainer::value_type& y) {
return x + y->open_credit;
});
}
/// Returns the total available credit for all sinks in `paths_` in O(n).
size_t total_credit() const;
long total_credit() const;
/// Returns the maximum credit of all sinks in `paths_` in O(n).
template <class PathContainer>
static size_t max_credit(const PathContainer& xs) {
static long max_credit(const PathContainer& xs) {
return fold(xs, 0,
[](size_t x, const typename PathContainer::value_type& y) {
[](long x, const typename PathContainer::value_type& y) {
return std::max(x, y->open_credit);
});
}
/// Returns the maximum credit of all sinks in `paths_` in O(n).
size_t max_credit() const;
long max_credit() const;
/// Returns the minimal credit of all sinks in `xs` in O(n).
template <class PathContainer>
static size_t min_credit(const PathContainer& xs) {
return fold(xs, std::numeric_limits<size_t>::max(),
[](size_t x, const typename PathContainer::value_type& y) {
static long min_credit(const PathContainer& xs) {
return fold(xs, std::numeric_limits<long>::max(),
[](long x, const typename PathContainer::value_type& y) {
return std::min(x, y->open_credit);
});
}
/// Returns the minimal credit of all sinks in `paths_` in O(n).
size_t min_credit() const;
/// Returns the minimal net credit of all sinks in `paths_` in O(n).
long min_credit() const;
/// Broadcasts the first `*hint` elements of the buffer on all paths. If
/// `hint == nullptr` then `min_credit()` is used instead.
virtual void broadcast(size_t* hint = nullptr) = 0;
virtual void broadcast(long* hint = nullptr) = 0;
/// Sends `*hint` elements of the buffer to available paths. If
/// `hint == nullptr` then `total_credit()` is used instead.
virtual void anycast(size_t* hint = nullptr) = 0;
virtual void anycast(long* hint = nullptr) = 0;
/// Returns the size of the output buffer.
virtual size_t buf_size() const = 0;
virtual long buf_size() const = 0;
/// Adds a path with in-flight `stream_msg::open` message.
bool add_path(strong_actor_ptr ptr);
......@@ -140,7 +140,9 @@ public:
path* find(const strong_actor_ptr& ptr) const;
size_t available_credit() const;
long total_net_credit() const;
virtual long num_paths() const;
inline local_actor* self() const {
return self_;
......@@ -158,8 +160,14 @@ public:
return paths_.empty();
}
/// Returns how many items should be stored on individual paths in order to
/// minimize latency between received demand and sent batches.
long min_buffer_size() const {
return min_buffer_size_;
}
protected:
void send_batch(downstream_path& dest, size_t chunk_size, message chunk);
void send_batch(downstream_path& dest, long chunk_size, message chunk);
/// Sorts `xs` in descending order by available credit.
template <class PathContainer>
......@@ -174,19 +182,17 @@ protected:
/// Sorts `paths_` in descending order by available credit.
void sort_by_credit();
template <class PathContainer, class F>
static size_t fold(PathContainer& xs, size_t init, F f) {
template <class T, class PathContainer, class F>
static T fold(PathContainer& xs, T init, F f) {
auto b = xs.begin();
auto e = xs.end();
auto g = [&](size_t x, const typename PathContainer::value_type& y) {
return f(x, y);
};
return b != e ? std::accumulate(b, e, init, g) : 0;
return b != e ? std::accumulate(b, e, init, f) : 0;
}
local_actor* self_;
stream_id sid_;
path_list paths_;
long min_buffer_size_;
std::unique_ptr<downstream_policy> policy_;
};
......
......@@ -58,12 +58,12 @@ public:
void abort(strong_actor_ptr& cause, const error& reason);
/// Assigns credit to upstream actors according to the configured policy.
void assign_credit(size_t buf_size, size_t downstream_credit);
void assign_credit(long total_downstream_net_credit);
/// Adds a new upstream actor and returns the initial credit.
expected<size_t> add_path(strong_actor_ptr hdl, const stream_id& sid,
stream_priority prio, size_t buf_size,
size_t downstream_credit);
expected<long> add_path(strong_actor_ptr hdl, const stream_id& sid,
stream_priority prio,
long total_downstream_net_credit);
bool remove_path(const strong_actor_ptr& hdl);
......
......@@ -57,9 +57,9 @@ public:
return buf_;
}
void broadcast(size_t* hint) override {
void broadcast(long* hint) override {
auto chunk = get_chunk(hint ? *hint : min_credit());
auto csize = chunk.size();
auto csize = static_cast<long>(chunk.size());
if (csize == 0)
return;
auto wrapped_chunk = make_message(std::move(chunk));
......@@ -69,7 +69,7 @@ public:
}
}
void anycast(size_t*) override {
void anycast(long*) override {
this->sort_by_credit();
for (auto& x : paths_) {
auto chunk = get_chunk(x->open_credit);
......@@ -77,21 +77,23 @@ public:
if (csize == 0)
return;
x->open_credit -= csize;
send_batch(*x, csize, std::move(make_message(std::move(chunk))));
send_batch(*x, static_cast<long>(csize),
std::move(make_message(std::move(chunk))));
}
}
size_t buf_size() const override {
return buf_.size();
long buf_size() const override {
return static_cast<long>(buf_.size());
}
protected:
/// @pre `n <= buf_.size()`
static chunk_type get_chunk(queue_type& buf, size_t n) {
static chunk_type get_chunk(queue_type& buf, long n) {
CAF_ASSERT(n >= 0);
chunk_type xs;
if (n > 0) {
xs.reserve(n);
if (n < buf.size()) {
xs.reserve(static_cast<size_t>(n));
if (static_cast<size_t>(n) < buf.size()) {
auto first = buf.begin();
auto last = first + static_cast<ptrdiff_t>(n);
std::move(first, last, std::back_inserter(xs));
......@@ -105,7 +107,7 @@ protected:
}
/// @pre `n <= buf_.size()`
chunk_type get_chunk(size_t n) {
chunk_type get_chunk(long n) {
return get_chunk(buf_, n);
}
......
......@@ -43,7 +43,7 @@ public:
int64_t next_batch_id;
/// Currently available credit for this path.
size_t open_credit;
long open_credit;
/// Stores whether the downstream actor is failsafe, i.e., allows the runtime
/// to redeploy it on failure. If this field is set to `false` then
......
......@@ -31,12 +31,13 @@ class downstream_policy {
public:
virtual ~downstream_policy();
/// Queries the optimal amount of data for the next `push` operation to `x`.
virtual size_t available_credit(const abstract_downstream& x) = 0;
/// Returns an accumulated value of all individual net credits. For example,
/// a broadcast policy would return the minimum value.
virtual long total_net_credit(const abstract_downstream& x) = 0;
/// Pushes data to the downstream paths of `x`, optionally passing the last
/// result of `available_credit` for `x` as second argument.
virtual void push(abstract_downstream& x, size_t* hint = nullptr) = 0;
virtual void push(abstract_downstream& x, long* hint = nullptr) = 0;
// TODO: callbacks für new and closed downstream pahts
......
......@@ -36,7 +36,7 @@ namespace caf {
/// of workers in order to handle only a subset of the overall data on each
/// lane.
template <class T, class Key, class KeyCompare = std::equal_to<Key>,
size_t KeyIndex = 0>
long KeyIndex = 0>
class filtering_downstream : public downstream<T> {
public:
/// Base type.
......@@ -59,13 +59,13 @@ public:
// nop
}
void broadcast(size_t* hint) override {
void broadcast(long* hint) override {
fan_out();
for (auto& kvp : lanes_) {
auto& l = kvp.second;
auto chunk = super::get_chunk(l.queue,
hint ? *hint : super::min_credit(l.paths));
auto csize = chunk.size();
auto csize = static_cast<long>(chunk.size());
if (csize == 0)
continue;
auto wrapped_chunk = make_message(std::move(chunk));
......@@ -76,14 +76,14 @@ public:
}
}
void anycast(size_t*) override {
void anycast(long*) override {
fan_out();
for (auto& kvp : lanes_) {
auto& l = kvp.second;
super::sort_by_credit(l.paths);
for (auto& x : l.paths) {
auto chunk = super::get_chunk(l.queue, x->open_credit);
auto csize = chunk.size();
auto csize = static_cast<long>(chunk.size());
if (csize == 0)
break;
x->open_credit -= csize;
......
......@@ -42,8 +42,8 @@ public:
}
error confirm_downstream(const strong_actor_ptr& rebind_from,
strong_actor_ptr& ptr, size_t initial_demand,
bool redeployable) final {
strong_actor_ptr& ptr, long initial_demand,
bool redeployable) override {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(initial_demand)
<< CAF_ARG(redeployable));
CAF_ASSERT(ptr != nullptr);
......@@ -54,7 +54,7 @@ public:
return sec::invalid_downstream;
}
error push(size_t* hint = nullptr) override {
error push(long* hint = nullptr) override {
CAF_LOG_TRACE("");
if (out().buf_size() > 0)
out().policy().push(out(), hint);
......
......@@ -27,8 +27,8 @@ namespace policy {
class anycast final : public downstream_policy {
public:
void push(abstract_downstream& out, size_t* hint) override;
size_t available_credit(const abstract_downstream& out) override;
long total_net_credit(const abstract_downstream& out) override;
void push(abstract_downstream& out, long* hint) override;
};
} // namespace policy
......
......@@ -27,8 +27,12 @@ namespace policy {
class broadcast final : public downstream_policy {
public:
void push(abstract_downstream& out, size_t* hint) override;
size_t available_credit(const abstract_downstream& out) override;
void push(abstract_downstream& out, long* hint) override;
long total_net_credit(const abstract_downstream& x) override;
inline static std::unique_ptr<downstream_policy> make() {
return std::unique_ptr<downstream_policy>(new broadcast);
}
};
} // namespace policy
......
......@@ -30,14 +30,16 @@ class greedy final : public upstream_policy {
public:
greedy();
void assign_credit(assignment_vec& xs, size_t buf_size,
size_t downstream_credit) override;
void assign_credit(assignment_vec& xs,
long total_downstream_net_credit) override;
size_t low_watermark;
long low_watermark;
size_t high_watermark;
long high_watermark;
size_t min_buffer_size;
inline static std::unique_ptr<upstream_policy> make() {
return std::unique_ptr<upstream_policy>(new greedy);
}
};
} // namespace policy
......
......@@ -49,28 +49,28 @@ public:
/// the downstream actor on failure.
/// @pre `hdl != nullptr`
virtual error confirm_downstream(const strong_actor_ptr& rebind_from,
strong_actor_ptr& hdl, size_t initial_demand,
strong_actor_ptr& hdl, long initial_demand,
bool redeployable);
/// Handles new demand from a downstream actor.
/// @pre `hdl != nullptr`
/// @pre `new_demand > 0`
virtual error downstream_demand(strong_actor_ptr& hdl, size_t new_demand);
virtual error downstream_demand(strong_actor_ptr& hdl, long new_demand);
/// Push new data to downstream actors by sending batches. The amount of
/// pushed data is limited by `hint` or the available credit if
/// `hint == nullptr`.
virtual error push(size_t* hint = nullptr);
virtual error push(long* hint = nullptr);
// -- handler for upstream events --------------------------------------------
/// Add a new upstream actor to the stream and return an initial credit.
virtual expected<size_t> add_upstream(strong_actor_ptr& hdl,
virtual expected<long> add_upstream(strong_actor_ptr& hdl,
const stream_id& sid,
stream_priority prio);
/// Handles data from an upstream actor.
virtual error upstream_batch(strong_actor_ptr& hdl, size_t, message& xs);
virtual error upstream_batch(strong_actor_ptr& hdl, long, message& xs);
/// Closes an upstream.
virtual error close_upstream(strong_actor_ptr& hdl);
......
......@@ -41,10 +41,10 @@ public:
bool done() const override;
error upstream_batch(strong_actor_ptr& src, size_t xs_size,
message& xs) final;
error upstream_batch(strong_actor_ptr& src, long xs_size,
message& xs) override;
void abort(strong_actor_ptr& cause, const error& reason) final;
void abort(strong_actor_ptr& cause, const error& reason) override;
void last_upstream_closed();
......@@ -52,6 +52,10 @@ public:
return *in_ptr_;
}
long min_buffer_size() const {
return min_buffer_size_;
}
protected:
/// Consumes a batch.
virtual error consume(message& xs) = 0;
......@@ -64,6 +68,7 @@ private:
strong_actor_ptr original_sender_;
std::vector<strong_actor_ptr> next_stages_;
message_id original_msg_id_;
long min_buffer_size_;
};
} // namespace caf
......
......@@ -48,14 +48,14 @@ public:
// nop
}
expected<size_t> add_upstream(strong_actor_ptr& ptr, const stream_id& sid,
stream_priority prio) final {
expected<long> add_upstream(strong_actor_ptr& ptr, const stream_id& sid,
stream_priority prio) override {
if (ptr)
return in().add_path(ptr, sid, prio, 0, 0);
return in().add_path(ptr, sid, prio, min_buffer_size());
return sec::invalid_argument;
}
error consume(message& msg) final {
error consume(message& msg) override {
using vec_type = std::vector<input_type>;
if (msg.match_elements<vec_type>()) {
auto& xs = msg.get_as<vec_type>(0);
......@@ -66,11 +66,11 @@ public:
return sec::unexpected_message;
}
message finalize() final {
message finalize() override {
return trait::make_result(state_, fin_);
}
optional<abstract_upstream&> get_upstream() final {
optional<abstract_upstream&> get_upstream() override {
return in_;
}
......
......@@ -39,7 +39,7 @@ public:
bool done() const override;
error downstream_demand(strong_actor_ptr& hdl, size_t value) override;
error downstream_demand(strong_actor_ptr& hdl, long value) override;
void abort(strong_actor_ptr& cause, const error& reason) override;
......@@ -49,7 +49,7 @@ public:
protected:
/// Queries the current amount of elements in the output buffer.
virtual size_t buf_size() const = 0;
virtual long buf_size() const = 0;
/// Generate new elements for the output buffer. The size hint `n` indicates
/// how much elements can be shipped immediately.
......
......@@ -46,19 +46,19 @@ public:
// nop
}
void generate(size_t num) final {
void generate(size_t num) override {
fun_(state_, out_, num);
}
size_t buf_size() const final {
return out_.buf().size();
long buf_size() const override {
return static_cast<long>(out_.buf().size());
}
bool at_end() const final {
bool at_end() const override {
return pred_(state_);
}
optional<abstract_downstream&> get_downstream() final {
optional<abstract_downstream&> get_downstream() override {
return out_;
}
......
......@@ -43,9 +43,9 @@ public:
void abort(strong_actor_ptr&, const error&) override;
error downstream_demand(strong_actor_ptr&, size_t) override;
error downstream_demand(strong_actor_ptr&, long) override;
error upstream_batch(strong_actor_ptr&, size_t, message&) override;
error upstream_batch(strong_actor_ptr&, long, message&) override;
void last_upstream_closed();
......
......@@ -63,15 +63,15 @@ public:
// nop
}
expected<size_t> add_upstream(strong_actor_ptr& ptr, const stream_id& sid,
stream_priority prio) final {
expected<long> add_upstream(strong_actor_ptr& ptr, const stream_id& sid,
stream_priority prio) override {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(sid) << CAF_ARG(prio));
if (ptr)
return in().add_path(ptr, sid, prio, out_.buf_size(),
out_.available_credit());
return in().add_path(ptr, sid, prio, out_.total_net_credit());
return sec::invalid_argument;
}
error process_batch(message& msg) final {
error process_batch(message& msg) override {
using vec_type = std::vector<output_type>;
if (msg.match_elements<vec_type>()) {
auto& xs = msg.get_as<vec_type>(0);
......@@ -82,15 +82,15 @@ public:
return sec::unexpected_message;
}
message make_output_token(const stream_id& x) const final {
message make_output_token(const stream_id& x) const override {
return make_message(stream<output_type>{x});
}
optional<abstract_downstream&> get_downstream() final {
optional<abstract_downstream&> get_downstream() override {
return out_;
}
optional<abstract_upstream&> get_upstream() final {
optional<abstract_upstream&> get_upstream() override {
return in_;
}
......
......@@ -51,7 +51,7 @@ public:
int64_t last_batch_id;
/// Amount of credit we have signaled upstream.
size_t assigned_credit;
long assigned_credit;
upstream_path(strong_actor_ptr ptr, stream_id id, stream_priority p);
};
......
......@@ -33,7 +33,7 @@ public:
virtual ~upstream_policy();
/// Describes an assignment of credit to an upstream actor.
using assignment_pair = std::pair<upstream_path*, size_t>;
using assignment_pair = std::pair<upstream_path*, long>;
/// Describes an assignment of credit to all upstream actors.
using assignment_vec = std::vector<assignment_pair>;
......@@ -42,12 +42,12 @@ public:
/// @param xs Stores assignment decisions. Note that the second element of
/// each pair is uninitialized and must be set to 0 for all paths
/// that do not receive credit.
/// @param buf_size Denotes how many stream elements are currently buffered.
/// @param downstream_credit Denotes how many items we could send downstream.
/// Usually 0, unless the downstream policy avoids
/// small batches.
virtual void assign_credit(assignment_vec& xs, size_t buf_size,
size_t downstream_credit) = 0;
/// @param total_downstream_net_credit Denotes how many items we could send
/// downstream. A negative value indicates
/// that we are already buffering more
/// items than we can send.
virtual void assign_credit(assignment_vec& xs,
long total_downstream_net_credit) = 0;
};
} // namespace caf
......
......@@ -32,6 +32,7 @@ abstract_downstream::abstract_downstream(local_actor* selfptr,
std::unique_ptr<downstream_policy> ptr)
: self_(selfptr),
sid_(sid),
min_buffer_size_(5), // TODO: make configurable
policy_(std::move(ptr)) {
// nop
}
......@@ -40,15 +41,15 @@ abstract_downstream::~abstract_downstream() {
// nop
}
size_t abstract_downstream::total_credit() const {
long abstract_downstream::total_credit() const {
return total_credit(paths_);
}
size_t abstract_downstream::max_credit() const {
long abstract_downstream::max_credit() const {
return max_credit(paths_);
}
size_t abstract_downstream::min_credit() const {
long abstract_downstream::min_credit() const {
return min_credit(paths_);
}
......@@ -116,11 +117,15 @@ abstract_downstream::find(const strong_actor_ptr& ptr) const {
return find(paths_, ptr);
}
size_t abstract_downstream::available_credit() const {
return policy().available_credit(*this);
long abstract_downstream::total_net_credit() const {
return policy().total_net_credit(*this);
}
void abstract_downstream::send_batch(downstream_path& dest, size_t chunk_size,
long abstract_downstream::num_paths() const {
return static_cast<long>(paths_.size());
}
void abstract_downstream::send_batch(downstream_path& dest, long chunk_size,
message chunk) {
auto scs = static_cast<int32_t>(chunk_size);
auto batch_id = dest.next_batch_id++;
......
......@@ -42,9 +42,8 @@ void abstract_upstream::abort(strong_actor_ptr& cause, const error& reason) {
unsafe_send_as(self_, x->hdl, make<stream_msg::abort>(x->sid, reason));
}
void abstract_upstream::assign_credit(size_t buf_size,
size_t downstream_credit) {
policy_->assign_credit(policy_vec_, buf_size, downstream_credit);
void abstract_upstream::assign_credit(long total_downstream_net_credit) {
policy_->assign_credit(policy_vec_, total_downstream_net_credit);
for (auto& x : policy_vec_) {
auto n = x.second;
if (n > 0) {
......@@ -57,12 +56,12 @@ void abstract_upstream::assign_credit(size_t buf_size,
}
}
expected<size_t> abstract_upstream::add_path(strong_actor_ptr hdl,
const stream_id& sid,
stream_priority prio,
size_t buf_size,
size_t downstream_credit) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(sid) << CAF_ARG(prio));
expected<long> abstract_upstream::add_path(strong_actor_ptr hdl,
const stream_id& sid,
stream_priority prio,
long total_downstream_net_credit) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(sid) << CAF_ARG(prio)
<< CAF_ARG(total_downstream_net_credit));
CAF_ASSERT(hdl != nullptr);
if (!find(hdl)) {
CAF_LOG_DEBUG("add new upstraem path" << CAF_ARG(hdl));
......@@ -71,7 +70,7 @@ expected<size_t> abstract_upstream::add_path(strong_actor_ptr hdl,
// use a one-shot actor to calculate initial credit
upstream_policy::assignment_vec tmp;
tmp.emplace_back(paths_.back().get(), 0);
policy_->assign_credit(tmp, buf_size, downstream_credit);
policy_->assign_credit(tmp, total_downstream_net_credit);
paths_.back()->assigned_credit += tmp.back().second;
return tmp.back().second;
}
......@@ -79,6 +78,7 @@ expected<size_t> abstract_upstream::add_path(strong_actor_ptr hdl,
}
bool abstract_upstream::remove_path(const strong_actor_ptr& hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
// Find element in our paths list.
auto has_hdl = [&](const path_uptr& x) {
CAF_ASSERT(x != nullptr);
......
......@@ -24,12 +24,17 @@
namespace caf {
namespace policy {
void anycast::push(abstract_downstream& out, size_t* hint) {
out.anycast(hint);
long anycast::total_net_credit(const abstract_downstream& out) {
// The total amount of available net credit is calculated as:
// `av + (n * mb) - bs`, where `av` is the sum of all available credit on all
// paths, `n` is the number of downstream paths, `mb` is the minimum buffer
// size, and `bs` is the current buffer size.
return (out.total_credit() + (out.num_paths() * out.min_buffer_size()))
- out.buf_size();
}
size_t anycast::available_credit(const abstract_downstream& out) {
return out.total_credit();
void anycast::push(abstract_downstream& out, long* hint) {
out.anycast(hint);
}
} // namespace policy
......
......@@ -24,12 +24,16 @@
namespace caf {
namespace policy {
void broadcast::push(abstract_downstream& out, size_t* hint) {
out.broadcast(hint);
long broadcast::total_net_credit(const abstract_downstream& out) {
// The buffer on `out` is shared on all paths. Our total available number of
// net credit is thus calculates as `av_min + mb - bs`, where `av_min` is the
// minimum of available credits on all paths, `mb` is the minimum buffer
// size, and `bs` is the current buffer size.
return out.min_credit() + out.min_buffer_size() - out.buf_size();
}
size_t broadcast::available_credit(const abstract_downstream& out) {
return out.min_credit();
void broadcast::push(abstract_downstream& out, long* hint) {
out.broadcast(hint);
}
} // namespace policy
......
......@@ -27,31 +27,22 @@
namespace caf {
namespace policy {
greedy::greedy() : low_watermark(0), high_watermark(5), min_buffer_size(5) {
greedy::greedy() : low_watermark(0), high_watermark(5) {
// nop
}
void greedy::assign_credit(assignment_vec& xs, size_t buf_size,
size_t downstream_credit) {
CAF_LOG_TRACE(CAF_ARG(xs) << CAF_ARG(buf_size) << CAF_ARG(downstream_credit));
/// Calculate how much credit we can hand out and how much credit we have
/// already assigned. Buffered elements are counted as assigned credit,
/// because we "release" credit only after pushing elements downstram.
size_t max_available = downstream_credit + min_buffer_size;
size_t assigned =
buf_size + std::accumulate(xs.begin(), xs.end(), size_t{0},
[](size_t x, const assignment_pair& y) {
return x + y.first->assigned_credit;
});
if (assigned >= max_available) {
// zero out assignment vector
void greedy::assign_credit(assignment_vec& xs,
long total_downstream_net_credit) {
CAF_LOG_TRACE(CAF_ARG(xs) << CAF_ARG(total_downstream_net_credit));
// Zero-out assignment vector if no credit is available at downstream paths.
if (total_downstream_net_credit <= 0) {
for (auto& x : xs)
x.second = 0;
return;
}
// Assign credit to upstream paths until no more credit is available. We must
// make sure to write to each element in the vector.
auto available = max_available - assigned;
auto available = total_downstream_net_credit;
for (auto& p : xs) {
auto& x = p.first->assigned_credit;
if (x < high_watermark) {
......
......@@ -37,30 +37,29 @@ error stream_handler::add_downstream(strong_actor_ptr&) {
}
error stream_handler::confirm_downstream(const strong_actor_ptr&,
strong_actor_ptr&, size_t, bool) {
strong_actor_ptr&, long, bool) {
CAF_LOG_ERROR("Cannot add downstream to a stream marked as no-downstreams");
return sec::cannot_add_downstream;
}
error stream_handler::downstream_demand(strong_actor_ptr&, size_t) {
error stream_handler::downstream_demand(strong_actor_ptr&, long) {
CAF_LOG_ERROR("Received downstream messages in "
"a stream marked as no-downstreams");
return sec::invalid_downstream;
}
error stream_handler::push(size_t*) {
error stream_handler::push(long*) {
CAF_LOG_ERROR("Cannot push to a stream marked as no-downstreams");
return sec::invalid_downstream;
}
expected<size_t> stream_handler::add_upstream(strong_actor_ptr&,
const stream_id&,
stream_priority) {
expected<long> stream_handler::add_upstream(strong_actor_ptr&, const stream_id&,
stream_priority) {
CAF_LOG_ERROR("Cannot add upstream to a stream marked as no-upstreams");
return sec::cannot_add_upstream;
}
error stream_handler::upstream_batch(strong_actor_ptr&, size_t, message&) {
error stream_handler::upstream_batch(strong_actor_ptr&, long, message&) {
CAF_LOG_ERROR("Received upstream messages in "
"a stream marked as no-upstreams");
return sec::invalid_upstream;
......
......@@ -106,7 +106,7 @@ auto stream_msg_visitor::operator()(stream_msg::abort& x) -> result_type {
auto stream_msg_visitor::operator()(stream_msg::ack_open& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
if (i_ != e_) {
auto d = static_cast<size_t>(x.initial_demand);
auto d = static_cast<long>(x.initial_demand);
return {i_->second->confirm_downstream(x.rebind_from,
self_->current_sender(), d, false),
i_};
......@@ -119,7 +119,7 @@ auto stream_msg_visitor::operator()(stream_msg::batch& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
if (i_ != e_)
return {i_->second->upstream_batch(self_->current_sender(),
static_cast<size_t>(x.xs_size), x.xs),
static_cast<long>(x.xs_size), x.xs),
i_};
CAF_LOG_DEBUG("received stream_msg::batch for unknown stream");
return {sec::unexpected_message, e_};
......@@ -129,7 +129,7 @@ auto stream_msg_visitor::operator()(stream_msg::ack_batch& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
if (i_ != e_)
return {i_->second->downstream_demand(self_->current_sender(),
static_cast<size_t>(x.new_capacity)),
static_cast<long>(x.new_capacity)),
i_};
CAF_LOG_DEBUG("received stream_msg::batch for unknown stream");
return {sec::unexpected_message, e_};
......
......@@ -31,7 +31,8 @@ stream_sink::stream_sink(abstract_upstream* in_ptr,
: in_ptr_(in_ptr),
original_sender_(std::move(orig_sender)),
next_stages_(std::move(trailing_stages)),
original_msg_id_(mid) {
original_msg_id_(mid),
min_buffer_size_(5) { // TODO: make configurable
// nop
}
......@@ -41,7 +42,7 @@ bool stream_sink::done() const {
}
error stream_sink::upstream_batch(strong_actor_ptr& hdl, size_t xs_size,
error stream_sink::upstream_batch(strong_actor_ptr& hdl, long xs_size,
message& xs) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(xs_size) << CAF_ARG(xs));
auto path = in().find(hdl);
......@@ -51,7 +52,7 @@ error stream_sink::upstream_batch(strong_actor_ptr& hdl, size_t xs_size,
path->assigned_credit -= xs_size;
auto err = consume(xs);
if (err == none)
in().assign_credit(0, 0);
in().assign_credit(min_buffer_size());
return err;
}
return sec::invalid_upstream;
......
......@@ -40,7 +40,7 @@ bool stream_source::done() const {
return out_ptr_->closed();
}
error stream_source::downstream_demand(strong_actor_ptr& hdl, size_t value) {
error stream_source::downstream_demand(strong_actor_ptr& hdl, long value) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(value));
auto path = out_ptr_->find(hdl);
if (path) {
......@@ -48,7 +48,7 @@ error stream_source::downstream_demand(strong_actor_ptr& hdl, size_t value) {
if (!at_end()) {
// produce new elements
auto current_size = buf_size();
auto size_hint = out().available_credit();
auto size_hint = out().total_net_credit();
if (current_size < size_hint)
generate(size_hint - current_size);
return push(&size_hint);
......
......@@ -39,7 +39,7 @@ bool stream_stage::done() const {
return in_ptr_->closed() && out_ptr_->closed();
}
error stream_stage::upstream_batch(strong_actor_ptr& hdl, size_t xs_size,
error stream_stage::upstream_batch(strong_actor_ptr& hdl, long xs_size,
message& xs) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(xs_size) << CAF_ARG(xs));
auto path = in().find(hdl);
......@@ -50,14 +50,14 @@ error stream_stage::upstream_batch(strong_actor_ptr& hdl, size_t xs_size,
auto err = process_batch(xs);
if (err == none) {
push();
in().assign_credit(out().buf_size(), out().available_credit());
in().assign_credit(out().total_net_credit());
}
return err;
}
return sec::invalid_upstream;
}
error stream_stage::downstream_demand(strong_actor_ptr& hdl, size_t value) {
error stream_stage::downstream_demand(strong_actor_ptr& hdl, long value) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(value));
auto path = out_ptr_->find(hdl);
if (path) {
......@@ -66,7 +66,7 @@ error stream_stage::downstream_demand(strong_actor_ptr& hdl, size_t value) {
push();
else if (in().closed() && !out().remove_path(hdl))
return sec::invalid_downstream;
in().assign_credit(out().buf_size(), out().available_credit());
in().assign_credit(out().total_net_credit());
return none;
}
return sec::invalid_downstream;
......
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