Commit 9ecfd120 authored by Dominik Charousset's avatar Dominik Charousset

Add message queue to BASP for local reordering

parent 65f675e4
......@@ -8,44 +8,45 @@ file(GLOB_RECURSE LIBCAF_IO_HDRS "caf/*.hpp")
# list cpp files excluding platform-dependent files
set(LIBCAF_IO_SRCS
src/abstract_broker.cpp
src/acceptor.cpp
src/acceptor_manager.cpp
src/acceptor_manager.cpp
src/basp_broker.cpp
src/broker.cpp
src/connection_helper.cpp
src/datagram_handler.cpp
src/datagram_manager.cpp
src/datagram_servant.cpp
src/datagram_servant_impl.cpp
src/default_multiplexer.cpp
src/doorman.cpp
src/doorman_impl.cpp
src/event_handler.cpp
src/header.cpp
src/instance.cpp
src/interfaces.cpp
src/ip_endpoint.cpp
src/manager.cpp
src/message_queue.cpp
src/message_type.cpp
src/middleman.cpp
src/middleman_actor.cpp
src/middleman_actor_impl.cpp
src/multiplexer.cpp
src/multiplexer.cpp
src/native_socket.cpp
src/pipe_reader.cpp
src/protocol.cpp
src/receive_buffer.cpp
src/routing_table.cpp
src/scribe.cpp
src/stream_manager.cpp
src/test_multiplexer.cpp
src/acceptor.cpp
src/datagram_handler.cpp
src/datagram_servant_impl.cpp
src/doorman_impl.cpp
src/event_handler.cpp
src/pipe_reader.cpp
src/scribe_impl.cpp
src/socket_guard.cpp
src/stream.cpp
src/stream_manager.cpp
src/tcp.cpp
src/test_multiplexer.cpp
src/udp.cpp
src/native_socket.cpp
src/socket_guard.cpp
)
add_custom_target(libcaf_io)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <mutex>
#include <vector>
#include "caf/actor_control_block.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
namespace caf {
namespace io {
namespace basp {
/// Enforces strict order of message delivery, i.e., deliver messages in the
/// same order as if they were deserialized by a single thread.
class message_queue {
public:
// -- member types -----------------------------------------------------------
/// Request for sending a message to an actor at a later time.
struct actor_msg {
uint64_t id;
strong_actor_ptr receiver;
mailbox_element_ptr content;
};
// -- constructors, destructors, and assignment operators --------------------
message_queue();
// -- member variables -------------------------------------------------------
/// Protects all other properties.
std::mutex lock;
/// The next available ascending ID. The counter is large enough to overflow
/// after roughly 600 years if we dispatch a message every microsecond.
uint64_t next_id;
/// The next ID that we can ship.
uint64_t next_undelivered;
/// Keeps messages in sorted order in case a message other than
/// `next_undelivered` gets ready first.
std::vector<actor_msg> pending;
// -- mutators ---------------------------------------------------------------
/// Adds a new message to the queue or deliver it immediately if possible.
void push(execution_unit* ctx, uint64_t id, strong_actor_ptr receiver,
mailbox_element_ptr content);
/// Returns the next ascending ID.
uint64_t new_id();
};
} // namespace basp
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/basp/message_queue.hpp"
namespace caf {
namespace io {
namespace basp {
message_queue::message_queue() : next_id(0), next_undelivered(0) {
// nop
}
void message_queue::push(execution_unit* ctx, uint64_t id,
strong_actor_ptr receiver,
mailbox_element_ptr content) {
CAF_ASSERT(receiver != nullptr);
std::unique_lock<std::mutex> guard{lock};
CAF_ASSERT(id >= next_undelivered);
CAF_ASSERT(id < next_id);
auto first = pending.begin();
auto last = pending.end();
if (id == next_undelivered) {
// Dispatch current head.
receiver->enqueue(std::move(content), ctx);
// Check whether we can deliver more.
if (first == last || first->id != next_undelivered + 1) {
++next_undelivered;
CAF_ASSERT(next_undelivered <= next_id);
return;
}
// We look for the last element that we cannot ship right away.
auto pred = [](const actor_msg& x, const actor_msg& y) {
return x.id + 1 > y.id;
};
auto last_hit = std::adjacent_find(first, last, pred);
CAF_ASSERT(last_hit != first);
auto new_last = last_hit == last ? last : last_hit + 1;
for (auto i = first; i != new_last; ++i)
i->receiver->enqueue(std::move(i->content), ctx);
next_undelivered += static_cast<size_t>(std::distance(first, new_last)) + 1;
pending.erase(first, new_last);
CAF_ASSERT(next_undelivered <= next_id);
return;
}
// Get the insertion point.
auto pred = [&](const actor_msg& x) {
return x.id >= id;
};
auto i = std::find_if(first, last, pred);
pending.emplace(i, actor_msg{id, std::move(receiver), std::move(content)});
}
uint64_t message_queue::new_id() {
std::unique_lock<std::mutex> guard{lock};
return next_id++;
}
} // namespace basp
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE io_basp_message_queue
#include "caf/io/basp/message_queue.hpp"
#include "caf/test/dsl.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp"
#include "caf/behavior.hpp"
using namespace caf;
namespace {
behavior testee_impl() {
return {
[](ok_atom, int) {
// nop
}
};
}
struct fixture : test_coordinator_fixture<> {
io::basp::message_queue queue;
strong_actor_ptr testee;
fixture() {
auto hdl = sys.spawn<lazy_init>(testee_impl);
testee = actor_cast<strong_actor_ptr>(hdl);
}
void acquire_ids(size_t num) {
for (size_t i = 0; i < num; ++i)
queue.new_id();
}
void push(int msg_id) {
queue.push(nullptr, static_cast<uint64_t>(msg_id), testee,
make_mailbox_element(self->ctrl(), make_message_id(), {},
ok_atom::value, msg_id));
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(message_queue_tests, fixture)
CAF_TEST(default construction) {
CAF_CHECK_EQUAL(queue.next_id, 0u);
CAF_CHECK_EQUAL(queue.next_undelivered, 0u);
CAF_CHECK_EQUAL(queue.pending.size(), 0u);
}
CAF_TEST(ascending IDs) {
CAF_CHECK_EQUAL(queue.new_id(), 0u);
CAF_CHECK_EQUAL(queue.new_id(), 1u);
CAF_CHECK_EQUAL(queue.new_id(), 2u);
CAF_CHECK_EQUAL(queue.next_undelivered, 0u);
}
CAF_TEST(push order 0-1-2) {
acquire_ids(3);
push(0);
expect((ok_atom, int), from(self).to(testee).with(_, 0));
push(1);
expect((ok_atom, int), from(self).to(testee).with(_, 1));
push(2);
expect((ok_atom, int), from(self).to(testee).with(_, 2));
}
CAF_TEST(push order 0-2-1) {
acquire_ids(3);
push(0);
expect((ok_atom, int), from(self).to(testee).with(_, 0));
push(2);
disallow((ok_atom, int), from(self).to(testee));
push(1);
expect((ok_atom, int), from(self).to(testee).with(_, 1));
expect((ok_atom, int), from(self).to(testee).with(_, 2));
}
CAF_TEST(push order 1-0-2) {
acquire_ids(3);
push(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(_, 1));
push(2);
expect((ok_atom, int), from(self).to(testee).with(_, 2));
}
CAF_TEST(push order 1-2-0) {
acquire_ids(3);
push(1);
disallow((ok_atom, int), from(self).to(testee));
push(2);
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(_, 1));
expect((ok_atom, int), from(self).to(testee).with(_, 2));
}
CAF_TEST(push order 2-0-1) {
acquire_ids(3);
push(2);
disallow((ok_atom, int), from(self).to(testee));
push(0);
expect((ok_atom, int), from(self).to(testee).with(_, 0));
push(1);
expect((ok_atom, int), from(self).to(testee).with(_, 1));
expect((ok_atom, int), from(self).to(testee).with(_, 2));
}
CAF_TEST(push order 2-1-0) {
acquire_ids(3);
push(2);
disallow((ok_atom, int), from(self).to(testee));
push(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(_, 1));
expect((ok_atom, int), from(self).to(testee).with(_, 2));
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -464,7 +464,7 @@ public:
return;
auto res = try_extract<Ts...>(dest_);
if (res != caf::none)
CAF_FAIL("received disallowed message: " << CAF_ARG(*res));
CAF_FAIL("received disallowed message: " << caf::deep_to_string(*ptr));
};
}
......
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