Commit 962d1439 authored by Dominik Charousset's avatar Dominik Charousset

Re-implement mailbox_element as simple struct

parent ddfd8def
...@@ -117,7 +117,6 @@ set(CAF_CORE_SOURCES ...@@ -117,7 +117,6 @@ set(CAF_CORE_SOURCES
src/message_builder.cpp src/message_builder.cpp
src/message_handler.cpp src/message_handler.cpp
src/message_priority_strings.cpp src/message_priority_strings.cpp
src/message_view.cpp
src/monitorable_actor.cpp src/monitorable_actor.cpp
src/node_id.cpp src/node_id.cpp
src/outbound_path.cpp src/outbound_path.cpp
......
...@@ -109,9 +109,9 @@ protected: ...@@ -109,9 +109,9 @@ protected:
void on_cleanup(const error& reason) override; void on_cleanup(const error& reason) override;
private: private:
bool bool filter(upgrade_lock<detail::shared_spinlock>&,
filter(upgrade_lock<detail::shared_spinlock>&, const strong_actor_ptr& sender, const strong_actor_ptr& sender, message_id mid, message& msg,
message_id mid, message_view& mv, execution_unit* eu); execution_unit* eu);
// call without workers_mtx_ held // call without workers_mtx_ held
void quit(execution_unit* host); void quit(execution_unit* host);
......
...@@ -22,14 +22,13 @@ ...@@ -22,14 +22,13 @@
#include <type_traits> #include <type_traits>
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/message_view.hpp"
#include "caf/result.hpp" #include "caf/result.hpp"
namespace caf { namespace caf {
template <class F> template <class F>
struct catch_all { struct catch_all {
using fun_type = std::function<result<message>(message_view&)>; using fun_type = std::function<result<message>(message&)>;
F handler; F handler;
...@@ -44,7 +43,7 @@ struct catch_all { ...@@ -44,7 +43,7 @@ struct catch_all {
static_assert(std::is_convertible<F, fun_type>::value, static_assert(std::is_convertible<F, fun_type>::value,
"catch-all handler must have signature " "catch-all handler must have signature "
"result<message> (message_view&)"); "result<message> (message&)");
fun_type lift() const { fun_type lift() const {
return handler; return handler;
......
...@@ -35,7 +35,7 @@ public: ...@@ -35,7 +35,7 @@ public:
virtual ~blocking_behavior(); virtual ~blocking_behavior();
virtual result<message> fallback(message_view&); virtual result<message> fallback(message&);
virtual timespan timeout(); virtual timespan timeout();
...@@ -54,7 +54,7 @@ public: ...@@ -54,7 +54,7 @@ public:
blocking_behavior_v2(blocking_behavior_v2&&) = default; blocking_behavior_v2(blocking_behavior_v2&&) = default;
result<message> fallback(message_view& x) override { result<message> fallback(message& x) override {
return f.handler(x); return f.handler(x);
} }
}; };
...@@ -93,7 +93,7 @@ public: ...@@ -93,7 +93,7 @@ public:
blocking_behavior_v4(blocking_behavior_v4&&) = default; blocking_behavior_v4(blocking_behavior_v4&&) = default;
result<message> fallback(message_view& x) override { result<message> fallback(message& x) override {
return f1.handler(x); return f1.handler(x);
} }
......
...@@ -46,15 +46,13 @@ public: ...@@ -46,15 +46,13 @@ public:
} }
behavior make_behavior() override { behavior make_behavior() override {
auto f = [=](scheduled_actor*, message_view& xs) -> result<message> { auto f = [=](scheduled_actor*, message& msg) -> result<message> {
auto msg = xs.move_content_to_message();
auto rp = this->make_response_promise(); auto rp = this->make_response_promise();
split_(workset_, msg); split_(workset_, msg);
for (auto& x : workset_) for (auto& x : workset_)
this->send(x.first, std::move(x.second)); this->send(x.first, std::move(x.second));
auto g auto g
= [=](scheduled_actor*, message_view& ys) mutable -> result<message> { = [=](scheduled_actor*, message& res) mutable -> result<message> {
auto res = ys.move_content_to_message();
join_(value_, res); join_(value_, res);
if (--awaited_results_ == 0) { if (--awaited_results_ == 0) {
rp.deliver(value_); rp.deliver(value_);
......
...@@ -87,8 +87,8 @@ struct CAF_CORE_EXPORT function_view_storage_catch_all { ...@@ -87,8 +87,8 @@ struct CAF_CORE_EXPORT function_view_storage_catch_all {
// nop // nop
} }
result<message> operator()(message_view& xs) { result<message> operator()(message& msg) {
*storage_ = xs.move_content_to_message(); *storage_ = std::move(msg);
return message{}; return message{};
} }
}; };
......
...@@ -307,7 +307,7 @@ using stream_manager_ptr = intrusive_ptr<stream_manager>; ...@@ -307,7 +307,7 @@ using stream_manager_ptr = intrusive_ptr<stream_manager>;
// -- unique pointer aliases --------------------------------------------------- // -- unique pointer aliases ---------------------------------------------------
using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>; using mailbox_element_ptr = std::unique_ptr<mailbox_element>;
using tracing_data_ptr = std::unique_ptr<tracing_data>; using tracing_data_ptr = std::unique_ptr<tracing_data>;
using type_erased_value_ptr = std::unique_ptr<type_erased_value>; using type_erased_value_ptr = std::unique_ptr<type_erased_value>;
......
...@@ -33,7 +33,6 @@ ...@@ -33,7 +33,6 @@
#include "caf/memory_managed.hpp" #include "caf/memory_managed.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/message_view.hpp"
#include "caf/meta/omittable_if_empty.hpp" #include "caf/meta/omittable_if_empty.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
...@@ -43,9 +42,7 @@ ...@@ -43,9 +42,7 @@
namespace caf { namespace caf {
class CAF_CORE_EXPORT mailbox_element class CAF_CORE_EXPORT mailbox_element
: public intrusive::singly_linked<mailbox_element>, : public intrusive::singly_linked<mailbox_element> {
public memory_managed,
public message_view {
public: public:
using forwarding_stack = std::vector<strong_actor_ptr>; using forwarding_stack = std::vector<strong_actor_ptr>;
...@@ -66,13 +63,15 @@ public: ...@@ -66,13 +63,15 @@ public:
tracing_data_ptr tracing_id; tracing_data_ptr tracing_id;
#endif // CAF_ENABLE_ACTOR_PROFILER #endif // CAF_ENABLE_ACTOR_PROFILER
mailbox_element(); /// Stores the payload.
message payload;
mailbox_element(strong_actor_ptr&& x, message_id y, forwarding_stack&& z); mailbox_element() = default;
~mailbox_element() override; mailbox_element(strong_actor_ptr sender, message_id mid,
forwarding_stack stages, message payload);
inline bool is_high_priority() const { bool is_high_priority() const {
return mid.category() == message_id::urgent_message_category; return mid.category() == message_id::urgent_message_category;
} }
...@@ -80,31 +79,19 @@ public: ...@@ -80,31 +79,19 @@ public:
mailbox_element(const mailbox_element&) = delete; mailbox_element(const mailbox_element&) = delete;
mailbox_element& operator=(mailbox_element&&) = delete; mailbox_element& operator=(mailbox_element&&) = delete;
mailbox_element& operator=(const mailbox_element&) = delete; mailbox_element& operator=(const mailbox_element&) = delete;
};
/// Corrects the message ID for down- and upstream messages to make sure the // -- backward compatibility -------------------------------------------------
/// category for a `mailbox_element` matches its content.
template <class...>
struct mailbox_category_corrector {
static constexpr message_id apply(message_id x) noexcept {
return x;
}
};
template <> message& content() noexcept {
struct mailbox_category_corrector<downstream_msg> { return payload;
static constexpr message_id apply(message_id x) noexcept {
return x.with_category(message_id::downstream_message_category);
} }
};
template <> const message& content() const noexcept {
struct mailbox_category_corrector<upstream_msg> { return payload;
static constexpr message_id apply(message_id x) noexcept {
return x.with_category(message_id::upstream_message_category);
} }
}; };
/// @relates mailbox_element /// @relates mailbox_element
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, mailbox_element& x) { typename Inspector::result_type inspect(Inspector& f, mailbox_element& x) {
...@@ -113,112 +100,28 @@ typename Inspector::result_type inspect(Inspector& f, mailbox_element& x) { ...@@ -113,112 +100,28 @@ typename Inspector::result_type inspect(Inspector& f, mailbox_element& x) {
#ifdef CAF_ENABLE_ACTOR_PROFILER #ifdef CAF_ENABLE_ACTOR_PROFILER
x.tracing_id, x.tracing_id,
#endif // CAF_ENABLE_ACTOR_PROFILER #endif // CAF_ENABLE_ACTOR_PROFILER
x.content()); x.payload);
} }
/// Encapsulates arbitrary data in a message element.
template <class... Ts>
class mailbox_element_vals final
: public mailbox_element,
public detail::tuple_vals_impl<type_erased_tuple, Ts...> {
public:
template <class... Us>
mailbox_element_vals(strong_actor_ptr&& x0, message_id x1,
forwarding_stack&& x2, Us&&... xs)
: mailbox_element(std::move(x0),
mailbox_category_corrector<Ts...>::apply(x1),
std::move(x2)),
detail::tuple_vals_impl<type_erased_tuple, Ts...>(
std::forward<Us>(xs)...) {
// nop
}
type_erased_tuple& content() override {
return *this;
}
const type_erased_tuple& content() const override {
return *this;
}
message move_content_to_message() override {
message_factory f;
auto& xs = this->data();
return detail::apply_moved_args(f, detail::get_indices(xs), xs);
}
message copy_content_to_message() const override {
message_factory f;
auto& xs = this->data();
return detail::apply_args(f, detail::get_indices(xs), xs);
}
void dispose() noexcept {
this->deref();
}
};
/// Provides a view for treating arbitrary data as message element.
template <class... Ts>
class mailbox_element_view final
: public mailbox_element,
public detail::type_erased_tuple_view<Ts...> {
public:
mailbox_element_view(strong_actor_ptr&& x0, message_id x1,
forwarding_stack&& x2, Ts&... xs)
: mailbox_element(std::move(x0),
mailbox_category_corrector<Ts...>::apply(x1),
std::move(x2)),
detail::type_erased_tuple_view<Ts...>(xs...) {
// nop
}
type_erased_tuple& content() override {
return *this;
}
const type_erased_tuple& content() const override {
return *this;
}
message move_content_to_message() override {
message_factory f;
auto& xs = this->data();
return detail::apply_moved_args(f, detail::get_indices(xs), xs);
}
message copy_content_to_message() const override {
message_factory f;
auto& xs = this->data();
return detail::apply_args(f, detail::get_indices(xs), xs);
}
};
/// @relates mailbox_element /// @relates mailbox_element
using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>; using mailbox_element_ptr = std::unique_ptr<mailbox_element>;
/// @relates mailbox_element /// @relates mailbox_element
CAF_CORE_EXPORT mailbox_element_ptr CAF_CORE_EXPORT mailbox_element_ptr
make_mailbox_element(strong_actor_ptr sender, message_id id, make_mailbox_element(strong_actor_ptr sender, message_id id,
mailbox_element::forwarding_stack stages, message msg); mailbox_element::forwarding_stack stages, message content);
/// @relates mailbox_element /// @relates mailbox_element
template <class T, class... Ts> template <class T, class... Ts>
typename std::enable_if< std::enable_if_t<!std::is_same<typename std::decay<T>::type, message>::value
!std::is_same<typename std::decay<T>::type, message>::value
|| (sizeof...(Ts) > 0), || (sizeof...(Ts) > 0),
mailbox_element_ptr>::type mailbox_element_ptr>
make_mailbox_element(strong_actor_ptr sender, message_id id, make_mailbox_element(strong_actor_ptr sender, message_id id,
mailbox_element::forwarding_stack stages, T&& x, mailbox_element::forwarding_stack stages, T&& x,
Ts&&... xs) { Ts&&... xs) {
using impl = mailbox_element_vals< return make_mailbox_element(std::move(sender), id, std::move(stages),
typename unbox_message_element< make_message(std::forward<T>(x),
typename detail::strip_and_convert<T>::type>::type, std::forward<Ts>(xs)...));
typename unbox_message_element<
typename detail::strip_and_convert<Ts>::type>::type...>;
auto ptr = new impl(std::move(sender), id, std::move(stages),
std::forward<T>(x), std::forward<Ts>(xs)...);
return mailbox_element_ptr{ptr};
} }
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#pragma once
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
namespace caf {
/// Represents an object pointing to a `type_erased_tuple` that
/// is convertible to a `message`
class CAF_CORE_EXPORT message_view {
public:
virtual ~message_view();
virtual type_erased_tuple& content() = 0;
virtual const type_erased_tuple& content() const = 0;
virtual message move_content_to_message() = 0;
virtual message copy_content_to_message() const = 0;
};
} // namespace caf
...@@ -84,22 +84,22 @@ namespace caf { ...@@ -84,22 +84,22 @@ namespace caf {
/// @relates scheduled_actor /// @relates scheduled_actor
/// Default handler function that sends the message back to the sender. /// Default handler function that sends the message back to the sender.
CAF_CORE_EXPORT result<message> reflect(scheduled_actor*, message_view&); CAF_CORE_EXPORT result<message> reflect(scheduled_actor*, message&);
/// @relates scheduled_actor /// @relates scheduled_actor
/// Default handler function that sends /// Default handler function that sends
/// the message back to the sender and then quits. /// the message back to the sender and then quits.
CAF_CORE_EXPORT result<message> CAF_CORE_EXPORT result<message>
reflect_and_quit(scheduled_actor*, message_view&); reflect_and_quit(scheduled_actor*, message&);
/// @relates scheduled_actor /// @relates scheduled_actor
/// Default handler function that prints messages /// Default handler function that prints messages
/// message via `aout` and drops them afterwards. /// message via `aout` and drops them afterwards.
CAF_CORE_EXPORT result<message> print_and_drop(scheduled_actor*, message_view&); CAF_CORE_EXPORT result<message> print_and_drop(scheduled_actor*, message&);
/// @relates scheduled_actor /// @relates scheduled_actor
/// Default handler function that simply drops messages. /// Default handler function that simply drops messages.
CAF_CORE_EXPORT result<message> drop(scheduled_actor*, message_view&); CAF_CORE_EXPORT result<message> drop(scheduled_actor*, message&);
/// A cooperatively scheduled, event-based actor implementation. /// A cooperatively scheduled, event-based actor implementation.
/// @extends local_actor /// @extends local_actor
...@@ -192,7 +192,7 @@ public: ...@@ -192,7 +192,7 @@ public:
/// Function object for handling unmatched messages. /// Function object for handling unmatched messages.
using default_handler using default_handler
= std::function<result<message>(pointer, message_view&)>; = std::function<result<message>(pointer, message&)>;
/// Function object for handling error messages. /// Function object for handling error messages.
using error_handler = std::function<void(pointer, error&)>; using error_handler = std::function<void(pointer, error&)>;
......
...@@ -31,8 +31,7 @@ namespace caf { ...@@ -31,8 +31,7 @@ namespace caf {
/// skipping to the runtime. /// skipping to the runtime.
class CAF_CORE_EXPORT skip_t { class CAF_CORE_EXPORT skip_t {
public: public:
using fun using fun = std::function<result<message>(scheduled_actor* self, message&)>;
= std::function<result<message>(scheduled_actor* self, message_view&)>;
constexpr skip_t() { constexpr skip_t() {
// nop // nop
...@@ -45,7 +44,7 @@ public: ...@@ -45,7 +44,7 @@ public:
operator fun() const; operator fun() const;
private: private:
static result<message> skip_fun_impl(scheduled_actor*, message_view&); static result<message> skip_fun_impl(scheduled_actor*, message&);
}; };
/// Tells the runtime system to skip a message when used as message /// Tells the runtime system to skip a message when used as message
......
...@@ -54,7 +54,7 @@ void broadcast_dispatch(actor_system&, actor_pool::uplock&, ...@@ -54,7 +54,7 @@ void broadcast_dispatch(actor_system&, actor_pool::uplock&,
const actor_pool::actor_vec& vec, const actor_pool::actor_vec& vec,
mailbox_element_ptr& ptr, execution_unit* host) { mailbox_element_ptr& ptr, execution_unit* host) {
CAF_ASSERT(!vec.empty()); CAF_ASSERT(!vec.empty());
auto msg = ptr->move_content_to_message(); auto msg = ptr->payload;
for (auto& worker : vec) for (auto& worker : vec)
worker->enqueue(ptr->sender, ptr->mid, msg, host); worker->enqueue(ptr->sender, ptr->mid, msg, host);
} }
...@@ -117,7 +117,7 @@ actor actor_pool::make(execution_unit* eu, size_t num_workers, ...@@ -117,7 +117,7 @@ actor actor_pool::make(execution_unit* eu, size_t num_workers,
void actor_pool::enqueue(mailbox_element_ptr what, execution_unit* eu) { void actor_pool::enqueue(mailbox_element_ptr what, execution_unit* eu) {
upgrade_lock<detail::shared_spinlock> guard{workers_mtx_}; upgrade_lock<detail::shared_spinlock> guard{workers_mtx_};
if (filter(guard, what->sender, what->mid, *what, eu)) if (filter(guard, what->sender, what->mid, what->payload, eu))
return; return;
policy_(home_system(), guard, workers_, what, eu); policy_(home_system(), guard, workers_, what, eu);
} }
...@@ -145,22 +145,20 @@ void actor_pool::on_cleanup(const error& reason) { ...@@ -145,22 +145,20 @@ void actor_pool::on_cleanup(const error& reason) {
bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard, bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
const strong_actor_ptr& sender, message_id mid, const strong_actor_ptr& sender, message_id mid,
message_view& mv, execution_unit* eu) { message& content, execution_unit* eu) {
auto& content = mv.content();
CAF_LOG_TRACE(CAF_ARG(mid) << CAF_ARG(content)); CAF_LOG_TRACE(CAF_ARG(mid) << CAF_ARG(content));
if (content.match_elements<exit_msg>()) { if (content.match_elements<exit_msg>()) {
// acquire second mutex as well // acquire second mutex as well
std::vector<actor> workers; std::vector<actor> workers;
auto em = content.get_as<exit_msg>(0).reason; auto em = content.get_as<exit_msg>(0).reason;
if (cleanup(std::move(em), eu)) { if (cleanup(std::move(em), eu)) {
auto tmp = mv.move_content_to_message();
// send exit messages *always* to all workers and clear vector afterwards // send exit messages *always* to all workers and clear vector afterwards
// but first swap workers_ out of the critical section // but first swap workers_ out of the critical section
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard}; upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
workers_.swap(workers); workers_.swap(workers);
unique_guard.unlock(); unique_guard.unlock();
for (auto& w : workers) for (auto& w : workers)
anon_send(w, tmp); anon_send(w, content);
unregister_from_system(); unregister_from_system();
} }
return true; return true;
......
...@@ -195,7 +195,7 @@ blocking_actor::mailbox_visitor::operator()(mailbox_element& x) { ...@@ -195,7 +195,7 @@ blocking_actor::mailbox_visitor::operator()(mailbox_element& x) {
return check_if_done(); return check_if_done();
case match_result::no_match: { // Blocking actors can have fallback case match_result::no_match: { // Blocking actors can have fallback
// handlers for catch-all rules. // handlers for catch-all rules.
auto sres = bhvr.fallback(*self->current_element_); auto sres = bhvr.fallback(self->current_element_->payload);
if (sres.flag != rt_skip) { if (sres.flag != rt_skip) {
visitor.visit(sres); visitor.visit(sres);
return check_if_done(); return check_if_done();
...@@ -204,10 +204,9 @@ blocking_actor::mailbox_visitor::operator()(mailbox_element& x) { ...@@ -204,10 +204,9 @@ 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.is_response()) { if (mid.is_response()) {
auto err = make_error(sec::unexpected_response, auto err = make_error(sec::unexpected_response, std::move(x.payload));
x.move_content_to_message()); mailbox_element tmp{std::move(x.sender), x.mid, std::move(x.stages),
mailbox_element_view<error> tmp{std::move(x.sender), x.mid, make_message(std::move(err))};
std::move(x.stages), err};
self->current_element_ = &tmp; self->current_element_ = &tmp;
bhvr.nested(tmp.content()); bhvr.nested(tmp.content());
return check_if_done(); return check_if_done();
......
...@@ -39,8 +39,7 @@ struct splitter_state { ...@@ -39,8 +39,7 @@ struct splitter_state {
behavior fan_out_fan_in(stateful_actor<splitter_state>* self, behavior fan_out_fan_in(stateful_actor<splitter_state>* self,
const std::vector<strong_actor_ptr>& workers) { const std::vector<strong_actor_ptr>& workers) {
auto f = [=](local_actor*, message_view& x) -> result<message> { auto f = [=](local_actor*, message& msg) -> result<message> {
auto msg = x.move_content_to_message();
self->state.rp = self->make_response_promise(); self->state.rp = self->make_response_promise();
self->state.pending = workers.size(); self->state.pending = workers.size();
// request().await() has LIFO ordering // request().await() has LIFO ordering
......
...@@ -28,7 +28,7 @@ blocking_behavior::blocking_behavior(behavior& x) : nested(x) { ...@@ -28,7 +28,7 @@ blocking_behavior::blocking_behavior(behavior& x) : nested(x) {
// nop // nop
} }
result<message> blocking_behavior::fallback(message_view&) { result<message> blocking_behavior::fallback(message&) {
return skip; return skip;
} }
......
...@@ -58,8 +58,8 @@ void forwarding_actor_proxy::enqueue(mailbox_element_ptr what, ...@@ -58,8 +58,8 @@ void forwarding_actor_proxy::enqueue(mailbox_element_ptr what,
execution_unit*) { execution_unit*) {
CAF_PUSH_AID(0); CAF_PUSH_AID(0);
CAF_ASSERT(what); CAF_ASSERT(what);
forward_msg(std::move(what->sender), what->mid, forward_msg(std::move(what->sender), what->mid, std::move(what->payload),
what->move_content_to_message(), &what->stages); &what->stages);
} }
bool forwarding_actor_proxy::add_backlink(abstract_actor* x) { bool forwarding_actor_proxy::add_backlink(abstract_actor* x) {
......
...@@ -161,8 +161,8 @@ public: ...@@ -161,8 +161,8 @@ public:
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// instead of dropping "unexpected" messages, // instead of dropping "unexpected" messages,
// we simply forward them to our acquaintances // we simply forward them to our acquaintances
auto fwd = [=](scheduled_actor*, message_view& x) -> result<message> { auto fwd = [=](scheduled_actor*, message& msg) -> result<message> {
send_to_acquaintances(x.move_content_to_message()); send_to_acquaintances(std::move(msg));
return message{}; return message{};
}; };
set_default_handler(fwd); set_default_handler(fwd);
...@@ -313,9 +313,9 @@ behavior proxy_broker::make_behavior() { ...@@ -313,9 +313,9 @@ behavior proxy_broker::make_behavior() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// instead of dropping "unexpected" messages, // instead of dropping "unexpected" messages,
// we simply forward them to our acquaintances // we simply forward them to our acquaintances
auto fwd = [=](local_actor*, message_view& x) -> result<message> { auto fwd = [=](local_actor*, message& msg) -> result<message> {
group_->send_all_subscribers(current_element_->sender, group_->send_all_subscribers(current_element_->sender, std::move(msg),
x.move_content_to_message(), context()); context());
return message{}; return message{};
}; };
set_default_handler(fwd); set_default_handler(fwd);
......
...@@ -18,80 +18,60 @@ ...@@ -18,80 +18,60 @@
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/message_builder.hpp" #include <memory>
namespace caf { namespace caf {
namespace { namespace {
message_id dynamic_category_correction(const message& msg, message_id mid) { /// Corrects the message ID for down- and upstream messages to make sure the
if (msg.match_elements<downstream_msg>()) /// category for a `mailbox_element` matches its content.
return mailbox_category_corrector<downstream_msg>::apply(mid); template <class...>
if (msg.match_elements<upstream_msg>()) struct mailbox_category_corrector {
return mailbox_category_corrector<upstream_msg>::apply(mid); static constexpr message_id apply(message_id x) noexcept {
return mid; return x;
}
/// Wraps a `message` into a mailbox element.
class mailbox_element_wrapper final : public mailbox_element {
public:
mailbox_element_wrapper(strong_actor_ptr&& x0, message_id x1,
forwarding_stack&& x2, message&& x3)
: mailbox_element(std::move(x0), dynamic_category_correction(x3, x1),
std::move(x2)),
msg_(std::move(x3)) {
/// Make sure that `content` can access the pointer safely.
if (msg_.vals() == nullptr) {
message_builder mb;
msg_ = mb.to_message();
}
}
type_erased_tuple& content() override {
return msg_;
}
const type_erased_tuple& content() const override {
return msg_.content();
} }
};
message move_content_to_message() override { template <>
return std::move(msg_); struct mailbox_category_corrector<downstream_msg> {
static constexpr message_id apply(message_id x) noexcept {
return x.with_category(message_id::downstream_message_category);
} }
};
message copy_content_to_message() const override { template <>
return msg_; struct mailbox_category_corrector<upstream_msg> {
static constexpr message_id apply(message_id x) noexcept {
return x.with_category(message_id::upstream_message_category);
} }
private:
/// Stores the content of this mailbox element.
message msg_;
}; };
} // namespace message_id dynamic_category_correction(const message& msg, message_id mid) {
if (msg.match_elements<downstream_msg>())
mailbox_element::mailbox_element() { return mailbox_category_corrector<downstream_msg>::apply(mid);
// nop if (msg.match_elements<upstream_msg>())
return mailbox_category_corrector<upstream_msg>::apply(mid);
return mid;
} }
mailbox_element::mailbox_element(strong_actor_ptr&& x, message_id y, } // namespace
forwarding_stack&& z)
:sender(std::move(x)),
mid(y),
stages(std::move(z)) {
// nop
}
mailbox_element::~mailbox_element() { mailbox_element::mailbox_element(strong_actor_ptr sender, message_id mid,
forwarding_stack stages, message payload)
: sender(std::move(sender)),
mid(dynamic_category_correction(payload, mid)),
stages(std::move(stages)),
payload(std::move(payload)) {
// nop // nop
} }
mailbox_element_ptr mailbox_element_ptr
make_mailbox_element(strong_actor_ptr sender, message_id id, make_mailbox_element(strong_actor_ptr sender, message_id id,
mailbox_element::forwarding_stack stages, message msg) { mailbox_element::forwarding_stack stages,
auto ptr = new mailbox_element_wrapper(std::move(sender), id, message payload) {
std::move(stages), std::move(msg)); return std::make_unique<mailbox_element>(
return mailbox_element_ptr{ptr}; std::move(sender), id, std::move(stages), std::move(payload));
} }
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/message_view.hpp"
namespace caf {
message_view::~message_view() {
// nop
}
} // namespace caf
...@@ -43,7 +43,7 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) { ...@@ -43,7 +43,7 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
if (!pr.second(x.content())) { if (!pr.second(x.content())) {
// try again with error if first attempt failed // try again with error if first attempt failed
auto msg = make_message( auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message())); make_error(sec::unexpected_response, std::move(x.payload)));
pr.second(msg); pr.second(msg);
} }
awaited_responses_.pop_front(); awaited_responses_.pop_front();
...@@ -58,7 +58,7 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) { ...@@ -58,7 +58,7 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
if (!mrh->second(x.content())) { if (!mrh->second(x.content())) {
// try again with error if first attempt failed // try again with error if first attempt failed
auto msg = make_message( auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message())); make_error(sec::unexpected_response, std::move(x.payload)));
mrh->second(msg); mrh->second(msg);
} }
multiplexed_responses_.erase(mrh); multiplexed_responses_.erase(mrh);
...@@ -92,7 +92,7 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) { ...@@ -92,7 +92,7 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
setf(has_timeout_flag); setf(has_timeout_flag);
}); });
auto call_default_handler = [&] { auto call_default_handler = [&] {
auto sres = call_handler(default_handler_, this, x); auto sres = call_handler(default_handler_, this, x.payload);
switch (sres.flag) { switch (sres.flag) {
default: default:
break; break;
......
...@@ -36,25 +36,25 @@ namespace caf { ...@@ -36,25 +36,25 @@ namespace caf {
// -- related free functions --------------------------------------------------- // -- related free functions ---------------------------------------------------
result<message> reflect(scheduled_actor*, message_view& x) { result<message> reflect(scheduled_actor*, message& msg) {
return x.move_content_to_message(); return std::move(msg);
} }
result<message> reflect_and_quit(scheduled_actor* ptr, message_view& x) { result<message> reflect_and_quit(scheduled_actor* ptr, message& msg) {
error err = exit_reason::normal; error err = exit_reason::normal;
scheduled_actor::default_error_handler(ptr, err); scheduled_actor::default_error_handler(ptr, err);
return reflect(ptr, x); return reflect(ptr, msg);
} }
result<message> print_and_drop(scheduled_actor* ptr, message_view& x) { result<message> print_and_drop(scheduled_actor* ptr, message& msg) {
CAF_LOG_WARNING("unexpected message" << CAF_ARG(x.content())); CAF_LOG_WARNING("unexpected message:" << msg);
aout(ptr) << "*** unexpected message [id: " << ptr->id() aout(ptr) << "*** unexpected message [id: " << ptr->id()
<< ", name: " << ptr->name() << "]: " << x.content().stringify() << ", name: " << ptr->name() << "]: " << to_string(msg)
<< std::endl; << std::endl;
return sec::unexpected_message; return sec::unexpected_message;
} }
result<message> drop(scheduled_actor*, message_view&) { result<message> drop(scheduled_actor*, message&) {
return sec::unexpected_message; return sec::unexpected_message;
} }
...@@ -67,7 +67,7 @@ void silently_ignore(scheduled_actor*, T&) { ...@@ -67,7 +67,7 @@ void silently_ignore(scheduled_actor*, T&) {
// nop // nop
} }
result<message> drop_after_quit(scheduled_actor* self, message_view&) { result<message> drop_after_quit(scheduled_actor* self, message&) {
if (self->current_message_id().is_request()) if (self->current_message_id().is_request())
return make_error(sec::request_receiver_down); return make_error(sec::request_receiver_down);
return make_message(); return make_message();
...@@ -648,7 +648,7 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) { ...@@ -648,7 +648,7 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
if (!invoke(this, f, x)) { if (!invoke(this, f, x)) {
// try again with error if first attempt failed // try again with error if first attempt failed
auto msg = make_message( auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message())); make_error(sec::unexpected_response, std::move(x.payload)));
f(msg); f(msg);
} }
return invoke_message_result::consumed; return invoke_message_result::consumed;
...@@ -665,7 +665,7 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) { ...@@ -665,7 +665,7 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
if (!invoke(this, bhvr, x)) { if (!invoke(this, bhvr, x)) {
CAF_LOG_DEBUG("got unexpected_response, invoke as again as error"); CAF_LOG_DEBUG("got unexpected_response, invoke as again as error");
auto msg = make_message( auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message())); make_error(sec::unexpected_response, std::move(x.payload)));
bhvr(msg); bhvr(msg);
} }
return invoke_message_result::consumed; return invoke_message_result::consumed;
...@@ -689,7 +689,7 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) { ...@@ -689,7 +689,7 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
setf(has_timeout_flag); setf(has_timeout_flag);
}); });
auto call_default_handler = [&] { auto call_default_handler = [&] {
auto sres = call_handler(default_handler_, this, x); auto sres = call_handler(default_handler_, this, x.payload);
switch (sres.flag) { switch (sres.flag) {
default: default:
break; break;
...@@ -1119,7 +1119,7 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) { ...@@ -1119,7 +1119,7 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
}; };
// Utility for invoking the default handler. // Utility for invoking the default handler.
auto fallback = [&] { auto fallback = [&] {
auto sres = call_handler(default_handler_, this, x); auto sres = call_handler(default_handler_, this, x.payload);
switch (sres.flag) { switch (sres.flag) {
default: default:
CAF_LOG_DEBUG( CAF_LOG_DEBUG(
......
...@@ -23,7 +23,7 @@ ...@@ -23,7 +23,7 @@
namespace caf { namespace caf {
result<message> skip_t::skip_fun_impl(scheduled_actor*, message_view&) { result<message> skip_t::skip_fun_impl(scheduled_actor*, message&) {
return skip(); return skip();
} }
......
...@@ -47,8 +47,8 @@ CAF_TEST(catch_all) { ...@@ -47,8 +47,8 @@ CAF_TEST(catch_all) {
[](float) { [](float) {
CAF_FAIL("received unexpected float"); CAF_FAIL("received unexpected float");
}, },
others >> [](message_view& x) -> result<message> { others >> [](message& msg) -> result<message> {
CAF_CHECK_EQUAL(to_string(x.content()), "(42)"); CAF_CHECK_EQUAL(to_string(msg), "(42)");
return sec::unexpected_message; return sec::unexpected_message;
} }
); );
......
...@@ -91,7 +91,7 @@ public: ...@@ -91,7 +91,7 @@ public:
} }
void enqueue(mailbox_element_ptr what, execution_unit*) override { void enqueue(mailbox_element_ptr what, execution_unit*) override {
mbox.push_back(what->move_content_to_message()); mbox.push_back(std::move(what->payload));
} }
void attach(attachable_ptr) override { void attach(attachable_ptr) override {
......
...@@ -73,26 +73,6 @@ CAF_TEST(non_empty_message) { ...@@ -73,26 +73,6 @@ CAF_TEST(non_empty_message) {
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));
} }
CAF_TEST(message_roundtrip) {
auto msg = make_message(1, 2, 3);
auto msg_ptr = msg.cvals().get();
auto m1 = make_mailbox_element(nullptr, make_message_id(),
no_stages, std::move(msg));
auto msg2 = m1->move_content_to_message();
CAF_CHECK_EQUAL(msg2.cvals().get(), msg_ptr);
}
CAF_TEST(message_roundtrip_with_copy) {
auto msg = make_message(1, 2, 3);
auto msg_ptr = msg.cvals().get();
auto m1 = make_mailbox_element(nullptr, make_message_id(),
no_stages, std::move(msg));
auto msg2 = m1->copy_content_to_message();
auto msg3 = m1->move_content_to_message();
CAF_CHECK_EQUAL(msg2.cvals().get(), msg_ptr);
CAF_CHECK_EQUAL(msg3.cvals().get(), msg_ptr);
}
CAF_TEST(tuple) { 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);
...@@ -103,26 +83,6 @@ CAF_TEST(tuple) { ...@@ -103,26 +83,6 @@ CAF_TEST(tuple) {
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));
} }
CAF_TEST(move_tuple) {
auto m1 = make_mailbox_element(nullptr, make_message_id(),
no_stages, "hello", "world");
using strings = tuple<string, string>;
CAF_CHECK_EQUAL((fetch<string, string>(*m1)), strings("hello", "world"));
auto msg = m1->move_content_to_message();
CAF_CHECK_EQUAL((fetch<string, string>(msg)), strings("hello", "world"));
CAF_CHECK_EQUAL((fetch<string, string>(*m1)), strings("", ""));
}
CAF_TEST(copy_tuple) {
auto m1 = make_mailbox_element(nullptr, make_message_id(),
no_stages, "hello", "world");
using strings = tuple<string, string>;
CAF_CHECK_EQUAL((fetch<string, string>(*m1)), strings("hello", "world"));
auto msg = m1->copy_content_to_message();
CAF_CHECK_EQUAL((fetch<string, string>(msg)), strings("hello", "world"));
CAF_CHECK_EQUAL((fetch<string, string>(*m1)), strings("hello", "world"));
}
CAF_TEST(high_priority) { CAF_TEST(high_priority) {
auto m1 = make_mailbox_element(nullptr, auto m1 = make_mailbox_element(nullptr,
make_message_id(message_priority::high), make_message_id(message_priority::high),
......
...@@ -36,7 +36,8 @@ public: ...@@ -36,7 +36,8 @@ public:
broker_servant(handle_type x) broker_servant(handle_type x)
: hdl_(x), : hdl_(x),
value_(strong_actor_ptr{}, make_message_id(), value_(strong_actor_ptr{}, make_message_id(),
mailbox_element::forwarding_stack{}, SysMsgType{x, {}}) { mailbox_element::forwarding_stack{},
make_message(SysMsgType{x, {}})) {
// nop // nop
} }
...@@ -103,9 +104,9 @@ protected: ...@@ -103,9 +104,9 @@ protected:
std::is_same<handle_type, accept_handle>::value, std::is_same<handle_type, accept_handle>::value,
acceptor_passivated_msg, acceptor_passivated_msg,
datagram_servant_passivated_msg>::type>::type; datagram_servant_passivated_msg>::type>::type;
using tmp_t = mailbox_element_vals<passiv_t>; mailbox_element tmp{strong_actor_ptr{}, make_message_id(),
tmp_t tmp{strong_actor_ptr{}, make_message_id(), mailbox_element::forwarding_stack{},
mailbox_element::forwarding_stack{}, passiv_t{hdl()}}; make_message(passiv_t{hdl()})};
invoke_mailbox_element_impl(ctx, tmp); invoke_mailbox_element_impl(ctx, tmp);
return activity_tokens_ != size_t{0}; return activity_tokens_ != size_t{0};
} }
...@@ -113,11 +114,11 @@ protected: ...@@ -113,11 +114,11 @@ protected:
} }
SysMsgType& msg() { SysMsgType& msg() {
return value_.template get_mutable_as<SysMsgType>(0); return value_.payload.template get_mutable_as<SysMsgType>(0);
} }
handle_type hdl_; handle_type hdl_;
mailbox_element_vals<SysMsgType> value_; mailbox_element value_;
optional<size_t> activity_tokens_; optional<size_t> activity_tokens_;
}; };
......
...@@ -65,10 +65,9 @@ void datagram_servant::datagram_sent(execution_unit* ctx, datagram_handle hdl, ...@@ -65,10 +65,9 @@ void datagram_servant::datagram_sent(execution_unit* ctx, datagram_handle hdl,
if (detached()) if (detached())
return; return;
using sent_t = datagram_sent_msg; using sent_t = datagram_sent_msg;
using tmp_t = mailbox_element_vals<datagram_sent_msg>; mailbox_element tmp{strong_actor_ptr{}, make_message_id(),
tmp_t tmp{strong_actor_ptr{}, make_message_id(),
mailbox_element::forwarding_stack{}, mailbox_element::forwarding_stack{},
sent_t{hdl, written, std::move(buffer)}}; make_message(sent_t{hdl, written, std::move(buffer)})};
invoke_mailbox_element_impl(ctx, tmp); invoke_mailbox_element_impl(ctx, tmp);
} }
......
...@@ -65,10 +65,9 @@ void scribe::data_transferred(execution_unit* ctx, size_t written, ...@@ -65,10 +65,9 @@ void scribe::data_transferred(execution_unit* ctx, size_t written,
if (detached()) if (detached())
return; return;
using transferred_t = data_transferred_msg; using transferred_t = data_transferred_msg;
using tmp_t = mailbox_element_vals<data_transferred_msg>; mailbox_element tmp{strong_actor_ptr{}, make_message_id(),
tmp_t tmp{strong_actor_ptr{}, make_message_id(),
mailbox_element::forwarding_stack{}, mailbox_element::forwarding_stack{},
transferred_t{hdl(), written, remaining}}; make_message(transferred_t{hdl(), written, remaining})};
invoke_mailbox_element_impl(ctx, tmp); invoke_mailbox_element_impl(ctx, tmp);
// data_transferred_msg tmp{hdl(), written, remaining}; // data_transferred_msg tmp{hdl(), written, remaining};
// auto ptr = make_mailbox_element(nullptr, invalid_message_id, {}, tmp); // auto ptr = make_mailbox_element(nullptr, invalid_message_id, {}, tmp);
......
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