Commit f8688545 authored by Dominik Charousset's avatar Dominik Charousset

Add message_queue::drop

parent 9602b7e5
...@@ -69,6 +69,9 @@ public: ...@@ -69,6 +69,9 @@ public:
void push(execution_unit* ctx, uint64_t id, strong_actor_ptr receiver, void push(execution_unit* ctx, uint64_t id, strong_actor_ptr receiver,
mailbox_element_ptr content); mailbox_element_ptr content);
/// Marks given ID as dropped, effectively skipping it without effect.
void drop(execution_unit* ctx, uint64_t id);
/// Returns the next ascending ID. /// Returns the next ascending ID.
uint64_t new_id(); uint64_t new_id();
}; };
......
...@@ -18,6 +18,8 @@ ...@@ -18,6 +18,8 @@
#include "caf/io/basp/message_queue.hpp" #include "caf/io/basp/message_queue.hpp"
#include <iterator>
namespace caf { namespace caf {
namespace io { namespace io {
namespace basp { namespace basp {
...@@ -29,7 +31,6 @@ message_queue::message_queue() : next_id(0), next_undelivered(0) { ...@@ -29,7 +31,6 @@ message_queue::message_queue() : next_id(0), next_undelivered(0) {
void message_queue::push(execution_unit* ctx, uint64_t id, void message_queue::push(execution_unit* ctx, uint64_t id,
strong_actor_ptr receiver, strong_actor_ptr receiver,
mailbox_element_ptr content) { mailbox_element_ptr content) {
CAF_ASSERT(receiver != nullptr);
std::unique_lock<std::mutex> guard{lock}; std::unique_lock<std::mutex> guard{lock};
CAF_ASSERT(id >= next_undelivered); CAF_ASSERT(id >= next_undelivered);
CAF_ASSERT(id < next_id); CAF_ASSERT(id < next_id);
...@@ -37,6 +38,7 @@ void message_queue::push(execution_unit* ctx, uint64_t id, ...@@ -37,6 +38,7 @@ void message_queue::push(execution_unit* ctx, uint64_t id,
auto last = pending.end(); auto last = pending.end();
if (id == next_undelivered) { if (id == next_undelivered) {
// Dispatch current head. // Dispatch current head.
if (receiver != nullptr)
receiver->enqueue(std::move(content), ctx); receiver->enqueue(std::move(content), ctx);
// Check whether we can deliver more. // Check whether we can deliver more.
if (first == last || first->id != next_undelivered + 1) { if (first == last || first->id != next_undelivered + 1) {
...@@ -52,6 +54,7 @@ void message_queue::push(execution_unit* ctx, uint64_t id, ...@@ -52,6 +54,7 @@ void message_queue::push(execution_unit* ctx, uint64_t id,
CAF_ASSERT(last_hit != first); CAF_ASSERT(last_hit != first);
auto new_last = last_hit == last ? last : last_hit + 1; auto new_last = last_hit == last ? last : last_hit + 1;
for (auto i = first; i != new_last; ++i) for (auto i = first; i != new_last; ++i)
if (i->receiver != nullptr)
i->receiver->enqueue(std::move(i->content), ctx); i->receiver->enqueue(std::move(i->content), ctx);
next_undelivered += static_cast<size_t>(std::distance(first, new_last)) + 1; next_undelivered += static_cast<size_t>(std::distance(first, new_last)) + 1;
pending.erase(first, new_last); pending.erase(first, new_last);
...@@ -62,8 +65,12 @@ void message_queue::push(execution_unit* ctx, uint64_t id, ...@@ -62,8 +65,12 @@ void message_queue::push(execution_unit* ctx, uint64_t id,
auto pred = [&](const actor_msg& x) { auto pred = [&](const actor_msg& x) {
return x.id >= id; return x.id >= id;
}; };
auto i = std::find_if(first, last, pred); pending.emplace(std::find_if(first, last, pred),
pending.emplace(i, actor_msg{id, std::move(receiver), std::move(content)}); actor_msg{id, std::move(receiver), std::move(content)});
}
void message_queue::drop(execution_unit* ctx, uint64_t id) {
push(ctx, id, nullptr, nullptr);
} }
uint64_t message_queue::new_id() { uint64_t message_queue::new_id() {
......
...@@ -145,4 +145,15 @@ CAF_TEST(push order 2-1-0) { ...@@ -145,4 +145,15 @@ CAF_TEST(push order 2-1-0) {
expect((ok_atom, int), from(self).to(testee).with(_, 2)); expect((ok_atom, int), from(self).to(testee).with(_, 2));
} }
CAF_TEST(dropping) {
acquire_ids(3);
push(2);
disallow((ok_atom, int), from(self).to(testee));
queue.drop(nullptr, 1);
disallow((ok_atom, int), from(self).to(testee));
push(0);
expect((ok_atom, int), from(self).to(testee).with(_, 0));
expect((ok_atom, int), from(self).to(testee).with(_, 2));
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
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