Commit ff61ae0f authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Fix bug in cached_queue::consume

parent d1bb08aa
...@@ -184,52 +184,58 @@ public: ...@@ -184,52 +184,58 @@ public:
/// @returns `true` if `f` consumed at least one item. /// @returns `true` if `f` consumed at least one item.
template <class F> template <class F>
bool consume(F& f) noexcept(noexcept(f(std::declval<value_type&>()))) { bool consume(F& f) noexcept(noexcept(f(std::declval<value_type&>()))) {
long consumed = 0;
if (!cache_.empty()) if (!cache_.empty())
list_.prepend(cache_); list_.prepend(cache_);
if (!list_.empty()) { if (list_.empty())
auto ptr = list_.front(); return false;
auto ts = policy().task_size(*ptr); long consumed = 0;
while (ts <= deficit_) { auto ptr = list_.front();
CAF_ASSERT(ts > 0); auto ts = policy().task_size(*ptr);
auto next = ptr->next; while (ts <= deficit_) {
dec_total_task_size(ts); CAF_ASSERT(ts > 0);
// Make sure the queue is in a consistent state before calling `f` in dec_total_task_size(ts);
// case `f` recursively calls consume. auto next = ptr->next;
list_.before_begin()->next = next; // Make sure the queue is in a consistent state before calling `f` in
if (total_task_size() == 0) { // case `f` recursively calls consume.
CAF_ASSERT(list_.begin() == list_.end()); list_.before_begin()->next = next;
list_.end()->next = list_.begin().ptr; if (next == list_.end().ptr)
list_.end()->next = list_.before_begin().ptr;
CAF_ASSERT(total_task_size() != 0 || list_.begin() == list_.end());
/*
if (total_task_size() == 0) {
CAF_ASSERT(list_.begin() == list_.end());
list_.end()->next = list_.begin().ptr;
}
*/
// Always decrease the deficit_ counter, again because `f` is allowed
// to call consume again.
deficit_ -= ts;
auto res = f(*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;
} }
// Always decrease the deficit_ counter, again because `f` is allowed // Fix deficit counter since we didn't actually use it.
// to call consume again. deficit_ += ts;
deficit_ -= ts; } else {
auto res = f(*ptr); deleter_type d;
if (res == task_result::skip) { d(ptr);
// Fix deficit and push the unconsumed item to the cache. ++consumed;
deficit_ += ts; if (!cache_.empty())
cache_.push_back(ptr); list_.prepend(cache_);
if (list_.empty()) { if (list_.empty()) {
deficit_ = 0; deficit_ = 0;
return consumed != 0; return consumed != 0;
}
} else {
deleter_type d;
d(ptr);
++consumed;
if (!cache_.empty())
list_.prepend(cache_);
if (list_.empty()) {
deficit_ = 0;
return consumed != 0;
}
if (res == task_result::stop)
return consumed != 0;
} }
// Next iteration. if (res == task_result::stop)
ptr = list_.front(); return consumed != 0;
ts = policy().task_size(*ptr);
} }
// Next iteration.
ptr = list_.front();
ts = policy().task_size(*ptr);
} }
return consumed != 0; return consumed != 0;
} }
......
...@@ -117,6 +117,34 @@ CAF_TEST(new_round) { ...@@ -117,6 +117,34 @@ CAF_TEST(new_round) {
CAF_CHECK_EQUAL(queue.deficit(), 0); CAF_CHECK_EQUAL(queue.deficit(), 0);
} }
CAF_TEST(skipping) {
// Define a function object for consuming even numbers.
std::string seq;
auto f = [&](inode& x) -> task_result {
if ((x.value & 0x01) == 1)
return task_result::skip;
seq += to_string(x);
return task_result::resume;
};
CAF_MESSAGE("make a round on an empty queue");
CAF_CHECK_EQUAL(queue.new_round(10, f), false);
CAF_MESSAGE("make a round on a queue with only odd numbers (skip all)");
fill(queue, 1, 3, 5);
CAF_CHECK_EQUAL(queue.new_round(10, f), false);
CAF_MESSAGE("make a round on a queue with an even number at the front");
fill(queue, 2);
CAF_CHECK_EQUAL(queue.new_round(10, f), true);
CAF_CHECK_EQUAL(seq, "2");
CAF_MESSAGE("make a round on a queue with an even number in between");
fill(queue, 7, 9, 4, 11, 13);
CAF_CHECK_EQUAL(queue.new_round(10, f), true);
CAF_CHECK_EQUAL(seq, "24");
CAF_MESSAGE("make a round on a queue with an even number at the back");
fill(queue, 15, 17, 6);
CAF_CHECK_EQUAL(queue.new_round(10, f), true);
CAF_CHECK_EQUAL(seq, "246");
}
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;
......
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