Unverified Commit 40baa6b3 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #724

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