Commit d6e09d4b authored by Dominik Charousset's avatar Dominik Charousset

Allow stream handlers to keep track of batch IDs

parent ec7e1e41
......@@ -48,7 +48,12 @@ public:
<< CAF_ARG(redeployable));
CAF_ASSERT(ptr != nullptr);
if (out().confirm_path(rebind_from, ptr, redeployable)) {
dptr()->downstream_demand(ptr, initial_demand);
auto path = out().find(ptr);
if (!path) {
CAF_LOG_ERROR("Unable to find path after confirming it");
return sec::invalid_downstream;
}
downstream_demand(path, initial_demand);
return none;
}
return sec::invalid_downstream;
......@@ -60,6 +65,9 @@ public:
return none;
}
protected:
virtual void downstream_demand(downstream_path* ptr, long demand) = 0;
private:
Subtype* dptr() {
return static_cast<Subtype*>(this);
......
......@@ -52,10 +52,11 @@ public:
strong_actor_ptr& hdl, long initial_demand,
bool redeployable);
/// Handles new demand from a downstream actor.
/// Handles cumulative ACKs with new demand from a downstream actor.
/// @pre `hdl != nullptr`
/// @pre `new_demand > 0`
virtual error downstream_demand(strong_actor_ptr& hdl, long new_demand);
virtual error downstream_ack(strong_actor_ptr& hdl, int64_t batch_id,
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
......
......@@ -39,7 +39,7 @@ public:
bool done() const override;
error downstream_demand(strong_actor_ptr& hdl, long value) override;
error downstream_ack(strong_actor_ptr& hdl, int64_t bid, long value) override;
void abort(strong_actor_ptr& cause, const error& reason) override;
......@@ -51,6 +51,8 @@ public:
void generate();
protected:
void downstream_demand(downstream_path* path, long demand) override;
/// Queries the current amount of elements in the output buffer.
virtual long buf_size() const = 0;
......
......@@ -43,7 +43,7 @@ public:
void abort(strong_actor_ptr&, const error&) override;
error downstream_demand(strong_actor_ptr&, long) override;
error downstream_ack(strong_actor_ptr&, int64_t, long) override;
error upstream_batch(strong_actor_ptr&, int64_t, long, message&) override;
......@@ -58,6 +58,8 @@ public:
}
protected:
void downstream_demand(downstream_path*, long) override;
virtual error process_batch(message& msg) = 0;
private:
......
......@@ -42,7 +42,7 @@ error stream_handler::confirm_downstream(const strong_actor_ptr&,
return sec::cannot_add_downstream;
}
error stream_handler::downstream_demand(strong_actor_ptr&, long) {
error stream_handler::downstream_ack(strong_actor_ptr&, int64_t, long) {
CAF_LOG_ERROR("Received downstream messages in "
"a stream marked as no-downstreams");
return sec::invalid_downstream;
......
......@@ -128,7 +128,8 @@ auto stream_msg_visitor::operator()(stream_msg::batch& x) -> result_type {
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(),
return {i_->second->downstream_ack(self_->current_sender(),
x.acknowledged_id,
static_cast<long>(x.new_capacity)),
i_};
CAF_LOG_DEBUG("received stream_msg::batch for unknown stream");
......
......@@ -40,11 +40,12 @@ bool stream_source::done() const {
return out_ptr_->closed();
}
error stream_source::downstream_demand(strong_actor_ptr& hdl, long value) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(value));
error stream_source::downstream_ack(strong_actor_ptr& hdl, int64_t,
long demand) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(demand));
auto path = out_ptr_->find(hdl);
if (path) {
path->open_credit += value;
path->open_credit += demand;
if (!at_end()) {
// produce new elements
auto capacity = out().credit();
......@@ -90,4 +91,35 @@ void stream_source::generate() {
}
}
void stream_source::downstream_demand(downstream_path* path, long demand) {
CAF_ASSERT(path != nullptr);
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(demand));
path->open_credit += demand;
if (!at_end()) {
// produce new elements
auto capacity = out().credit();
// TODO: fix issue where a source does not emit the last pieces if
// min_batch_size cannot be reached
while (capacity > out().min_batch_size()) {
auto current_size = buf_size();
auto size_hint = std::min(capacity, out().max_batch_size());
if (size_hint > current_size)
generate(static_cast<size_t>(size_hint - current_size));
push();
// 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;
}
// transmit cached elements before closing paths
if (buf_size() > 0)
push();
auto hdl = path->hdl;
out_ptr_->remove_path(hdl);
}
} // namespace caf
......@@ -59,19 +59,12 @@ error stream_stage::upstream_batch(strong_actor_ptr& hdl, int64_t xs_id,
return sec::invalid_upstream;
}
error stream_stage::downstream_demand(strong_actor_ptr& hdl, long value) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(value));
error stream_stage::downstream_ack(strong_actor_ptr& hdl, int64_t,
long demand) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(demand));
auto path = out_ptr_->find(hdl);
if (path) {
path->open_credit += value;
if(out().buf_size() > 0)
push();
else if (in().closed() && !out().remove_path(hdl))
return sec::invalid_downstream;
auto current_size = out().buf_size();
auto desired_size = out().credit();
if (current_size < desired_size)
in().assign_credit(desired_size - current_size);
downstream_demand(path, demand);
return none;
}
return sec::invalid_downstream;
......@@ -87,4 +80,17 @@ void stream_stage::last_upstream_closed() {
out().close();
}
void stream_stage::downstream_demand(downstream_path* path, long demand) {
auto hdl = path->hdl;
path->open_credit += demand;
if(out().buf_size() > 0)
push();
else if (in().closed())
out().remove_path(hdl); // don't pass path->hdl: path can become invalid
auto current_size = out().buf_size();
auto desired_size = out().credit();
if (current_size < desired_size)
in().assign_credit(desired_size - current_size);
}
} // namespace caf
......@@ -132,7 +132,7 @@ upstream_path* upstream_policy::find(const strong_actor_ptr& x) const {
return i != e ? i->get() : nullptr;
}
long upstream_policy::initial_credit(upstream_path* ptr,
long upstream_policy::initial_credit(upstream_path*,
long downstream_credit) {
return std::min(max_credit_, downstream_credit);
}
......
......@@ -140,11 +140,13 @@ public:
error add_downstream(strong_actor_ptr& hdl) override;
void downstream_demand(downstream_path* path, long demand);
error confirm_downstream(const strong_actor_ptr& rebind_from,
strong_actor_ptr& hdl, long initial_demand,
bool redeployable) override;
error downstream_demand(strong_actor_ptr& hdl, long new_demand) override;
error downstream_ack(strong_actor_ptr& hdl, int64_t, long demand) override;
error push() override;
......@@ -248,6 +250,12 @@ error stream_governor::add_downstream(strong_actor_ptr&) {
return sec::invalid_stream_state;
}
void stream_governor::downstream_demand(downstream_path* path, long demand) {
path->open_credit += demand;
push();
assign_credit();
}
error stream_governor::confirm_downstream(const strong_actor_ptr& rebind_from,
strong_actor_ptr& hdl,
long initial_demand,
......@@ -261,7 +269,8 @@ error stream_governor::confirm_downstream(const strong_actor_ptr& rebind_from,
CAF_LOG_ERROR("Cannot rebind to registered downstream.");
return sec::invalid_stream_state;
}
return downstream_demand(hdl, initial_demand);
downstream_demand(path, initial_demand);
return none;
}
auto i = peers_.find(rebind_from);
if (i != peers_.end()) {
......@@ -275,19 +284,24 @@ error stream_governor::confirm_downstream(const strong_actor_ptr& rebind_from,
CAF_LOG_DEBUG("Confirmed path to another core"
<< CAF_ARG(rebind_from) << CAF_ARG(hdl));
res.first->second->out.confirm_path(rebind_from, hdl, initial_demand);
return downstream_demand(hdl, initial_demand);
auto pp = res.first->second->out.find(hdl);
if (!pp) {
CAF_LOG_ERROR("Unable to find peer after confirming it");
return sec::invalid_downstream;
}
downstream_demand(pp, initial_demand);
return none;
}
CAF_LOG_ERROR("Cannot confirm path to unknown downstream.");
return sec::invalid_downstream;
}
error stream_governor::downstream_demand(strong_actor_ptr& hdl, long value) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(value));
error stream_governor::downstream_ack(strong_actor_ptr& hdl, int64_t,
long demand) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(demand));
auto path = local_subscribers_.find(hdl);
if (path) {
path->open_credit += value;
push();
assign_credit();
downstream_demand(path, demand);
return none;
}
auto i = peers_.find(hdl);
......@@ -295,10 +309,8 @@ error stream_governor::downstream_demand(strong_actor_ptr& hdl, long value) {
auto pp = i->second->out.find(hdl);
if (!pp)
return sec::invalid_stream_state;
CAF_LOG_DEBUG("grant" << value << "new credit to" << hdl);
pp->open_credit += value;
push();
assign_credit();
CAF_LOG_DEBUG("grant" << demand << "new credit to" << hdl);
downstream_demand(pp, demand);
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