Commit 940e37a3 authored by Dominik Charousset's avatar Dominik Charousset

Implement new, scheduled BASP broker

parent be404167
...@@ -37,6 +37,7 @@ set(LIBCAF_IO_SRCS ...@@ -37,6 +37,7 @@ set(LIBCAF_IO_SRCS
src/native_socket.cpp src/native_socket.cpp
src/pipe_reader.cpp src/pipe_reader.cpp
src/protocol.cpp src/protocol.cpp
src/proxy.cpp
src/receive_buffer.cpp src/receive_buffer.cpp
src/routing_table.cpp src/routing_table.cpp
src/scribe.cpp src/scribe.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 "caf/actor.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/wdrr_dynamic_multiplexed_queue.hpp"
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
#include "caf/policy/categorized.hpp"
#include "caf/policy/downstream_messages.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/policy/upstream_messages.hpp"
#include "caf/policy/urgent_messages.hpp"
#include "caf/resumable.hpp"
namespace caf {
namespace io {
namespace basp {
/// Serializes any message it receives and forwards it to the BASP broker.
class proxy : public actor_proxy, public resumable {
public:
// -- member types -----------------------------------------------------------
using super = actor_proxy;
/// Required by `spawn`, `anon_send`, etc. for type deduction.
using signatures = none_t;
/// Stores asynchronous messages with default priority.
using normal_queue = intrusive::drr_queue<policy::normal_messages>;
/// Stores asynchronous messages with hifh priority.
using urgent_queue = intrusive::drr_queue<policy::urgent_messages>;
/// Stores upstream messages.
using upstream_queue = intrusive::drr_queue<policy::upstream_messages>;
/// Stores downstream messages.
using downstream_queue = intrusive::drr_queue<
policy::downstream_messages::nested>;
/// Configures the FIFO inbox with four nested queues:
///
/// 1. Default asynchronous messages
/// 2. High-priority asynchronous messages
/// 3. Upstream messages
/// 4. Downstream messages
///
/// The queue for downstream messages is in turn composed of a nested queues,
/// one for each active input slot.
struct mailbox_policy {
using deficit_type = size_t;
using mapped_type = mailbox_element;
using unique_pointer = mailbox_element_ptr;
using queue_type = intrusive::wdrr_fixed_multiplexed_queue<
policy::categorized, urgent_queue, normal_queue, upstream_queue,
downstream_queue>;
};
/// A queue optimized for single-reader-many-writers.
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
// -- constructors and destructors -------------------------------------------
proxy(actor_config& cfg, actor dispatcher);
~proxy() override;
// -- properties -------------------------------------------------------------
const actor& dispatcher() {
return dispatcher_;
}
execution_unit* context() {
return context_;
}
// -- implementation of abstract_actor ---------------------------------------
void enqueue(mailbox_element_ptr ptr, execution_unit* eu) override;
bool add_backlink(abstract_actor* x) override;
bool remove_backlink(abstract_actor* x) override;
mailbox_element* peek_at_next_mailbox_element() override;
// -- implementation of monitorable_actor ------------------------------------
void on_cleanup(const error& reason) override;
// -- implementation of actor_proxy ------------------------------------------
void kill_proxy(execution_unit* ctx, error reason) override;
// -- implementation of resumable --------------------------------------------
resume_result resume(execution_unit* ctx, size_t max_throughput) override;
void intrusive_ptr_add_ref_impl() override;
void intrusive_ptr_release_impl() override;
private:
// -- member variables -------------------------------------------------------
/// Stores incoming messages.
mailbox_type mailbox_;
/// Actor for dispatching serialized BASP messages.
actor dispatcher_;
/// Points to the current execution unit when being resumed.
execution_unit* context_;
};
} // 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/proxy.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/intrusive/task_result.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/io/basp/message_type.hpp"
#include "caf/logger.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/send.hpp"
namespace caf {
namespace io {
namespace basp {
// -- constructors and destructors ---------------------------------------------
proxy::proxy(actor_config& cfg, actor dispatcher)
: super(cfg),
mailbox_(unit, unit, unit, unit, nullptr),
dispatcher_(std::move(dispatcher)) {
// Tell the dispatcher we have proxy now that needs monitoring of the remote
// actor it represents.
anon_send(dispatcher_, monitor_atom::value, ctrl());
// All proxies behave as-if spawned with `lazy_init`.
mailbox_.try_block();
}
proxy::~proxy() {
// nop
}
// -- implementation of actor_proxy --------------------------------------------
void proxy::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
CAF_ASSERT(ptr != nullptr);
CAF_LOG_TRACE(CAF_ARG(*ptr));
CAF_LOG_SEND_EVENT(ptr);
auto mid = ptr->mid;
auto sender = ptr->sender;
switch (mailbox_.push_back(std::move(ptr))) {
case intrusive::inbox_result::unblocked_reader: {
CAF_LOG_ACCEPT_EVENT(true);
// add a reference count to this actor and re-schedule it
intrusive_ptr_add_ref(ctrl());
if (eu != nullptr)
eu->exec_later(this);
else
home_system().scheduler().enqueue(this);
break;
}
case intrusive::inbox_result::queue_closed: {
CAF_LOG_REJECT_EVENT();
if (mid.is_request()) {
detail::sync_request_bouncer f{exit_reason()};
f(sender, mid);
}
break;
}
case intrusive::inbox_result::success:
// enqueued to a running actors' mailbox; nothing to do
CAF_LOG_ACCEPT_EVENT(false);
break;
}
}
mailbox_element* proxy::peek_at_next_mailbox_element() {
return mailbox_.closed() || mailbox_.blocked() ? nullptr : mailbox_.peek();
}
bool proxy::add_backlink(abstract_actor* x){
if (monitorable_actor::add_backlink(x)) {
anon_send(this, link_atom::value, x->ctrl());
return true;
}
return false;
}
bool proxy::remove_backlink(abstract_actor* x){
if (monitorable_actor::remove_backlink(x)) {
anon_send(this, unlink_atom::value, x->ctrl());
return true;
}
return false;
}
void proxy::on_cleanup(const error& reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
CAF_IGNORE_UNUSED(reason);
dispatcher_ = nullptr;
}
void proxy::kill_proxy(execution_unit*, error err) {
anon_send(this, sys_atom::value, exit_msg{nullptr, std::move(err)});
}
// -- implementation of resumable ----------------------------------------------
namespace {
class mailbox_visitor {
public:
mailbox_visitor(proxy& self, size_t& handled_msgs, size_t max_throughput)
: self_(self),
handled_msgs_(handled_msgs),
max_throughput_(max_throughput) {
// nop
}
intrusive::task_result operator()(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x));
// Check for kill_proxy event.
if (x.content().match_elements<sys_atom, exit_msg>()) {
auto& err = x.content().get_mutable_as<exit_msg>(1);
self_.cleanup(std::move(err.reason), self_.context());
return intrusive::task_result::stop_all;
}
// Stores the payload.
std::vector<char> buf;
binary_serializer sink{self_.home_system(), buf};
// Differntiate between routed and direct messages based on the sender.
message_type msg_type;
if (x.sender == nullptr || x.sender->node() == self_.home_system().node()) {
msg_type = message_type::direct_message;
if (auto err = sink(x.stages)) {
CAF_LOG_ERROR("cannot serialize stages:" << x);
return intrusive::task_result::stop_all;
}
if (auto err = message::save(sink, x.content())) {
CAF_LOG_ERROR("cannot serialize content:" << x);
return intrusive::task_result::stop_all;
}
} else {
msg_type = message_type::routed_message;
if (auto err = sink(x.sender->node(), self_.node(), x.stages)) {
CAF_LOG_ERROR("cannot serialize source, destination, and stages:" << x);
return intrusive::task_result::stop_all;
}
if (auto err = message::save(sink, x.content())) {
CAF_LOG_ERROR("cannot serialize content:" << x);
return intrusive::task_result::stop_all;
}
}
// Fill in the header and ship the message to the BASP broker.
header hdr{msg_type,
0,
static_cast<uint32_t>(buf.size()),
x.mid.integer_value(),
x.sender == nullptr ? 0 : x.sender->id(),
self_.id()};
anon_send(self_.dispatcher(), self_.ctrl(), hdr, std::move(buf));
return ++handled_msgs_ < max_throughput_ ? intrusive::task_result::resume
: intrusive::task_result::stop_all;
}
template <class Queue>
intrusive::task_result operator()(size_t, Queue&, mailbox_element& x) {
return (*this)(x);
}
template <class OuterQueue, class InnerQueue>
intrusive::task_result operator()(size_t, OuterQueue&, stream_slot,
InnerQueue&, mailbox_element& x) {
return (*this)(x);
}
private:
proxy& self_;
size_t& handled_msgs_;
size_t max_throughput_;
};
} // namespace
resumable::resume_result proxy::resume(execution_unit* ctx, size_t max_throughput) {
CAF_PUSH_AID(id());
CAF_LOG_TRACE(CAF_ARG(max_throughput));
context_ = ctx;
size_t handled_msgs = 0;
mailbox_visitor f{*this, handled_msgs, max_throughput};
while (handled_msgs < max_throughput) {
CAF_LOG_DEBUG("start new DRR round");
// TODO: maybe replace '3' with configurable / adaptive value?
// Dispatch on the different message categories in our mailbox.
if (!mailbox_.new_round(3, f).consumed_items) {
// Check whether cleanup() was called.
if (dispatcher_ == nullptr)
return resumable::done;
if (mailbox_.try_block())
return resumable::awaiting_message;
}
}
CAF_LOG_DEBUG("max throughput reached");
return mailbox_.try_block() ? resumable::awaiting_message
: resumable::resume_later;
}
void proxy::intrusive_ptr_add_ref_impl() {
intrusive_ptr_add_ref(ctrl());
}
void proxy::intrusive_ptr_release_impl() {
intrusive_ptr_release(ctrl());
}
} // 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_proxy
#include "caf/io/basp/proxy.hpp"
#include "caf/test/dsl.hpp"
#include <deque>
#include "caf/binary_deserializer.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/make_actor.hpp"
#include "caf/send.hpp"
using std::string;
using buffer = std::vector<char>;
using namespace caf;
namespace {
// -- convenience structs for holding direct or routed messages ----------------
struct direct_msg {
io::basp::header hdr;
mailbox_element::forwarding_stack stages;
message content;
static direct_msg from(actor_system& sys, const io::basp::header& hdr,
const buffer& buf) {
direct_msg result;
result.hdr = hdr;
binary_deserializer source{sys, buf};
if (result.hdr.operation != io::basp::message_type::direct_message)
CAF_FAIL("direct message != " << to_string(result.hdr.operation));
if (result.hdr.payload_len != buf.size())
CAF_FAIL("BASP header has invalid payload size: expected "
<< buf.size() << ", got " << result.hdr.payload_len);
if (auto err = source(result.stages, result.content))
CAF_FAIL("failed to deserialize payload: " << sys.render(err));
return result;
}
};
struct routed_msg {
io::basp::header hdr;
mailbox_element::forwarding_stack stages;
message content;
node_id src;
node_id dst;
static routed_msg from(actor_system& sys, const io::basp::header& hdr,
const buffer& buf) {
routed_msg result;
result.hdr = hdr;
binary_deserializer source{sys, buf};
if (result.hdr.operation != io::basp::message_type::routed_message)
CAF_FAIL("routed message != " << to_string(result.hdr.operation));
if (result.hdr.payload_len != buf.size())
CAF_FAIL("BASP header has invalid payload size: expected "
<< buf.size() << ", got " << result.hdr.payload_len);
if (auto err = source(result.src, result.dst, result.stages,
result.content))
CAF_FAIL("failed to deserialize payload: " << sys.render(err));
return result;
}
};
// -- fake dispatcher that mimics a BASP broker --------------------------------
/// A dispatcher that simply exposes everything it receives via a FIFO queue.
struct dispatcher_state {
struct item {
strong_actor_ptr sender;
io::basp::header hdr;
buffer buf;
};
std::deque<item> items;
item next() {
auto result = std::move(items.front());
items.pop_front();
return result;
}
};
using dispatcher_type = stateful_actor<dispatcher_state>;
behavior fake_dispatcher(dispatcher_type* self) {
self->set_default_handler(drop);
return {[=](strong_actor_ptr receiver, io::basp::header hdr, buffer& buf) {
using item = dispatcher_state::item;
self->state.items.emplace_back(item{receiver, hdr, std::move(buf)});
}};
}
// -- simple dummy actor for testing message delivery --------------------------
class dummy_actor:public event_based_actor {
public:
dummy_actor(actor_config& cfg) : event_based_actor(cfg) {
//nop
}
behavior make_behavior() override {
return {[] {}};
}
};
// -- fixture setup ------------------------------------------------------------
struct fixture : test_coordinator_fixture<> {
fixture()
: mars(42, "0011223344556677889900112233445566778899"),
jupiter(23, "99887766554433221100998877665544332211FF") {
dispatcher = sys.spawn<lazy_init>(fake_dispatcher);
actor_config cfg;
aut = make_actor<io::basp::proxy, actor>(42, mars, &sys, cfg, dispatcher);
expect((monitor_atom, strong_actor_ptr), to(dispatcher));
}
~fixture() {
anon_send_exit(dispatcher, exit_reason::user_shutdown);
}
dispatcher_state::item next_item() {
return deref<dispatcher_type>(dispatcher).state.next();
}
/// Gets the next item from the dispatcher and deserializes a direct_message
/// from the buffer content.
direct_msg next_direct_msg() {
auto item = next_item();
if (item.sender != aut)
CAF_FAIL("message is not directed at our actor-under-test");
return direct_msg::from(sys, item.hdr, item.buf);
}
/// Gets the next item from the dispatcher and deserializes a routed_message
/// from the buffer content.
routed_msg next_routed_msg() {
auto item = next_item();
if (item.sender != aut)
CAF_FAIL("message is not directed at our actor-under-test");
return routed_msg::from(sys, item.hdr, item.buf);
}
node_id mars;
node_id jupiter;
actor dispatcher;
actor aut;
};
} // namespace <anonymous>
// -- unit tests ---------------------------------------------------------------
CAF_TEST_FIXTURE_SCOPE(proxy_tests, fixture)
CAF_TEST(forward anonymous message) {
anon_send(aut, "hello proxy!");
expect((std::string), to(aut));
expect((strong_actor_ptr, io::basp::header, buffer), to(dispatcher));
auto msg = next_direct_msg();
CAF_CHECK_EQUAL(msg.stages.size(), 0u);
CAF_CHECK_EQUAL(to_string(msg.content), R"__(("hello proxy!"))__");
}
CAF_TEST(forward message from local actor) {
self->send(aut, "hi there!");
expect((std::string), to(aut));
expect((strong_actor_ptr, io::basp::header, buffer), to(dispatcher));
auto msg = next_direct_msg();
CAF_CHECK_EQUAL(msg.stages.size(), 0u);
CAF_CHECK_EQUAL(to_string(msg.content), R"__(("hi there!"))__");
}
CAF_TEST(forward message from remote actor) {
actor_config cfg;
auto testee = make_actor<dummy_actor>(42, jupiter, &sys, cfg);
send_as(testee, aut, "hello from jupiter!");
expect((std::string), to(aut));
expect((strong_actor_ptr, io::basp::header, buffer), to(dispatcher));
auto msg = next_routed_msg();
CAF_CHECK_EQUAL(msg.src, jupiter);
CAF_CHECK_EQUAL(msg.dst, mars);
CAF_CHECK_EQUAL(msg.stages.size(), 0u);
CAF_CHECK_EQUAL(to_string(msg.content), R"__(("hello from jupiter!"))__");
}
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