Unverified Commit c31c0988 authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #952

Fix excessive buffering in stream stages
parents 16730fc6 efbe5827
...@@ -144,10 +144,10 @@ public: ...@@ -144,10 +144,10 @@ public:
/// @param self Points to the parent actor, i.e., sender of the message. /// @param self Points to the parent actor, i.e., sender of the message.
/// @param queued_items Accumulated size of all batches that are currently /// @param queued_items Accumulated size of all batches that are currently
/// waiting in the mailbox. /// waiting in the mailbox.
/// @param now Current timestamp.
/// @param cycle Time between credit rounds. /// @param cycle Time between credit rounds.
/// @param desired_batch_complexity Desired processing time per batch. /// @param desired_batch_complexity Desired processing time per batch.
void emit_ack_batch(local_actor* self, int32_t queued_items, void emit_ack_batch(local_actor* self, int32_t queued_items,
int32_t max_downstream_capacity,
actor_clock::time_point now, timespan cycle, actor_clock::time_point now, timespan cycle,
timespan desired_batch_complexity); timespan desired_batch_complexity);
......
...@@ -133,12 +133,10 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) { ...@@ -133,12 +133,10 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
} }
void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items, void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
int32_t max_downstream_capacity,
actor_clock::time_point now, timespan cycle, actor_clock::time_point now, timespan cycle,
timespan complexity) { timespan complexity) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(queued_items) CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(queued_items) << CAF_ARG(cycle)
<< CAF_ARG(max_downstream_capacity) << CAF_ARG(cycle) << CAF_ARG(complexity));
<< CAF_ARG(complexity));
CAF_IGNORE_UNUSED(queued_items); CAF_IGNORE_UNUSED(queued_items);
// Update timestamps. // Update timestamps.
last_credit_decision = now; last_credit_decision = now;
...@@ -147,22 +145,28 @@ void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items, ...@@ -147,22 +145,28 @@ void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
// the downstream capacity. // the downstream capacity.
auto x = stats.calculate(cycle, complexity); auto x = stats.calculate(cycle, complexity);
auto stats_guard = detail::make_scope_guard([&] { stats.reset(); }); auto stats_guard = detail::make_scope_guard([&] { stats.reset(); });
auto max_capacity = std::min(x.max_throughput * 2, max_downstream_capacity); auto& out = mgr->out();
auto max_capacity = std::min(x.max_throughput * 2, out.max_capacity());
CAF_ASSERT(max_capacity > 0); CAF_ASSERT(max_capacity > 0);
// Protect against overflow on `assigned_credit`. // Protect against overflow on `assigned_credit`.
auto max_new_credit = std::numeric_limits<int32_t>::max() - assigned_credit; auto max_new_credit = std::numeric_limits<int32_t>::max() - assigned_credit;
// Compute the amount of credit we grant in this round. // Compute the amount of credit we grant in this round.
auto credit = std::min(std::max(max_capacity - assigned_credit, 0), auto credit = std::min(std::max(max_capacity
- static_cast<int32_t>(out.buffered())
- assigned_credit,
0),
max_new_credit); max_new_credit);
CAF_ASSERT(credit >= 0); CAF_ASSERT(credit >= 0);
// The manager can restrict or adjust the amount of credit. // The manager can restrict or adjust the amount of credit.
credit = std::min(mgr->acquire_credit(this, credit), max_new_credit); credit = std::min(mgr->acquire_credit(this, credit), max_new_credit);
CAF_STREAM_LOG_DEBUG(mgr->self()->name() << "grants" << credit CAF_STREAM_LOG_DEBUG(mgr->self()->name()
<< "new credit at slot" << slots.receiver << "grants" << credit << "new credit at slot"
<< "after receiving" << stats.num_elements << slots.receiver << "after receiving"
<< "elements that took" << stats.processing_time << stats.num_elements << "elements that took"
<< stats.processing_time
<< CAF_ARG2("max_throughput", x.max_throughput) << CAF_ARG2("max_throughput", x.max_throughput)
<< CAF_ARG(max_downstream_capacity) << CAF_ARG2("max_downstream_capacity",
out.max_capacity())
<< CAF_ARG(assigned_credit)); << CAF_ARG(assigned_credit));
if (credit == 0 && up_to_date()) if (credit == 0 && up_to_date())
return; return;
......
...@@ -1156,8 +1156,7 @@ scheduled_actor::advance_streams(actor_clock::time_point now) { ...@@ -1156,8 +1156,7 @@ scheduled_actor::advance_streams(actor_clock::time_point now) {
for (auto& kvp : qs) { for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get(); auto inptr = kvp.second.policy().handler.get();
auto bs = static_cast<int32_t>(kvp.second.total_task_size()); auto bs = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(this, bs, inptr->mgr->out().max_capacity(), inptr->emit_ack_batch(this, bs, now, cycle, bc);
now, cycle, bc);
} }
} }
return stream_ticks_.next_timeout(now, {max_batch_delay_ticks_, return stream_ticks_.next_timeout(now, {max_batch_delay_ticks_,
......
...@@ -161,8 +161,7 @@ void stream_manager::advance() { ...@@ -161,8 +161,7 @@ void stream_manager::advance() {
// Ignore inbound paths of other managers. // Ignore inbound paths of other managers.
if (inptr->mgr.get() == this) { if (inptr->mgr.get() == this) {
auto bs = static_cast<int32_t>(kvp.second.total_task_size()); auto bs = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(self_, bs, out().max_capacity(), now, interval, inptr->emit_ack_batch(self_, bs, now, interval, bc);
bc);
} }
} }
} }
......
...@@ -342,7 +342,7 @@ public: ...@@ -342,7 +342,7 @@ public:
for (auto& kvp : qs) { for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get(); auto inptr = kvp.second.policy().handler.get();
auto bs = static_cast<int32_t>(kvp.second.total_task_size()); auto bs = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(this, bs, 30, now(), cycle, inptr->emit_ack_batch(this, bs, now(), cycle,
desired_batch_complexity); desired_batch_complexity);
} }
} }
......
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