Commit 5de3e082 authored by Dominik Charousset's avatar Dominik Charousset Committed by Marian Triebe

Add scheduled_actor::trigger_downstreams

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