Commit c6c25889 authored by Dominik Charousset's avatar Dominik Charousset

Fix recursive dequeue operations in cache

parent 40b5b480
...@@ -17,8 +17,8 @@ ...@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_DETAIL_INTRUSIVE_LIST_HPP #ifndef CAF_DETAIL_INTRUSIVE_PARTITIONED_LIST_HPP
#define CAF_DETAIL_INTRUSIVE_LIST_HPP #define CAF_DETAIL_INTRUSIVE_PARTITIONED_LIST_HPP
#include <memory> #include <memory>
#include <iterator> #include <iterator>
...@@ -179,30 +179,38 @@ class intrusive_partitioned_list { ...@@ -179,30 +179,38 @@ class intrusive_partitioned_list {
bool invoke(Actor* self, iterator first, iterator last, Vs&... vs) { bool invoke(Actor* self, iterator first, iterator last, Vs&... vs) {
pointer prev = first->prev; pointer prev = first->prev;
pointer next = first->next; pointer next = first->next;
auto patch = [&] { auto move_on = [&](bool first_valid) {
prev->next = next; if (first_valid) {
next->prev = prev; prev = first.ptr;
}
first = next;
next = first->next;
}; };
while (first != last) { while (first != last) {
std::unique_ptr<value_type, deleter_type> tmp{first.ptr}; std::unique_ptr<value_type, deleter_type> tmp{first.ptr};
// since this function can be called recursively during
// self->invoke_message(tmp, vs...), we have to remove the
// element from the list proactively and put it back in if
// it's safe, i.e., if invoke_message returned im_skipped
prev->next = next;
next->prev = prev;
switch (self->invoke_message(tmp, vs...)) { switch (self->invoke_message(tmp, vs...)) {
case policy::im_dropped: case policy::im_dropped:
patch(); move_on(false);
first = next;
next = first->next;
break; break;
case policy::im_success: case policy::im_success:
patch();
return true; return true;
default: case policy::im_skipped:
if (tmp) { if (tmp) {
prev = tmp.release(); // re-integrate tmp and move on
first = next; prev->next = tmp.get();
next = first->next; next->prev = tmp.release();
} else { move_on(true);
patch(); }
first = next; else {
next = first->next; // only happens if the user does something
// really, really stupid; check it nonetheless
move_on(false);
} }
break; break;
} }
...@@ -234,4 +242,4 @@ class intrusive_partitioned_list { ...@@ -234,4 +242,4 @@ class intrusive_partitioned_list {
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_DETAIL_INTRUSIVE_LIST_HPP #endif // CAF_DETAIL_INTRUSIVE_PARTITIONED_LIST_HPP
...@@ -27,9 +27,13 @@ ...@@ -27,9 +27,13 @@
using std::begin; using std::begin;
using std::end; using std::end;
using namespace caf;
namespace { namespace {
size_t s_iint_instances = 0; size_t s_iint_instances = 0;
}
} // namespace <anonymous>
struct iint { struct iint {
iint* next; iint* next;
...@@ -54,18 +58,52 @@ inline bool operator==(const iint& lhs, int rhs) { return lhs.value == rhs; } ...@@ -54,18 +58,52 @@ inline bool operator==(const iint& lhs, int rhs) { return lhs.value == rhs; }
inline bool operator==(int lhs, const iint& rhs) { return lhs == rhs.value; } inline bool operator==(int lhs, const iint& rhs) { return lhs == rhs.value; }
using iint_queue = caf::detail::single_reader_queue<iint>; using iint_queue = detail::single_reader_queue<iint>;
struct pseudo_actor { struct pseudo_actor {
using mailbox_type = caf::detail::single_reader_queue<iint>; using mailbox_type = detail::single_reader_queue<iint>;
using uptr = mailbox_type::unique_pointer;
mailbox_type mbox; mailbox_type mbox;
mailbox_type& mailbox() { mailbox_type& mailbox() {
return mbox; return mbox;
} }
policy::invoke_message_result invoke_message(uptr& ptr, int i) {
if (ptr->value == 1) {
ptr.reset();
return policy::im_dropped;
}
if (ptr->value == i) {
// call reset on some of our messages
if (ptr->is_high_priority()) {
ptr.reset();
}
return policy::im_success;
}
return policy::im_skipped;
}
template <class Policy>
policy::invoke_message_result invoke_message(uptr& ptr, Policy& policy, int i,
std::vector<int>& remaining) {
switch (invoke_message(ptr, i)) {
case policy::im_dropped:
return policy::im_dropped;
case policy::im_skipped:
return policy::im_skipped;
case policy::im_success:
if (!remaining.empty()) {
auto next = remaining.front();
remaining.erase(remaining.begin());
policy.invoke_from_cache(this, policy, next, remaining);
}
return policy::im_success;
}
}
}; };
int main() { void test_prioritizing_dequeue() {
CAF_TEST(test_intrusive_containers);
pseudo_actor self; pseudo_actor self;
auto baseline = s_iint_instances; auto baseline = s_iint_instances;
CAF_CHECK_EQUAL(baseline, 3); // "begin", "separator", and "end" CAF_CHECK_EQUAL(baseline, 3); // "begin", "separator", and "end"
...@@ -74,7 +112,7 @@ int main() { ...@@ -74,7 +112,7 @@ int main() {
self.mbox.enqueue(new iint(3)); self.mbox.enqueue(new iint(3));
self.mbox.enqueue(new iint(4)); self.mbox.enqueue(new iint(4));
CAF_CHECK_EQUAL(baseline + 4, s_iint_instances); CAF_CHECK_EQUAL(baseline + 4, s_iint_instances);
caf::policy::prioritizing policy; policy::prioritizing policy;
// first "high priority message" // first "high priority message"
CAF_CHECK_EQUAL(self.mbox.count(), 4); CAF_CHECK_EQUAL(self.mbox.count(), 4);
CAF_CHECK_EQUAL(policy.has_next_message(&self), true); CAF_CHECK_EQUAL(policy.has_next_message(&self), true);
...@@ -99,5 +137,80 @@ int main() { ...@@ -99,5 +137,80 @@ int main() {
// back to baseline // back to baseline
CAF_CHECK_EQUAL(self.mbox.count(), 0); CAF_CHECK_EQUAL(self.mbox.count(), 0);
CAF_CHECK_EQUAL(baseline, s_iint_instances); CAF_CHECK_EQUAL(baseline, s_iint_instances);
}
template <class Policy>
void test_invoke_from_cache() {
pseudo_actor self;
auto baseline = s_iint_instances;
CAF_CHECK_EQUAL(baseline, 3); // "begin", "separator", and "end"
self.mbox.enqueue(new iint(1));
self.mbox.enqueue(new iint(2));
self.mbox.enqueue(new iint(3));
self.mbox.enqueue(new iint(4));
self.mbox.enqueue(new iint(5));
Policy policy;
CAF_CHECK_EQUAL(self.mbox.count(), 5);
// fill cache
auto ptr = policy.next_message(&self);
while (ptr) {
policy.push_to_cache(&self, std::move(ptr));
ptr = policy.next_message(&self);
}
CAF_CHECK_EQUAL(self.mbox.count(), 5);
// dequeue 3, 2, 4, 1, note: 1 is dropped on first dequeue
int expected;
//std::vector<int> expected{3, 2, 4, 5};
size_t remaining = 4;
//for (auto& i : expected) {
CAF_CHECK(policy.invoke_from_cache(&self, expected = 3));
CAF_CHECK_EQUAL(self.mbox.count(), --remaining);
CAF_CHECK(policy.invoke_from_cache(&self, expected = 2));
CAF_CHECK_EQUAL(self.mbox.count(), --remaining);
CAF_CHECK(policy.invoke_from_cache(&self, expected = 4));
CAF_CHECK_EQUAL(self.mbox.count(), --remaining);
CAF_CHECK(policy.invoke_from_cache(&self, expected = 5));
CAF_CHECK_EQUAL(self.mbox.count(), --remaining);
//}
}
template <class Policy>
void test_recursive_invoke_from_cache() {
pseudo_actor self;
auto baseline = s_iint_instances;
CAF_CHECK_EQUAL(baseline, 3); // "begin", "separator", and "end"
self.mbox.enqueue(new iint(1));
self.mbox.enqueue(new iint(2));
self.mbox.enqueue(new iint(3));
self.mbox.enqueue(new iint(4));
self.mbox.enqueue(new iint(5));
Policy policy;
CAF_CHECK_EQUAL(self.mbox.count(), 5);
// fill cache
auto ptr = policy.next_message(&self);
while (ptr) {
policy.push_to_cache(&self, std::move(ptr));
ptr = policy.next_message(&self);
}
CAF_CHECK_EQUAL(self.mbox.count(), 5);
// dequeue 3, 2, 4, 1, note: 1 is dropped on first dequeue
std::vector<int> remaining{2, 4, 5};
int first = 3;
policy.invoke_from_cache(&self, policy, first, remaining);
CAF_CHECK_EQUAL(self.mbox.count(), 0);
}
int main() {
CAF_TEST(test_intrusive_containers);
CAF_CHECKPOINT();
test_prioritizing_dequeue();
CAF_PRINT("test_invoke_from_cache<policy::prioritizing>");
test_invoke_from_cache<policy::prioritizing>();
CAF_PRINT("test_invoke_from_cache<policy::not_prioritizing>");
test_invoke_from_cache<policy::not_prioritizing>();
CAF_PRINT("test_recursive_invoke_from_cache<policy::prioritizing>");
test_recursive_invoke_from_cache<policy::prioritizing>();
CAF_PRINT("test_recursive_invoke_from_cache<policy::not_prioritizing>");
test_recursive_invoke_from_cache<policy::not_prioritizing>();
return CAF_TEST_RESULT(); return CAF_TEST_RESULT();
} }
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