Commit 8562c1af authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/1076'

parents 08f7a298 538ee173
...@@ -1170,9 +1170,15 @@ scheduled_actor::advance_streams(actor_clock::time_point now) { ...@@ -1170,9 +1170,15 @@ scheduled_actor::advance_streams(actor_clock::time_point now) {
auto bitmask = stream_ticks_.timeouts(now, {max_batch_delay_ticks_, auto bitmask = stream_ticks_.timeouts(now, {max_batch_delay_ticks_,
credit_round_ticks_}); credit_round_ticks_});
// Force batches on all output paths. // Force batches on all output paths.
if ((bitmask & 0x01) != 0) { if ((bitmask & 0x01) != 0 && !stream_managers_.empty()) {
std::vector<stream_manager*> managers;
managers.reserve(stream_managers_.size());
for (auto& kvp : stream_managers_) for (auto& kvp : stream_managers_)
kvp.second->out().force_emit_batches(); managers.emplace_back(kvp.second.get());
std::sort(managers.begin(), managers.end());
auto e = std::unique(managers.begin(), managers.end());
for (auto i = managers.begin(); i != e; ++i)
(*i)->out().force_emit_batches();
} }
// Fill up credit on each input path. // Fill up credit on each input path.
if ((bitmask & 0x02) != 0) { if ((bitmask & 0x02) != 0) {
......
...@@ -114,7 +114,8 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) { ...@@ -114,7 +114,8 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) {
void stream_manager::handle(stream_slots slots, upstream_msg::drop&) { void stream_manager::handle(stream_slots slots, upstream_msg::drop&) {
CAF_LOG_TRACE(CAF_ARG(slots)); CAF_LOG_TRACE(CAF_ARG(slots));
out().close(slots.receiver); error reason;
out().remove_path(slots.receiver, reason, false);
} }
void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) { void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) {
......
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