Commit ae96552a authored by Dominik Charousset's avatar Dominik Charousset

Add scheduled_actor::trigger_downstreams

parent c5a3ee26
......@@ -45,11 +45,10 @@ public:
return sec::downstream_already_exists;
}
error trigger_send(strong_actor_ptr&) {
error push() final {
if (out().buf_size() > 0)
out().policy().push(out());
return none;
}
private:
......
......@@ -447,6 +447,10 @@ public:
return streams_;
}
/// Tries to send more data on all downstream paths. Use this function to
/// manually trigger batches in a source after receiving more data to send.
void trigger_downstreams();
/// @cond PRIVATE
// -- timeout management -----------------------------------------------------
......
......@@ -258,7 +258,7 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) {
return resumable::resume_later;
}
// -- scheduler callbacks ----------------------------------------------------
// -- scheduler callbacks ------------------------------------------------------
proxy_registry* scheduled_actor::proxy_registry_ptr() {
return nullptr;
......@@ -272,6 +272,13 @@ void scheduled_actor::quit(error x) {
setf(is_terminated_flag);
}
// -- stream management --------------------------------------------------------
void scheduled_actor::trigger_downstreams() {
for (auto& s : streams_)
s.second->push();
}
// -- timeout management -------------------------------------------------------
uint32_t scheduled_actor::request_timeout(const duration& d) {
......
......@@ -51,11 +51,11 @@ error stream_source::downstream_demand(strong_actor_ptr& hdl, size_t value) {
auto size_hint = out_ptr_->policy().desired_buffer_size(*out_ptr_);
if (current_size < size_hint)
generate(size_hint - current_size);
return trigger_send(hdl);
return push();
}
// transmit cached elements before closing paths
if (buf_size() > 0)
return trigger_send(hdl);
return push();
if (out_ptr_->remove_path(hdl))
return none;
}
......
......@@ -44,8 +44,7 @@ error stream_stage::upstream_batch(strong_actor_ptr& hdl, size_t xs_size,
auto err = in_ptr_->pull(hdl, xs_size);
if (!err) {
process_batch(xs);
strong_actor_ptr dummy;
trigger_send(dummy);
push();
}
return err;
}
......@@ -55,8 +54,9 @@ error stream_stage::downstream_demand(strong_actor_ptr& hdl, size_t value) {
if (path) {
path->open_credit += value;
if(out_ptr_->buf_size() > 0) {
return trigger_send(hdl);
} if (in_ptr_->closed()) {
return push();
}
if (in_ptr_->closed()) {
if (!out_ptr_->remove_path(hdl)) {
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