Commit 0b5accf5 authored by Dominik Charousset's avatar Dominik Charousset

Fix behavior of DRR cached queues

Skipped elements must remain in the cache until a message was processed
successfully in `new_round` instead of prepended at the start of every
new round.
parent 95554bc0
...@@ -108,18 +108,14 @@ public: ...@@ -108,18 +108,14 @@ public:
return total_task_size() == 0; return total_task_size() == 0;
} }
/// Peeks at the first element of the cache or of the list in case the former /// Peeks at the first element of the list.
/// is empty.
pointer peek() noexcept { pointer peek() noexcept {
auto ptr = cache_.peek(); return list_.peek();
return ptr == nullptr ? list_.peek() : ptr;
} }
/// Applies `f` to each element in the queue, including cached elements. /// Applies `f` to each element in the queue, excluding cached elements.
template <class F> template <class F>
void peek_all(F f) const { void peek_all(F f) const {
for (auto i = cache_.begin(); i != cache_.end(); ++i)
f(*promote(i.ptr));
for (auto i = list_.begin(); i != list_.end(); ++i) for (auto i = list_.begin(); i != list_.end(); ++i)
f(*promote(i.ptr)); f(*promote(i.ptr));
} }
...@@ -153,10 +149,10 @@ public: ...@@ -153,10 +149,10 @@ 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 if the deficit allows it and /// Takes the first element out of the queue (after flushing the cache) if
/// returns the element. /// the deficit allows it and returns the element.
unique_pointer take_front() noexcept { unique_pointer take_front() noexcept {
list_.prepend(cache_); flush_cache();
unique_pointer result; unique_pointer result;
if (!list_.empty()) { if (!list_.empty()) {
auto ptr = list_.front(); auto ptr = list_.front();
...@@ -189,8 +185,6 @@ public: ...@@ -189,8 +185,6 @@ public:
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)
noexcept(noexcept(consumer(std::declval<value_type&>()))) { noexcept(noexcept(consumer(std::declval<value_type&>()))) {
if (!cache_.empty())
list_.prepend(cache_);
if (list_.empty()) if (list_.empty())
return {false, false}; return {false, false};
deficit_ += quantum; deficit_ += quantum;
...@@ -225,7 +219,7 @@ public: ...@@ -225,7 +219,7 @@ public:
d(ptr); d(ptr);
++consumed; ++consumed;
if (!cache_.empty()) if (!cache_.empty())
list_.prepend(cache_); flush_cache();
if (list_.empty()) { if (list_.empty()) {
deficit_ = 0; deficit_ = 0;
return {consumed != 0, res == task_result::stop_all}; return {consumed != 0, res == task_result::stop_all};
......
...@@ -196,8 +196,10 @@ CAF_TEST(peek_all) { ...@@ -196,8 +196,10 @@ CAF_TEST(peek_all) {
queue.emplace_back(2); queue.emplace_back(2);
CAF_CHECK_EQUAL(queue_to_string(), "2"); CAF_CHECK_EQUAL(queue_to_string(), "2");
queue.cache().emplace_back(1); queue.cache().emplace_back(1);
CAF_CHECK_EQUAL(queue_to_string(), "1, 2"); CAF_CHECK_EQUAL(queue_to_string(), "2");
queue.emplace_back(3); queue.emplace_back(3);
CAF_CHECK_EQUAL(queue_to_string(), "2, 3");
queue.flush_cache();
CAF_CHECK_EQUAL(queue_to_string(), "1, 2, 3"); CAF_CHECK_EQUAL(queue_to_string(), "1, 2, 3");
} }
...@@ -206,6 +208,8 @@ CAF_TEST(to_string) { ...@@ -206,6 +208,8 @@ CAF_TEST(to_string) {
fill(queue, 3, 4); fill(queue, 3, 4);
CAF_CHECK_EQUAL(deep_to_string(queue), "[3, 4]"); CAF_CHECK_EQUAL(deep_to_string(queue), "[3, 4]");
fill(queue.cache(), 1, 2); fill(queue.cache(), 1, 2);
CAF_CHECK_EQUAL(deep_to_string(queue), "[3, 4]");
queue.flush_cache();
CAF_CHECK_EQUAL(deep_to_string(queue), "[1, 2, 3, 4]"); CAF_CHECK_EQUAL(deep_to_string(queue), "[1, 2, 3, 4]");
} }
......
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