Commit 8e54d8e5 authored by Dominik Charousset's avatar Dominik Charousset

Consolidate take_front code in DRR queues

`drr_queue::take_front` and `drr_cached_queue::take_front` were
essentially the same function. Moreover, `drr_cached_queue::new_round`
duplicated the algorithm a third time (entangled with the actual logic
for a new credit round). The new `task_queue::next` funciton now
implements this algorithm once.

Moreover, `take_front` is only used directly from
`blocking_actor::dequeue`. This function bypasses the deficit counter on
the caller side. To clean up the awkward use of the queue, the new
`take_front` of the cached DRR queue bypasses the deficit counter
internally.

Restructuring the code seems to fix the original issue as well. After
debugging this extensively, it appears that GCC emitted code that didn't
properly track reads and writes to the head node. Reading `begin().ptr`
twice (via printf) causes the value to change between reads - even
though the printf statements where on consecutive lines.
parent dab5e804
......@@ -148,28 +148,23 @@ public:
list_.dec_total_task_size(std::forward<T>(x));
}
/// Takes the first element out of the queue (after flushing the cache) if
/// the deficit allows it and returns the element.
/// Takes the first element out of the queue if the deficit allows it and
/// returns the element.
/// @private
unique_pointer next() noexcept {
return list_.next(deficit_);
}
/// Takes the first element out of the queue (after flushing the cache) and
/// returns it, ignoring the deficit count.
unique_pointer take_front() noexcept {
flush_cache();
unique_pointer result;
if (!list_.empty()) {
auto ptr = list_.front();
auto ts = policy().task_size(*ptr);
CAF_ASSERT(ts > 0);
if (ts <= deficit_) {
deficit_ -= ts;
dec_total_task_size(ts);
list_.before_begin()->next = ptr->next;
if (total_task_size() == 0) {
CAF_ASSERT(list_.begin() == list_.end());
deficit_ = 0;
list_.end()->next = list_.before_begin().ptr;
}
result.reset(ptr);
}
// Don't modify the deficit counter.
auto dummy_deficit = std::numeric_limits<deficit_type>::max();
return list_.next(dummy_deficit);
}
return result;
return nullptr;
}
/// Consumes items from the queue until the queue is empty, there is not
......@@ -188,48 +183,39 @@ public:
return {false, false};
deficit_ += quantum;
long consumed = 0;
auto ptr = list_.front();
auto ts = policy().task_size(*ptr);
while (ts <= deficit_) {
CAF_ASSERT(ts > 0);
dec_total_task_size(ts);
auto next = ptr->next;
// Make sure the queue is in a consistent state before calling `f` in
// case `f` recursively calls consume.
list_.before_begin()->next = next;
if (next == list_.end().ptr)
list_.end()->next = list_.before_begin().ptr;
CAF_ASSERT(total_task_size() != 0 || list_.begin() == list_.end());
// Always decrease the deficit_ counter, again because `f` is allowed
// to call consume again.
deficit_ -= ts;
auto res = consumer(*ptr);
if (res == task_result::skip) {
// Push the unconsumed item to the cache.
cache_.push_back(ptr);
if (list_.empty()) {
deficit_ = 0;
return {consumed != 0, false};
}
// Fix deficit counter since we didn't actually use it.
deficit_ += ts;
} else {
typename unique_pointer::deleter_type d;
d(ptr);
++consumed;
if (!cache_.empty())
auto ptr = next();
if (ptr == nullptr)
return {false, false};
do {
auto consumer_res = consumer(*ptr);
switch (consumer_res) {
case task_result::skip:
// Fix deficit counter since we didn't actually use it.
deficit_ += policy().task_size(*ptr);
// Push the unconsumed item to the cache.
cache_.push_back(ptr.release());
if (list_.empty()) {
deficit_ = 0;
return {consumed != 0, false};
}
break;
case task_result::resume:
++consumed;
flush_cache();
if (list_.empty()) {
deficit_ = 0;
return {consumed != 0, false};
}
break;
default:
++consumed;
flush_cache();
if (list_.empty()) {
deficit_ = 0;
return {consumed != 0, res == task_result::stop_all};
}
if (res != task_result::resume)
return {consumed != 0, res == task_result::stop_all};
if (list_.empty())
deficit_ = 0;
return {consumed != 0, consumer_res == task_result::stop_all};
}
// Next iteration.
ptr = list_.front();
ts = policy().task_size(*ptr);
}
ptr = next();
} while (ptr != nullptr);
return {consumed != 0, false};
}
......
......@@ -83,29 +83,6 @@ public:
// nop
}
/// Takes the first element out of the queue if the deficit allows it and
/// returns the element.
unique_pointer take_front() noexcept {
unique_pointer result;
if (!super::empty()) {
auto ptr = promote(super::head_.next);
auto ts = super::policy_.task_size(*ptr);
CAF_ASSERT(ts > 0);
if (ts <= deficit_) {
deficit_ -= ts;
super::total_task_size_ -= ts;
super::head_.next = ptr->next;
if (super::total_task_size_ == 0) {
CAF_ASSERT(super::head_.next == &(super::tail_));
deficit_ = 0;
super::tail_.next = &(super::head_);
}
result.reset(ptr);
}
}
return result;
}
/// Consumes items from the queue until the queue is empty or there is not
/// enough deficit to dequeue the next task.
/// @returns `true` if `f` consumed at least one item.
......@@ -115,13 +92,20 @@ public:
return res.consumed_items;
}
/// Takes the first element out of the queue if the deficit allows it and
/// returns the element.
/// @private
unique_pointer next() noexcept {
return super::next(deficit_);
}
/// Run a new round with `quantum`, dispatching all tasks to `consumer`.
/// @returns `true` if at least one item was consumed, `false` otherwise.
template <class F>
new_round_result new_round(deficit_type quantum, F& consumer) {
if (!super::empty()) {
deficit_ += quantum;
auto ptr = take_front();
auto ptr = next();
if (ptr == nullptr)
return {false, false};
do {
......@@ -133,7 +117,7 @@ public:
case task_result::stop_all:
return {true, true};
}
ptr = take_front();
ptr = next();
} while (ptr != nullptr);
return {true, false};
}
......
......@@ -155,6 +155,28 @@ public:
dec_total_task_size(policy_.task_size(x));
}
/// @private
unique_pointer next(task_size_type& deficit) noexcept {
unique_pointer result;
if (!empty()) {
auto ptr = promote(head_.next);
auto ts = policy_.task_size(*ptr);
CAF_ASSERT(ts > 0);
if (ts <= deficit) {
deficit -= ts;
total_task_size_ -= ts;
head_.next = ptr->next;
if (total_task_size_ == 0) {
CAF_ASSERT(head_.next == &(tail_));
deficit = 0;
tail_.next = &(head_);
}
result.reset(ptr);
}
}
return result;
}
// -- iterator access --------------------------------------------------------
/// Returns an iterator to the dummy before the first element.
......
......@@ -260,19 +260,15 @@ bool blocking_actor::await_data(timeout_type timeout) {
}
mailbox_element_ptr blocking_actor::dequeue() {
mailbox().flush_cache();
await_data();
mailbox().fetch_more();
auto& qs = mailbox_.queue().queues();
auto& q1 = get<mailbox_policy::urgent_queue_index>(qs);
if (!q1.empty()) {
q1.inc_deficit(1);
return q1.take_front();
}
auto& q2 = get<mailbox_policy::default_queue_index>(qs);
if (!q2.empty()) {
q2.inc_deficit(1);
return q2.take_front();
}
return nullptr;
auto& qs = mailbox().queue().queues();
auto result = get<mailbox_policy::urgent_queue_index>(qs).take_front();
if (!result)
result = get<mailbox_policy::default_queue_index>(qs).take_front();
CAF_ASSERT(result != nullptr);
return result;
}
void blocking_actor::varargs_tup_receive(receive_cond& rcc, message_id mid,
......
......@@ -145,6 +145,28 @@ CAF_TEST(skipping) {
CAF_CHECK_EQUAL(seq, "246");
}
CAF_TEST(take_front) {
std::string seq;
fill(queue, 1, 2, 3, 4, 5, 6);
auto f = [&](inode& x) {
seq += to_string(x);
return task_result::resume;
};
CAF_CHECK_EQUAL(queue.deficit(), 0);
while (!queue.empty()) {
auto ptr = queue.take_front();
f(*ptr);
}
CAF_CHECK_EQUAL(seq, "123456");
fill(queue, 5, 4, 3, 2, 1);
while (!queue.empty()) {
auto ptr = queue.take_front();
f(*ptr);
}
CAF_CHECK_EQUAL(seq, "12345654321");
CAF_CHECK_EQUAL(queue.deficit(), 0);
}
CAF_TEST(alternating_consumer) {
using fun_type = std::function<task_result (inode&)>;
fun_type f;
......
......@@ -84,7 +84,7 @@ CAF_TEST(default_constructed) {
CAF_REQUIRE_EQUAL(queue.deficit(), 0);
CAF_REQUIRE_EQUAL(queue.total_task_size(), 0);
CAF_REQUIRE_EQUAL(queue.peek(), nullptr);
CAF_REQUIRE_EQUAL(queue.take_front(), nullptr);
CAF_REQUIRE_EQUAL(queue.next(), nullptr);
CAF_REQUIRE_EQUAL(queue.begin(), queue.end());
CAF_REQUIRE_EQUAL(queue.before_begin()->next, queue.end().ptr);
}
......@@ -98,7 +98,7 @@ CAF_TEST(inc_deficit) {
queue.inc_deficit(100);
CAF_REQUIRE_EQUAL(queue.deficit(), 100);
// Deficit must drop back down to 0 once the queue becomes empty.
queue.take_front();
queue.next();
CAF_REQUIRE_EQUAL(queue.deficit(), 0);
}
......@@ -131,6 +131,32 @@ CAF_TEST(new_round) {
CAF_CHECK_EQUAL(queue.deficit(), 0);
}
CAF_TEST(next) {
std::string seq;
fill(queue, 1, 2, 3, 4, 5, 6);
auto f = [&](inode& x) {
seq += to_string(x);
return task_result::resume;
};
auto take = [&] {
queue.flush_cache();
queue.inc_deficit(queue.peek()->value);
return queue.next();
};
while (!queue.empty()) {
auto ptr = take();
f(*ptr);
}
CAF_CHECK_EQUAL(seq, "123456");
fill(queue, 5, 4, 3, 2, 1);
while (!queue.empty()) {
auto ptr = take();
f(*ptr);
}
CAF_CHECK_EQUAL(seq, "12345654321");
CAF_CHECK_EQUAL(queue.deficit(), 0);
}
CAF_TEST(peek_all) {
auto queue_to_string = [&] {
std::string str;
......
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