Commit 15a0b49f authored by Dominik Charousset's avatar Dominik Charousset

Fix ordering issue in message_queue::push

parent 37d0d5bc
...@@ -40,24 +40,20 @@ void message_queue::push(execution_unit* ctx, uint64_t id, ...@@ -40,24 +40,20 @@ void message_queue::push(execution_unit* ctx, uint64_t id,
// Dispatch current head. // Dispatch current head.
if (receiver != nullptr) if (receiver != nullptr)
receiver->enqueue(std::move(content), ctx); receiver->enqueue(std::move(content), ctx);
auto next = id + 1;
// Check whether we can deliver more. // Check whether we can deliver more.
if (first == last || first->id != next_undelivered + 1) { if (first == last || first->id != next) {
++next_undelivered; next_undelivered = next;
CAF_ASSERT(next_undelivered <= next_id); CAF_ASSERT(next_undelivered <= next_id);
return; return;
} }
// We look for the last element that we cannot ship right away. // Deliver everything until reaching a non-consecutive ID or the end.
auto pred = [](const actor_msg& x, const actor_msg& y) { auto i = first;
return x.id + 1 > y.id; for (; i != last && i->id == next; ++i, ++next)
};
auto last_hit = std::adjacent_find(first, last, pred);
CAF_ASSERT(last_hit != first);
auto new_last = last_hit == last ? last : last_hit + 1;
for (auto i = first; i != new_last; ++i)
if (i->receiver != nullptr) if (i->receiver != nullptr)
i->receiver->enqueue(std::move(i->content), ctx); i->receiver->enqueue(std::move(i->content), ctx);
next_undelivered += static_cast<size_t>(std::distance(first, new_last)) + 1; next_undelivered = next;
pending.erase(first, new_last); pending.erase(first, i);
CAF_ASSERT(next_undelivered <= next_id); CAF_ASSERT(next_undelivered <= next_id);
return; return;
} }
......
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