Commit fd2e0134 authored by Dominik Charousset's avatar Dominik Charousset

Dequeue high-priority messages first

parent 3673feb6
......@@ -36,7 +36,7 @@
#include "caf/typed_actor.hpp"
#include "caf/policy/arg.hpp"
#include "caf/policy/priority_aware.hpp"
#include "caf/policy/categorized.hpp"
#include "caf/policy/downstream_messages.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/policy/upstream_messages.hpp"
......@@ -84,7 +84,7 @@ public:
using super = extended_base;
/// Stores asynchronous messages with default priority.
using default_queue = intrusive::drr_cached_queue<policy::normal_messages>;
using normal_queue = intrusive::drr_cached_queue<policy::normal_messages>;
/// Stores asynchronous messages with hifh priority.
using urgent_queue = intrusive::drr_cached_queue<policy::urgent_messages>;
......@@ -100,11 +100,10 @@ public:
using unique_pointer = mailbox_element_ptr;
using queue_type =
intrusive::wdrr_fixed_multiplexed_queue<policy::priority_aware,
default_queue, urgent_queue>;
using queue_type = intrusive::wdrr_fixed_multiplexed_queue<
policy::categorized, normal_queue, urgent_queue>;
static constexpr size_t default_queue_index = 0;
static constexpr size_t normal_queue_index = 0;
static constexpr size_t urgent_queue_index = 1;
};
......
......@@ -88,7 +88,7 @@ public:
// -- timeout management -----------------------------------------------------
/// Requests a new timeout for `mid`.
/// @pre `mid.valid()`
/// @pre `mid.is_request()`
void request_response_timeout(const duration& d, message_id mid);
// -- spawn functions --------------------------------------------------------
......
......@@ -78,30 +78,22 @@ public:
/// category for a `mailbox_element` matches its content.
template <class...>
struct mailbox_category_corrector {
static constexpr message_id apply(message_id x) {
static constexpr message_id apply(message_id x) noexcept {
return x;
}
};
template <>
struct mailbox_category_corrector<downstream_msg> {
static message_id apply(message_id x) {
CAF_IGNORE_UNUSED(x);
auto result = make_message_id(message_id::downstream_message_category
<< message_id::category_offset);
CAF_ASSERT(x.is_async() || x == result);
return result;
static constexpr message_id apply(message_id x) noexcept {
return x.with_category(message_id::downstream_message_category);
}
};
template <>
struct mailbox_category_corrector<upstream_msg> {
static message_id apply(message_id x) {
CAF_IGNORE_UNUSED(x);
auto result = make_message_id(message_id::upstream_message_category
<< message_id::category_offset);
CAF_ASSERT(x.is_async() || x == result);
return result;
static constexpr message_id apply(message_id x) noexcept {
return x.with_category(message_id::upstream_message_category);
}
};
......
This diff is collapsed.
......@@ -19,13 +19,23 @@
#pragma once
#include <cstdint>
#include <type_traits>
namespace caf {
/// Denotes the urgency of asynchronous messages.
enum class message_priority {
normal,
high = 3 // 0b11, see message_id.hpp why this is important
high = 0,
normal = 1,
};
/// @relates message_priority
using high_message_priority_constant = std::integral_constant<
message_priority, message_priority::high>;
/// @relates message_priority
using normal_message_priority_constant = std::integral_constant<
message_priority, message_priority::normal>;
} // namespace caf
......@@ -63,7 +63,9 @@ public:
template <template <class> class Queue>
static deficit_type quantum(const Queue<urgent_messages>&,
deficit_type x) noexcept {
return x * static_cast<deficit_type>(message_priority::high);
// Allow actors to consume twice as many urgent as normal messages per
// credit round.
return x + x;
}
template <class Queue>
......@@ -71,11 +73,10 @@ public:
return x;
}
static inline size_t id_of(const mailbox_element& x) noexcept {
static size_t id_of(const mailbox_element& x) noexcept {
return static_cast<size_t>(x.mid.category());
}
};
} // namespace policy
} // namespace caf
......@@ -142,7 +142,7 @@ public:
using stream_manager_map = std::map<stream_slot, stream_manager_ptr>;
/// Stores asynchronous messages with default priority.
using default_queue = intrusive::drr_cached_queue<policy::normal_messages>;
using normal_queue = intrusive::drr_cached_queue<policy::normal_messages>;
/// Stores asynchronous messages with hifh priority.
using urgent_queue = intrusive::drr_cached_queue<policy::urgent_messages>;
......@@ -171,18 +171,18 @@ public:
using unique_pointer = mailbox_element_ptr;
using queue_type =
intrusive::wdrr_fixed_multiplexed_queue<policy::categorized,
default_queue, upstream_queue,
downstream_queue, urgent_queue>;
intrusive::wdrr_fixed_multiplexed_queue<policy::categorized, urgent_queue,
normal_queue, upstream_queue,
downstream_queue>;
};
static constexpr size_t default_queue_index = 0;
static constexpr size_t urgent_queue_index = 0;
static constexpr size_t upstream_queue_index = 1;
static constexpr size_t normal_queue_index = 1;
static constexpr size_t downstream_queue_index = 2;
static constexpr size_t upstream_queue_index = 2;
static constexpr size_t urgent_queue_index = 3;
};
static constexpr size_t downstream_queue_index = 3;
/// A queue optimized for single-reader-many-writers.
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
......@@ -718,7 +718,7 @@ public:
void push_to_cache(mailbox_element_ptr ptr);
/// Returns the default queue of the mailbox that stores ordinary messages.
default_queue& get_default_queue();
normal_queue& get_normal_queue();
/// Returns the queue of the mailbox that stores `upstream_msg` messages.
upstream_queue& get_upstream_queue();
......
......@@ -221,9 +221,9 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
}
if (workers_.empty()) {
guard.unlock();
if (sender && mid.valid()) {
// tell client we have ignored this sync message by sending
// and empty message back
if (mid.is_request() && sender != nullptr) {
// Tell client we have ignored this request message by sending and empty
// message back.
sender->enqueue(nullptr, mid.response_id(), message{}, eu);
}
return true;
......
......@@ -164,7 +164,7 @@ blocking_actor::mailbox_visitor:: operator()(mailbox_element& x) {
return intrusive::task_result::stop;
};
// Skip messages that don't match our message ID.
if (mid.valid()) {
if (mid.is_response()) {
if (mid != x.mid) {
CAF_LOG_SKIP_EVENT();
return intrusive::task_result::skip;
......@@ -198,7 +198,7 @@ blocking_actor::mailbox_visitor:: operator()(mailbox_element& x) {
}
// Response handlers must get re-invoked with an error when receiving an
// unexpected message.
if (mid.valid()) {
if (mid.is_response()) {
auto err = make_error(sec::unexpected_response,
x.move_content_to_message());
mailbox_element_view<error> tmp{std::move(x.sender), x.mid,
......@@ -267,7 +267,7 @@ mailbox_element_ptr blocking_actor::dequeue() {
auto& qs = mailbox().queue().queues();
auto result = get<mailbox_policy::urgent_queue_index>(qs).take_front();
if (!result)
result = get<mailbox_policy::default_queue_index>(qs).take_front();
result = get<mailbox_policy::normal_queue_index>(qs).take_front();
CAF_ASSERT(result != nullptr);
return result;
}
......
......@@ -47,7 +47,7 @@ void forwarding_actor_proxy::forward_msg(strong_actor_ptr sender,
forwarding_stack tmp;
shared_lock<detail::shared_spinlock> guard(broker_mtx_);
if (broker_)
broker_->enqueue(nullptr, invalid_message_id,
broker_->enqueue(nullptr, make_message_id(),
make_message(forward_atom::value, std::move(sender),
fwd != nullptr ? *fwd : tmp,
strong_actor_ptr{ctrl()}, mid,
......@@ -65,7 +65,7 @@ void forwarding_actor_proxy::enqueue(mailbox_element_ptr what,
bool forwarding_actor_proxy::add_backlink(abstract_actor* x) {
if (monitorable_actor::add_backlink(x)) {
forward_msg(ctrl(), invalid_message_id,
forward_msg(ctrl(), make_message_id(),
make_message(link_atom::value, x->ctrl()));
return true;
}
......@@ -74,7 +74,7 @@ bool forwarding_actor_proxy::add_backlink(abstract_actor* x) {
bool forwarding_actor_proxy::remove_backlink(abstract_actor* x) {
if (monitorable_actor::remove_backlink(x)) {
forward_msg(ctrl(), invalid_message_id,
forward_msg(ctrl(), make_message_id(),
make_message(unlink_atom::value, x->ctrl()));
return true;
}
......
......@@ -66,14 +66,14 @@ public:
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(msg));
shared_guard guard(mtx_);
for (auto& s : subscribers_)
s->enqueue(sender, invalid_message_id, msg, host);
s->enqueue(sender, make_message_id(), msg, host);
}
void enqueue(strong_actor_ptr sender, message_id, message msg,
execution_unit* host) override {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(msg));
send_all_subscribers(sender, msg, host);
broker_->enqueue(sender, invalid_message_id, msg, host);
broker_->enqueue(sender, make_message_id(), msg, host);
}
std::pair<bool, size_t> add_subscriber(strong_actor_ptr who) {
......@@ -208,7 +208,7 @@ private:
CAF_LOG_DEBUG(CAF_ARG(acquaintances_.size())
<< CAF_ARG(src) << CAF_ARG(what));
for (auto& acquaintance : acquaintances_)
acquaintance->enqueue(src, invalid_message_id, what, context());
acquaintance->enqueue(src, make_message_id(), what, context());
}
local_group_ptr group_;
......
......@@ -145,7 +145,7 @@ void monitorable_actor::add_link(abstract_actor* x) {
}
});
if (send_exit_immediately)
x->enqueue(nullptr, invalid_message_id,
x->enqueue(nullptr, make_message_id(),
make_message(exit_msg{address(), fail_state}), nullptr);
}
......@@ -176,7 +176,7 @@ bool monitorable_actor::add_backlink(abstract_actor* x) {
success = true;
}
if (send_exit_immediately)
x->enqueue(nullptr, invalid_message_id,
x->enqueue(nullptr, make_message_id(),
make_message(exit_msg{address(), fail_state}), nullptr);
return success;
}
......
......@@ -66,7 +66,7 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
if (x.content().type_token() == make_type_token<timeout_msg>()) {
auto& tm = content.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
CAF_ASSERT(!x.mid.valid());
CAF_ASSERT(x.mid.is_async());
if (is_active_receive_timeout(tid)) {
CAF_LOG_DEBUG("handle timeout message");
if (bhvr_stack_.empty())
......
......@@ -41,8 +41,8 @@ response_promise::response_promise(strong_actor_ptr self,
source_(std::move(source)),
stages_(std::move(stages)),
id_(mid) {
// form an invalid request promise when initialized from a
// response ID, since CAF always drops messages in this case
// Form an invalid request promise when initialized from a response ID, since
// we always drop messages in this case.
if (mid.is_response()) {
source_ = nullptr;
stages_.clear();
......@@ -60,8 +60,7 @@ void response_promise::deliver(error x) {
}
void response_promise::deliver(unit_t) {
if (id_.valid())
deliver_impl(make_message());
deliver_impl(make_message());
}
bool response_promise::async() const {
......
......@@ -237,7 +237,7 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
// Clear mailbox.
if (!mailbox_.closed()) {
mailbox_.close();
get_default_queue().flush_cache();
get_normal_queue().flush_cache();
get_urgent_queue().flush_cache();
detail::sync_request_bouncer bounce{fail_state};
while (mailbox_.queue().new_round(1000, bounce).consumed_items)
......@@ -565,7 +565,7 @@ scheduled_actor::categorize(mailbox_element& x) {
}
return message_category::ordinary;
case make_type_token<timeout_msg>(): {
CAF_ASSERT(!x.mid.valid());
CAF_ASSERT(x.mid.is_async());
auto& tm = content.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
if (tm.type == receive_atom::value) {
......@@ -850,43 +850,40 @@ void scheduled_actor::push_to_cache(mailbox_element_ptr ptr) {
auto& p = mailbox_.queue().policy();
auto& qs = mailbox_.queue().queues();
// TODO: use generic lambda to avoid code duplication when switching to C++14
if (p.id_of(*ptr) == mailbox_policy::default_queue_index) {
auto& q = std::get<mailbox_policy::default_queue_index>(qs);
if (p.id_of(*ptr) == normal_queue_index) {
auto& q = std::get<normal_queue_index>(qs);
q.inc_total_task_size(q.policy().task_size(*ptr));
q.cache().push_back(ptr.release());
} else {
auto& q = std::get<mailbox_policy::default_queue_index>(qs);
auto& q = std::get<normal_queue_index>(qs);
q.inc_total_task_size(q.policy().task_size(*ptr));
q.cache().push_back(ptr.release());
}
}
scheduled_actor::default_queue& scheduled_actor::get_default_queue() {
constexpr size_t queue_id = mailbox_policy::default_queue_index;
return get<queue_id>(mailbox_.queue().queues());
scheduled_actor::normal_queue& scheduled_actor::get_normal_queue() {
return get<normal_queue_index>(mailbox_.queue().queues());
}
scheduled_actor::upstream_queue& scheduled_actor::get_upstream_queue() {
constexpr size_t queue_id = mailbox_policy::upstream_queue_index;
return get<queue_id>(mailbox_.queue().queues());
return get<upstream_queue_index>(mailbox_.queue().queues());
}
scheduled_actor::downstream_queue& scheduled_actor::get_downstream_queue() {
constexpr size_t queue_id = mailbox_policy::downstream_queue_index;
return get<queue_id>(mailbox_.queue().queues());
return get<downstream_queue_index>(mailbox_.queue().queues());
}
scheduled_actor::urgent_queue& scheduled_actor::get_urgent_queue() {
constexpr size_t queue_id = mailbox_policy::urgent_queue_index;
return get<queue_id>(mailbox_.queue().queues());
return get<urgent_queue_index>(mailbox_.queue().queues());
}
inbound_path* scheduled_actor::make_inbound_path(stream_manager_ptr mgr,
stream_slots slots,
strong_actor_ptr sender) {
static constexpr size_t queue_index = downstream_queue_index;
using policy_type = policy::downstream_messages::nested;
auto& qs = mailbox_.queue().queues();
auto res = get<2>(qs).queues().emplace(slots.receiver, policy_type{nullptr});
auto& qs = get<queue_index>(mailbox_.queue().queues()).queues();
auto res = qs.emplace(slots.receiver, policy_type{nullptr});
if (!res.second)
return nullptr;
auto path = new inbound_path(std::move(mgr), slots, std::move(sender));
......@@ -1166,7 +1163,7 @@ scheduled_actor::advance_streams(actor_clock::time_point now) {
auto cycle = stream_ticks_.interval();
cycle *= static_cast<decltype(cycle)::rep>(credit_round_ticks_);
auto bc = home_system().config().stream_desired_batch_complexity;
auto& qs = get<2>(mailbox_.queue().queues()).queues();
auto& qs = get_downstream_queue().queues();
for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get();
auto bs = static_cast<int32_t>(kvp.second.total_task_size());
......
......@@ -152,8 +152,7 @@ void stream_manager::advance() {
auto& cfg = self_->system().config();
auto bc = cfg.stream_desired_batch_complexity;
auto interval = cfg.stream_credit_round_interval;
auto& mbox = self_->mailbox();
auto& qs = get<2>(mbox.queue().queues()).queues();
auto& qs = self_->get_downstream_queue().queues();
// Iterate all queues for inbound traffic.
for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get();
......
......@@ -5,8 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* Copyright 2011-2018 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 *
......@@ -17,65 +16,83 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#define CAF_SUITE categorized
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/message_priority.hpp"
#include "caf/unit.hpp"
#include "caf/policy/categorized.hpp"
#include "caf/test/dsl.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/downstream_messages.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/policy/upstream_messages.hpp"
#include "caf/policy/urgent_messages.hpp"
#include "caf/unit.hpp"
namespace caf {
namespace policy {
using namespace caf;
/// Configures a cached WDRR fixed multiplexed queue for dispatching to two
/// nested queue (one for each message priority).
class priority_aware {
public:
// -- member types -----------------------------------------------------------
namespace {
using mapped_type = mailbox_element;
using urgent_queue = intrusive::drr_queue<policy::urgent_messages>;
using task_size_type = size_t;
using normal_queue = intrusive::drr_queue<policy::normal_messages>;
using deficit_type = size_t;
using upstream_queue = intrusive::drr_queue<policy::upstream_messages>;
using unique_pointer = mailbox_element_ptr;
using downstream_queue = intrusive::wdrr_dynamic_multiplexed_queue<
policy::downstream_messages>;
// -- constructors, destructors, and assignment operators --------------------
struct mailbox_policy {
using deficit_type = size_t;
priority_aware() = default;
using mapped_type = mailbox_element;
priority_aware(const priority_aware&) = default;
using unique_pointer = mailbox_element_ptr;
priority_aware& operator=(const priority_aware&) = default;
using queue_type = intrusive::wdrr_fixed_multiplexed_queue<
policy::categorized, urgent_queue, normal_queue, upstream_queue,
downstream_queue>;
};
constexpr priority_aware(unit_t) {
// nop
}
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
// -- interface required by wdrr_fixed_multiplexed_queue ---------------------
struct fixture{};
template <template <class> class Queue>
static deficit_type quantum(const Queue<urgent_messages>&,
deficit_type x) noexcept {
return x * static_cast<deficit_type>(message_priority::high);
}
struct consumer {
std::vector<int> ints;
template <class Queue>
static deficit_type quantum(const Queue&, deficit_type x) noexcept {
return x;
template <class Key, class Queue>
intrusive::task_result operator()(const Key&, const Queue&,
const mailbox_element& x) {
if (!x.content().match_elements<int>())
CAF_FAIL("unexepected message: " << x.content());
ints.emplace_back(x.content().get_as<int>(0));
return intrusive::task_result::resume;
}
static inline size_t id_of(const mailbox_element& x) noexcept {
return x.mid.category() != message_id::urgent_message_category ? 0u : 1u;
template <class Key, class Queue, class... Ts>
intrusive::task_result operator()(const Key&, const Queue&, const Ts&...) {
CAF_FAIL("unexepected message type");// << typeid(Ts).name());
return intrusive::task_result::resume;
}
};
} // namespace policy
} // namespace caf
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(categorized_tests, fixture)
CAF_TEST(priorities) {
mailbox_type mbox{unit, unit, unit, unit, unit};
mbox.push_back(make_mailbox_element(nullptr, make_message_id(), {}, 123));
mbox.push_back(make_mailbox_element(nullptr,
make_message_id(message_priority::high),
{}, 456));
consumer f;
mbox.new_round(1000, f);
CAF_CHECK_EQUAL(f.ints, std::vector<int>({456, 123}));
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -16,16 +16,17 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE dynamic_spawn
#include "caf/test/unit_test.hpp"
#include <stack>
#include "caf/actor_system.hpp"
#include "caf/test/dsl.hpp"
#include <atomic>
#include <chrono>
#include <iostream>
#include <functional>
#include <iostream>
#include <stack>
#include "caf/all.hpp"
......@@ -311,6 +312,29 @@ struct fixture {
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(dynamic_spawn_tests, test_coordinator_fixture<>)
CAF_TEST(mirror) {
auto mirror = self->spawn<simple_mirror>();
auto dummy = self->spawn([=](event_based_actor* ptr) -> behavior {
ptr->send(mirror, "hello mirror");
return {
[](const std::string& msg) { CAF_CHECK_EQUAL(msg, "hello mirror"); }};
});
run();
/*
self->send(mirror, "hello mirror");
run();
self->receive (
[](const std::string& msg) {
CAF_CHECK_EQUAL(msg, "hello mirror");
}
);
*/
}
CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(atom_tests, fixture)
CAF_TEST(count_mailbox) {
......@@ -338,17 +362,6 @@ CAF_TEST(self_receive_with_zero_timeout) {
);
}
CAF_TEST(mirror) {
scoped_actor self{system};
auto mirror = self->spawn<simple_mirror>();
self->send(mirror, "hello mirror");
self->receive (
[](const std::string& msg) {
CAF_CHECK_EQUAL(msg, "hello mirror");
}
);
}
CAF_TEST(detached_mirror) {
scoped_actor self{system};
auto mirror = self->spawn<simple_mirror, detached>();
......
......@@ -59,7 +59,7 @@ CAF_TEST(empty_message) {
auto m1 = make_mailbox_element(nullptr, make_message_id(),
no_stages, make_message());
CAF_CHECK(m1->mid.is_async());
CAF_CHECK(m1->mid.category() == message_id::default_message_category);
CAF_CHECK(m1->mid.category() == message_id::normal_message_category);
CAF_CHECK(m1->content().empty());
}
......@@ -67,7 +67,7 @@ CAF_TEST(non_empty_message) {
auto m1 = make_mailbox_element(nullptr, make_message_id(),
no_stages, make_message(1, 2, 3));
CAF_CHECK(m1->mid.is_async());
CAF_CHECK(m1->mid.category() == message_id::default_message_category);
CAF_CHECK(m1->mid.category() == message_id::normal_message_category);
CAF_CHECK(!m1->content().empty());
CAF_CHECK_EQUAL((fetch<int, int>(*m1)), none);
CAF_CHECK_EQUAL((fetch<int, int, int>(*m1)), make_tuple(1, 2, 3));
......@@ -97,7 +97,7 @@ CAF_TEST(tuple) {
auto m1 = make_mailbox_element(nullptr, make_message_id(),
no_stages, 1, 2, 3);
CAF_CHECK(m1->mid.is_async());
CAF_CHECK(m1->mid.category() == message_id::default_message_category);
CAF_CHECK(m1->mid.category() == message_id::normal_message_category);
CAF_CHECK(!m1->content().empty());
CAF_CHECK_EQUAL((fetch<int, int>(*m1)), none);
CAF_CHECK_EQUAL((fetch<int, int, int>(*m1)), make_tuple(1, 2, 3));
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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 message_id
#include "caf/message_id.hpp"
#include "caf/test/dsl.hpp"
using namespace caf;
namespace {
struct fixture {
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(message_id_tests, fixture)
CAF_TEST(default construction) {
message_id x;
CAF_CHECK_EQUAL(x.is_async(), true);
CAF_CHECK_EQUAL(x.is_request(), false);
CAF_CHECK_EQUAL(x.is_response(), false);
CAF_CHECK_EQUAL(x.is_answered(), false);
CAF_CHECK(x.category() == message_id::normal_message_category);
CAF_CHECK_EQUAL(x.is_urgent_message(), false);
CAF_CHECK_EQUAL(x.is_normal_message(), true);
CAF_CHECK_EQUAL(x.is_stream_message(), false);
CAF_CHECK_EQUAL(x.is_upstream_message(), false);
CAF_CHECK_EQUAL(x.is_downstream_message(), false);
CAF_CHECK_EQUAL(x, x.response_id());
CAF_CHECK_EQUAL(x.request_id().integer_value(), 0u);
CAF_CHECK(x.integer_value() == message_id::default_async_value);
}
CAF_TEST(make_message_id) {
auto x = make_message_id();
message_id y;
CAF_CHECK_EQUAL(x, y);
CAF_CHECK_EQUAL(x.integer_value(), y.integer_value());
}
CAF_TEST(from integer value) {
auto x = make_message_id(42);
CAF_CHECK_EQUAL(x.is_async(), false);
CAF_CHECK_EQUAL(x.is_request(), true);
CAF_CHECK_EQUAL(x.is_response(), false);
CAF_CHECK_EQUAL(x.is_answered(), false);
CAF_CHECK(x.category() == message_id::normal_message_category);
CAF_CHECK_EQUAL(x.is_urgent_message(), false);
CAF_CHECK_EQUAL(x.is_normal_message(), true);
CAF_CHECK_EQUAL(x.is_stream_message(), false);
CAF_CHECK_EQUAL(x.is_upstream_message(), false);
CAF_CHECK_EQUAL(x.is_downstream_message(), false);
CAF_CHECK_EQUAL(x.request_id().integer_value(), 42u);
}
CAF_TEST(response ID) {
auto x = make_message_id(42).response_id();
CAF_CHECK_EQUAL(x.is_async(), false);
CAF_CHECK_EQUAL(x.is_request(), false);
CAF_CHECK_EQUAL(x.is_response(), true);
CAF_CHECK_EQUAL(x.is_answered(), false);
CAF_CHECK(x.category() == message_id::normal_message_category);
CAF_CHECK_EQUAL(x.is_urgent_message(), false);
CAF_CHECK_EQUAL(x.is_normal_message(), true);
CAF_CHECK_EQUAL(x.is_stream_message(), false);
CAF_CHECK_EQUAL(x.is_upstream_message(), false);
CAF_CHECK_EQUAL(x.is_downstream_message(), false);
CAF_CHECK_EQUAL(x.request_id().integer_value(), 42u);
}
CAF_TEST(request with high priority) {
auto x = make_message_id(42).response_id();
CAF_CHECK_EQUAL(x.is_async(), false);
CAF_CHECK_EQUAL(x.is_request(), false);
CAF_CHECK_EQUAL(x.is_response(), true);
CAF_CHECK_EQUAL(x.is_answered(), false);
CAF_CHECK(x.category() == message_id::normal_message_category);
CAF_CHECK_EQUAL(x.is_urgent_message(), false);
CAF_CHECK_EQUAL(x.is_normal_message(), true);
CAF_CHECK_EQUAL(x.is_stream_message(), false);
CAF_CHECK_EQUAL(x.is_upstream_message(), false);
CAF_CHECK_EQUAL(x.is_downstream_message(), false);
CAF_CHECK_EQUAL(x.request_id().integer_value(), 42u);
}
CAF_TEST(with_category) {
auto x = make_message_id();
CAF_CHECK(x.category() == message_id::normal_message_category);
for (auto category : {message_id::urgent_message_category,
message_id::downstream_message_category,
message_id::upstream_message_category,
message_id::normal_message_category}) {
x = x.with_category(category);
CAF_CHECK_EQUAL(x.category(), category);
CAF_CHECK_EQUAL(x.is_request(), false);
CAF_CHECK_EQUAL(x.is_response(), false);
CAF_CHECK_EQUAL(x.is_answered(), false);
}
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -44,6 +44,7 @@
#include "caf/mailbox_element.hpp"
#include "caf/no_stages.hpp"
#include "caf/outbound_path.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/send.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_sink_driver.hpp"
......@@ -126,17 +127,18 @@ const char* name_of(const actor_addr& x) {
// -- queues -------------------------------------------------------------------
using default_queue = drr_queue<policy::normal_messages>;
using mboxqueue = scheduled_actor::mailbox_policy::queue_type;
using dmsg_queue = wdrr_dynamic_multiplexed_queue<policy::downstream_messages>;
template <size_t Value>
using uint_constant = std::integral_constant<size_t, Value>;
using umsg_queue = drr_queue<policy::upstream_messages>;
using urgent_async_id = uint_constant<scheduled_actor::urgent_queue_index>;
using urgent_queue = drr_queue<policy::urgent_messages>;
using normal_async_id = uint_constant<scheduled_actor::normal_queue_index>;
using mboxqueue =
wdrr_fixed_multiplexed_queue<policy::categorized, default_queue, umsg_queue,
dmsg_queue, urgent_queue>;
using umsg_id = uint_constant<scheduled_actor::upstream_queue_index>;
using dmsg_id = uint_constant<scheduled_actor::downstream_queue_index>;
// -- entity and mailbox visitor -----------------------------------------------
......@@ -336,7 +338,7 @@ public:
}
if (x % ticks_per_credit_interval == 0) {
// Fill credit on each input path up to 30.
auto& qs = get<2>(mbox.queues()).queues();
auto& qs = get<dmsg_id::value>(mbox.queues()).queues();
for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get();
auto bs = static_cast<int32_t>(kvp.second.total_task_size());
......@@ -350,7 +352,7 @@ public:
inbound_path* make_inbound_path(stream_manager_ptr mgr, stream_slots slots,
strong_actor_ptr sender) override {
using policy_type = policy::downstream_messages::nested;
auto res = get<2>(mbox.queues())
auto res = get<dmsg_id::value>(mbox.queues())
.queues().emplace(slots.receiver, policy_type{nullptr});
if (!res.second)
return nullptr;
......@@ -360,11 +362,11 @@ public:
}
void erase_inbound_path_later(stream_slot slot) override {
get<2>(mbox.queues()).erase_later(slot);
get<dmsg_id::value>(mbox.queues()).erase_later(slot);
}
void erase_inbound_paths_later(const stream_manager* mgr) override {
for (auto& kvp : get<2>(mbox.queues()).queues()) {
for (auto& kvp : get<dmsg_id::value>(mbox.queues()).queues()) {
auto& path = kvp.second.policy().handler;
if (path != nullptr && path->mgr == mgr)
erase_inbound_path_later(kvp.first);
......@@ -397,19 +399,18 @@ public:
struct msg_visitor {
// -- member types -----------------------------------------------------------
using is_default_async = std::integral_constant<size_t, 0>;
using is_umsg = std::integral_constant<size_t, 1>;
using is_dmsg = std::integral_constant<size_t, 2>;
using is_urgent_async = std::integral_constant<size_t, 3>;
using result_type = intrusive::task_result;
// -- operator() overloads ---------------------------------------------------
result_type operator()(is_default_async, default_queue&, mailbox_element& x) {
result_type operator()(urgent_async_id, entity::urgent_queue&,
mailbox_element&) {
CAF_FAIL("unexpected function call");
return intrusive::task_result::stop;
}
result_type operator()(normal_async_id, entity::normal_queue&,
mailbox_element& x) {
CAF_REQUIRE_EQUAL(x.content().type_token(),
make_type_token<open_stream_msg>());
self->current_mailbox_element(&x);
......@@ -418,12 +419,7 @@ struct msg_visitor {
return intrusive::task_result::resume;
}
result_type operator()(is_urgent_async, urgent_queue&, mailbox_element&) {
CAF_FAIL("unexpected function call");
return intrusive::task_result::stop;
}
result_type operator()(is_umsg, umsg_queue&, mailbox_element& x) {
result_type operator()(umsg_id, entity::upstream_queue&, mailbox_element& x) {
CAF_REQUIRE(x.content().type_token() == make_type_token<upstream_msg>());
self->current_mailbox_element(&x);
auto& um = x.content().get_mutable_as<upstream_msg>(0);
......@@ -446,7 +442,7 @@ struct msg_visitor {
return intrusive::task_result::resume;
}
result_type operator()(is_dmsg, dmsg_queue& qs, stream_slot,
result_type operator()(dmsg_id, entity::downstream_queue& qs, stream_slot,
policy::downstream_messages::nested_queue_type& q,
mailbox_element& x) {
CAF_REQUIRE(x.content().type_token() == make_type_token<downstream_msg>());
......@@ -554,7 +550,7 @@ struct fixture {
// Check whether all actors cleaned up their state properly.
entity* xs[] = {&alice, &bob, &carl};
for (auto x : xs) {
CAF_CHECK(get<2>(x->mbox.queues()).queues().empty());
CAF_CHECK(get<dmsg_id::value>(x->mbox.queues()).queues().empty());
CAF_CHECK(x->pending_stream_managers().empty());
CAF_CHECK(x->stream_managers().empty());
}
......
......@@ -292,7 +292,7 @@ void basp_broker_state::deliver(const node_id& src_nid, actor_id src_aid,
self->parent().notify<hook::invalid_message_received>(src_nid, src,
invalid_actor_id,
mid, msg);
if (mid.valid() && src) {
if (mid.is_request() && src != nullptr) {
detail::sync_request_bouncer srb{rsn};
srb(src, mid);
}
......@@ -362,7 +362,8 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
// send message to SpawnServ of remote node
basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag,
0, 0, this_node(), nid, tmp.id(), invalid_actor_id,
0, make_message_id().integer_value(), this_node(), nid,
tmp.id(), invalid_actor_id,
visit(seq_num_visitor{this}, path->hdl)};
// writing std::numeric_limits<actor_id>::max() is a hack to get
// this send-to-named-actor feature working with older CAF releases
......@@ -412,7 +413,8 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
});
basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag,
0, 0, this_node(), nid, tmp.id(), invalid_actor_id,
0, make_message_id().integer_value(), this_node(), nid,
tmp.id(), invalid_actor_id,
visit(seq_num_visitor{this}, path->hdl)};
instance.write(self->context(), get_buffer(path->hdl),
hdr, &writer);
......
......@@ -57,7 +57,7 @@ void manager::detach(execution_unit*, bool invoke_disconnect_message) {
ptr.swap(parent_);
detach_from(raw_ptr);
if (invoke_disconnect_message) {
auto mptr = make_mailbox_element(nullptr, invalid_message_id,
auto mptr = make_mailbox_element(nullptr, make_message_id(),
{}, detach_message());
switch (raw_ptr->consume(*mptr)) {
case im_success:
......
......@@ -39,6 +39,8 @@
namespace {
using caf::make_message_id;
struct anything { };
anything any_vals;
......@@ -69,6 +71,7 @@ bool operator==(const maybe<T>& x, const T& y) {
constexpr uint8_t no_flags = 0;
constexpr uint32_t no_payload = 0;
constexpr uint64_t no_operation_data = 0;
constexpr uint64_t default_operation_data = make_message_id().integer_value();
constexpr auto basp_atom = caf::atom("BASP");
constexpr auto spawn_serv_atom = caf::atom("SpawnServ");
......@@ -285,7 +288,7 @@ public:
.receive(hdl,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
no_operation_data,
default_operation_data,
this_node(), n.id,
any_vals, invalid_actor_id,
spawn_serv_atom,
......@@ -374,7 +377,7 @@ public:
auto end = first + hdr.payload_len;
payload.assign(first, end);
CAF_MESSAGE("erase " << std::distance(ob.begin(), end)
<< " bytes from output buffer");
<< " bytes from output buffer");
ob.erase(ob.begin(), end);
} else {
ob.erase(ob.begin(), ob.begin() + basp::header_size);
......@@ -559,13 +562,13 @@ CAF_TEST(message_forwarding) {
auto msg = make_message(1, 2, 3);
// send a message from node 0 to node 1, forwarded by this node
mock(jupiter().connection,
{basp::message_type::dispatch_message, 0, 0, 0,
{basp::message_type::dispatch_message, 0, 0, default_operation_data,
jupiter().id, mars().id,
invalid_actor_id, mars().dummy_actor->id()},
msg)
.receive(mars().connection,
basp::message_type::dispatch_message, no_flags, any_vals,
no_operation_data, jupiter().id, mars().id,
default_operation_data, jupiter().id, mars().id,
invalid_actor_id, mars().dummy_actor->id(),
msg);
}
......@@ -609,7 +612,7 @@ CAF_TEST(remote_actor_and_send) {
.receive(jupiter().connection,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
no_operation_data, this_node(), jupiter().id,
default_operation_data, this_node(), jupiter().id,
any_vals, invalid_actor_id,
spawn_serv_atom,
std::vector<actor_id>{},
......@@ -645,7 +648,7 @@ CAF_TEST(remote_actor_and_send) {
mock()
.receive(jupiter().connection,
basp::message_type::dispatch_message, no_flags, any_vals,
no_operation_data, this_node(), jupiter().id,
default_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id(),
std::vector<actor_id>{},
make_message(42));
......@@ -699,11 +702,10 @@ CAF_TEST(actor_serialize_and_deserialize) {
while (mpx()->output_buffer(jupiter().connection).empty())
mpx()->exec_runnable(); // process forwarded message in basp_broker
// output buffer must contain the reflected message
mock()
.receive(jupiter().connection,
basp::message_type::dispatch_message, no_flags, any_vals,
no_operation_data, this_node(), prx->node(), testee->id(), prx->id(),
std::vector<actor_id>{}, msg);
mock().receive(jupiter().connection, basp::message_type::dispatch_message,
no_flags, any_vals, default_operation_data, this_node(),
prx->node(), testee->id(), prx->id(), std::vector<actor_id>{},
msg);
}
CAF_TEST(indirect_connections) {
......@@ -729,7 +731,7 @@ CAF_TEST(indirect_connections) {
mx.receive(mars().connection,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
no_operation_data, this_node(), jupiter().id,
default_operation_data, this_node(), jupiter().id,
any_vals, invalid_actor_id,
spawn_serv_atom,
std::vector<actor_id>{},
......@@ -750,7 +752,7 @@ CAF_TEST(indirect_connections) {
mock()
.receive(mars().connection,
basp::message_type::dispatch_message, no_flags, any_vals,
no_operation_data, this_node(), jupiter().id,
default_operation_data, this_node(), jupiter().id,
self()->id(), jupiter().dummy_actor->id(),
std::vector<actor_id>{},
make_message("hello from earth!"));
......@@ -795,15 +797,15 @@ CAF_TEST(automatic_connection) {
make_message("hello from jupiter!"))
.receive(mars().connection,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals, no_operation_data,
this_node(), jupiter().id, any_vals, invalid_actor_id,
spawn_serv_atom,
basp::header::named_receiver_flag, any_vals,
default_operation_data, this_node(), jupiter().id,
any_vals, invalid_actor_id, spawn_serv_atom,
std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info"))
.receive(mars().connection,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
no_operation_data, this_node(), jupiter().id,
default_operation_data, this_node(), jupiter().id,
any_vals, // actor ID of an actor spawned by the BASP broker
invalid_actor_id,
config_serv_atom,
......@@ -864,7 +866,7 @@ CAF_TEST(automatic_connection) {
mock()
.receive(jupiter().connection,
basp::message_type::dispatch_message, no_flags, any_vals,
no_operation_data, this_node(), jupiter().id,
default_operation_data, this_node(), jupiter().id,
self()->id(), jupiter().dummy_actor->id(),
std::vector<actor_id>{},
make_message("hello from earth!"));
......
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