Commit f5e61e27 authored by Dominik Charousset's avatar Dominik Charousset

Track batch IDs properly, improve upstream policy

parent b7e22236
...@@ -66,11 +66,12 @@ public: ...@@ -66,11 +66,12 @@ public:
/// Add a new upstream actor to the stream and return an initial credit. /// Add a new upstream actor to the stream and return an initial credit.
virtual expected<long> add_upstream(strong_actor_ptr& hdl, virtual expected<long> add_upstream(strong_actor_ptr& hdl,
const stream_id& sid, const stream_id& sid,
stream_priority prio); stream_priority prio);
/// Handles data from an upstream actor. /// Handles data from an upstream actor.
virtual error upstream_batch(strong_actor_ptr& hdl, long, message& xs); virtual error upstream_batch(strong_actor_ptr& hdl, int64_t xs_id,
long xs_size, message& xs);
/// Closes an upstream. /// Closes an upstream.
virtual error close_upstream(strong_actor_ptr& hdl); virtual error close_upstream(strong_actor_ptr& hdl);
......
...@@ -41,8 +41,8 @@ public: ...@@ -41,8 +41,8 @@ public:
bool done() const override; bool done() const override;
error upstream_batch(strong_actor_ptr& src, long xs_size, error upstream_batch(strong_actor_ptr& src, int64_t xs_id,
message& xs) override; long xs_size, message& xs) override;
void abort(strong_actor_ptr& cause, const error& reason) override; void abort(strong_actor_ptr& cause, const error& reason) override;
......
...@@ -45,7 +45,7 @@ public: ...@@ -45,7 +45,7 @@ public:
error downstream_demand(strong_actor_ptr&, long) override; error downstream_demand(strong_actor_ptr&, long) override;
error upstream_batch(strong_actor_ptr&, long, message&) override; error upstream_batch(strong_actor_ptr&, int64_t, long, message&) override;
void last_upstream_closed(); void last_upstream_closed();
......
...@@ -137,6 +137,9 @@ protected: ...@@ -137,6 +137,9 @@ protected:
/// Assigns new credit to upstream actors by filling `assignment_vec_`. /// Assigns new credit to upstream actors by filling `assignment_vec_`.
virtual void fill_assignment_vec(long downstream_credit) = 0; virtual void fill_assignment_vec(long downstream_credit) = 0;
/// Creates initial credit for `ptr`.
virtual long initial_credit(upstream_path* ptr, long downstream_credit);
/// Pointer to the parent actor. /// Pointer to the parent actor.
local_actor* self_; local_actor* self_;
......
...@@ -59,7 +59,8 @@ expected<long> stream_handler::add_upstream(strong_actor_ptr&, const stream_id&, ...@@ -59,7 +59,8 @@ expected<long> stream_handler::add_upstream(strong_actor_ptr&, const stream_id&,
return sec::cannot_add_upstream; return sec::cannot_add_upstream;
} }
error stream_handler::upstream_batch(strong_actor_ptr&, long, message&) { error stream_handler::upstream_batch(strong_actor_ptr&, int64_t, long,
message&) {
CAF_LOG_ERROR("Received upstream messages in " CAF_LOG_ERROR("Received upstream messages in "
"a stream marked as no-upstreams"); "a stream marked as no-upstreams");
return sec::invalid_upstream; return sec::invalid_upstream;
......
...@@ -118,7 +118,7 @@ auto stream_msg_visitor::operator()(stream_msg::ack_open& x) -> result_type { ...@@ -118,7 +118,7 @@ auto stream_msg_visitor::operator()(stream_msg::ack_open& x) -> result_type {
auto stream_msg_visitor::operator()(stream_msg::batch& x) -> result_type { auto stream_msg_visitor::operator()(stream_msg::batch& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
if (i_ != e_) if (i_ != e_)
return {i_->second->upstream_batch(self_->current_sender(), return {i_->second->upstream_batch(self_->current_sender(), x.id,
static_cast<long>(x.xs_size), x.xs), static_cast<long>(x.xs_size), x.xs),
i_}; i_};
CAF_LOG_DEBUG("received stream_msg::batch for unknown stream"); CAF_LOG_DEBUG("received stream_msg::batch for unknown stream");
......
...@@ -42,14 +42,14 @@ bool stream_sink::done() const { ...@@ -42,14 +42,14 @@ bool stream_sink::done() const {
return in_ptr_->closed(); return in_ptr_->closed();
} }
error stream_sink::upstream_batch(strong_actor_ptr& hdl, int64_t xs_id,
error stream_sink::upstream_batch(strong_actor_ptr& hdl, long xs_size, long xs_size, message& xs) {
message& xs) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(xs_size) << CAF_ARG(xs)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(xs_size) << CAF_ARG(xs));
auto path = in().find(hdl); auto path = in().find(hdl);
if (path) { if (path) {
if (xs_size > path->assigned_credit) if (xs_size > path->assigned_credit)
return sec::invalid_stream_state; return sec::invalid_stream_state;
path->last_batch_id = xs_id;
path->assigned_credit -= xs_size; path->assigned_credit -= xs_size;
auto err = consume(xs); auto err = consume(xs);
if (err == none) if (err == none)
......
...@@ -47,12 +47,25 @@ error stream_source::downstream_demand(strong_actor_ptr& hdl, long value) { ...@@ -47,12 +47,25 @@ error stream_source::downstream_demand(strong_actor_ptr& hdl, long value) {
path->open_credit += value; path->open_credit += value;
if (!at_end()) { if (!at_end()) {
// produce new elements // produce new elements
auto current_size = buf_size(); auto capacity = out().credit();
auto size_hint = std::min(out().credit(), // TODO: fix issue where a source does not emit the last pieces if
out().max_batch_size()); // min_batch_size cannot be reached
if (current_size < size_hint) while (capacity > out().min_batch_size()) {
generate(static_cast<size_t>(size_hint - current_size)); auto current_size = buf_size();
return push(); auto size_hint = std::min(capacity, out().max_batch_size());
if (size_hint > current_size)
generate(static_cast<size_t>(size_hint - current_size));
auto err = push();
if (err)
return err;
// Check if `push` did actually use credit, otherwise break loop
auto new_capacity = out().credit();
if (new_capacity != capacity)
capacity = new_capacity;
else
break;
}
return none;
} }
// transmit cached elements before closing paths // transmit cached elements before closing paths
if (buf_size() > 0) if (buf_size() > 0)
......
...@@ -37,13 +37,14 @@ bool stream_stage::done() const { ...@@ -37,13 +37,14 @@ bool stream_stage::done() const {
return in_ptr_->closed() && out_ptr_->closed(); return in_ptr_->closed() && out_ptr_->closed();
} }
error stream_stage::upstream_batch(strong_actor_ptr& hdl, long xs_size, error stream_stage::upstream_batch(strong_actor_ptr& hdl, int64_t xs_id,
message& xs) { long xs_size, message& xs) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(xs_size) << CAF_ARG(xs)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(xs_size) << CAF_ARG(xs));
auto path = in().find(hdl); auto path = in().find(hdl);
if (path) { if (path) {
if (xs_size > path->assigned_credit) if (xs_size > path->assigned_credit)
return sec::invalid_stream_state; return sec::invalid_stream_state;
path->last_batch_id = xs_id;
path->assigned_credit -= xs_size; path->assigned_credit -= xs_size;
auto err = process_batch(xs); auto err = process_batch(xs);
if (err == none) { if (err == none) {
......
...@@ -55,20 +55,19 @@ void upstream_policy::assign_credit(long downstream_capacity) { ...@@ -55,20 +55,19 @@ void upstream_policy::assign_credit(long downstream_capacity) {
for (auto& x : assignment_vec_) for (auto& x : assignment_vec_)
used_capacity += static_cast<long>(x.first->assigned_credit); used_capacity += static_cast<long>(x.first->assigned_credit);
CAF_LOG_DEBUG(CAF_ARG(used_capacity)); CAF_LOG_DEBUG(CAF_ARG(used_capacity));
if (used_capacity >= downstream_capacity)
return;
fill_assignment_vec(downstream_capacity - used_capacity); fill_assignment_vec(downstream_capacity - used_capacity);
for (auto& x : assignment_vec_) { for (auto& x : assignment_vec_) {
auto n = x.second; auto n = x.second;
if (n > 0) { if (n > 0) {
auto ptr = x.first; auto ptr = x.first;
ptr->assigned_credit += n; ptr->assigned_credit += n;
ptr->last_acked_batch_id = ptr->last_batch_id;
CAF_LOG_DEBUG("ack batch" << ptr->last_batch_id CAF_LOG_DEBUG("ack batch" << ptr->last_batch_id
<< "with" << n << "new capacity"); << "with" << n << "new capacity");
unsafe_send_as(self_, ptr->hdl, unsafe_send_as(self_, ptr->hdl,
make<stream_msg::ack_batch>(ptr->sid, make<stream_msg::ack_batch>(ptr->sid,
static_cast<int32_t>(n), static_cast<int32_t>(n),
ptr->last_batch_id++)); ptr->last_batch_id));
} }
} }
} }
...@@ -85,7 +84,7 @@ expected<long> upstream_policy::add_path(strong_actor_ptr hdl, ...@@ -85,7 +84,7 @@ expected<long> upstream_policy::add_path(strong_actor_ptr hdl,
paths_.emplace_back(ptr); paths_.emplace_back(ptr);
assignment_vec_.emplace_back(ptr, 0); assignment_vec_.emplace_back(ptr, 0);
if (downstream_credit > 0) if (downstream_credit > 0)
ptr->assigned_credit = std::min(max_credit_, downstream_credit); ptr->assigned_credit = initial_credit(ptr, downstream_credit);
CAF_LOG_DEBUG("add new upstraem path" << ptr->hdl CAF_LOG_DEBUG("add new upstraem path" << ptr->hdl
<< "with initial credit" << ptr->assigned_credit); << "with initial credit" << ptr->assigned_credit);
return ptr->assigned_credit; return ptr->assigned_credit;
...@@ -133,4 +132,9 @@ upstream_path* upstream_policy::find(const strong_actor_ptr& x) const { ...@@ -133,4 +132,9 @@ upstream_path* upstream_policy::find(const strong_actor_ptr& x) const {
return i != e ? i->get() : nullptr; return i != e ? i->get() : nullptr;
} }
long upstream_policy::initial_credit(upstream_path* ptr,
long downstream_credit) {
return std::min(max_credit_, downstream_credit);
}
} // namespace caf } // namespace caf
...@@ -151,7 +151,8 @@ public: ...@@ -151,7 +151,8 @@ public:
expected<long> add_upstream(strong_actor_ptr& hdl, const stream_id& sid, expected<long> add_upstream(strong_actor_ptr& hdl, const stream_id& sid,
stream_priority prio) override; stream_priority prio) override;
error upstream_batch(strong_actor_ptr& hdl, long, message& xs) override; error upstream_batch(strong_actor_ptr& hdl, int64_t, long,
message& xs) override;
error close_upstream(strong_actor_ptr& hdl) override; error close_upstream(strong_actor_ptr& hdl) override;
...@@ -324,8 +325,8 @@ expected<long> stream_governor::add_upstream(strong_actor_ptr& hdl, ...@@ -324,8 +325,8 @@ expected<long> stream_governor::add_upstream(strong_actor_ptr& hdl,
return sec::invalid_argument; return sec::invalid_argument;
} }
error stream_governor::upstream_batch(strong_actor_ptr& hdl, long xs_size, error stream_governor::upstream_batch(strong_actor_ptr& hdl, int64_t xs_id,
message& xs) { long xs_size, message& xs) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(xs_size) << CAF_ARG(xs)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(xs_size) << CAF_ARG(xs));
using std::get; using std::get;
// Sanity checking. // Sanity checking.
...@@ -341,6 +342,7 @@ error stream_governor::upstream_batch(strong_actor_ptr& hdl, long xs_size, ...@@ -341,6 +342,7 @@ error stream_governor::upstream_batch(strong_actor_ptr& hdl, long xs_size,
// Decrease credit assigned to `hdl` and get currently available downstream // Decrease credit assigned to `hdl` and get currently available downstream
// credit on all paths. // credit on all paths.
CAF_LOG_DEBUG(CAF_ARG(path->assigned_credit)); CAF_LOG_DEBUG(CAF_ARG(path->assigned_credit));
path->last_batch_id = xs_id;
path->assigned_credit -= xs_size; path->assigned_credit -= xs_size;
// Forward data to all other peers. // Forward data to all other peers.
for (auto& kvp : peers_) for (auto& kvp : peers_)
......
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