Commit ae9f2396 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/684'

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