Commit 01bd0842 authored by Dominik Charousset's avatar Dominik Charousset

Introduce message_view for default handlers

parent 3acc6e10
...@@ -68,6 +68,7 @@ set (LIBCAF_CORE_SRCS ...@@ -68,6 +68,7 @@ set (LIBCAF_CORE_SRCS
src/message_builder.cpp src/message_builder.cpp
src/message_data.cpp src/message_data.cpp
src/message_handler.cpp src/message_handler.cpp
src/message_view.cpp
src/node_id.cpp src/node_id.cpp
src/parse_ini.cpp src/parse_ini.cpp
src/private_thread.cpp src/private_thread.cpp
......
...@@ -38,6 +38,7 @@ ...@@ -38,6 +38,7 @@
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/abstract_channel.hpp" #include "caf/abstract_channel.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/detail/functor_attachable.hpp" #include "caf/detail/functor_attachable.hpp"
......
...@@ -108,7 +108,7 @@ protected: ...@@ -108,7 +108,7 @@ protected:
private: private:
bool filter(upgrade_lock<detail::shared_spinlock>&, bool filter(upgrade_lock<detail::shared_spinlock>&,
const strong_actor_ptr& sender, message_id mid, const strong_actor_ptr& sender, message_id mid,
const type_erased_tuple& content, execution_unit* host); message_view& content, execution_unit* host);
// call without workers_mtx_ held // call without workers_mtx_ held
void quit(execution_unit* host); void quit(execution_unit* host);
......
...@@ -27,7 +27,7 @@ namespace caf { ...@@ -27,7 +27,7 @@ namespace caf {
template <class F> template <class F>
struct catch_all { struct catch_all {
using fun_type = std::function<result<message> (const type_erased_tuple*)>; using fun_type = std::function<result<message> (message_view&)>;
F handler; F handler;
...@@ -40,7 +40,7 @@ struct catch_all { ...@@ -40,7 +40,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> (const type_erased_tuple*)"); "result<message> (message_view&)");
fun_type lift() const { fun_type lift() const {
return handler; return handler;
......
...@@ -40,6 +40,12 @@ auto apply_args(F& f, detail::int_list<Is...>, Tuple& tup) ...@@ -40,6 +40,12 @@ auto apply_args(F& f, detail::int_list<Is...>, Tuple& tup)
return f(get<Is>(tup)...); return f(get<Is>(tup)...);
} }
template <class F, long... Is, class Tuple>
auto apply_moved_args(F& f, detail::int_list<Is...>, Tuple& tup)
-> decltype(f(std::move(get<Is>(tup))...)) {
return f(std::move(get<Is>(tup))...);
}
template <class F, class Tuple, class... Ts> template <class F, class Tuple, class... Ts>
auto apply_args_prefixed(F& f, detail::int_list<>, Tuple&, Ts&&... xs) auto apply_args_prefixed(F& f, detail::int_list<>, Tuple&, Ts&&... xs)
-> decltype(f(std::forward<Ts>(xs)...)) { -> decltype(f(std::forward<Ts>(xs)...)) {
......
...@@ -36,7 +36,7 @@ public: ...@@ -36,7 +36,7 @@ public:
virtual ~blocking_behavior(); virtual ~blocking_behavior();
virtual result<message> fallback(const type_erased_tuple*); virtual result<message> fallback(message_view&);
virtual duration timeout(); virtual duration timeout();
...@@ -56,7 +56,7 @@ public: ...@@ -56,7 +56,7 @@ public:
blocking_behavior_v2(blocking_behavior_v2&&) = default; blocking_behavior_v2(blocking_behavior_v2&&) = default;
result<message> fallback(const type_erased_tuple* x) override { result<message> fallback(message_view& x) override {
return f.handler(x); return f.handler(x);
} }
}; };
...@@ -98,7 +98,7 @@ public: ...@@ -98,7 +98,7 @@ public:
blocking_behavior_v4(blocking_behavior_v4&&) = default; blocking_behavior_v4(blocking_behavior_v4&&) = default;
result<message> fallback(const type_erased_tuple* x) override { result<message> fallback(message_view& x) override {
return f1.handler(x); return f1.handler(x);
} }
......
...@@ -49,15 +49,15 @@ public: ...@@ -49,15 +49,15 @@ public:
} }
behavior make_behavior() override { behavior make_behavior() override {
auto f = [=](scheduled_actor*, const type_erased_tuple& tup) -> result<message> { auto f = [=](scheduled_actor*, message_view& xs) -> result<message> {
auto msg = message::from(&tup); 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 = [=](scheduled_actor*, const type_erased_tuple& x) mutable auto g = [=](scheduled_actor*, message_view& ys) mutable
-> result<message> { -> result<message> {
auto res = message::from(&x); 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_);
......
...@@ -84,8 +84,8 @@ struct function_view_storage_catch_all { ...@@ -84,8 +84,8 @@ struct function_view_storage_catch_all {
// nop // nop
} }
result<message> operator()(const type_erased_tuple* x) { result<message> operator()(message_view& xs) {
*storage_ = message::from(x); *storage_ = xs.move_content_to_message();
return message{}; return message{};
} }
}; };
......
...@@ -67,6 +67,7 @@ class local_actor; ...@@ -67,6 +67,7 @@ class local_actor;
class actor_config; class actor_config;
class actor_system; class actor_system;
class deserializer; class deserializer;
class message_view;
class scoped_actor; class scoped_actor;
class abstract_actor; class abstract_actor;
class abstract_group; class abstract_group;
......
...@@ -26,18 +26,17 @@ ...@@ -26,18 +26,17 @@
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/message_view.hpp"
#include "caf/memory_managed.hpp"
#include "caf/type_erased_tuple.hpp" #include "caf/type_erased_tuple.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/detail/embedded.hpp"
#include "caf/detail/disposer.hpp" #include "caf/detail/disposer.hpp"
#include "caf/detail/tuple_vals.hpp" #include "caf/detail/tuple_vals.hpp"
#include "caf/detail/message_data.hpp"
#include "caf/detail/memory_cache_flag_type.hpp"
namespace caf { namespace caf {
class mailbox_element { class 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>;
...@@ -67,7 +66,9 @@ public: ...@@ -67,7 +66,9 @@ public:
virtual ~mailbox_element(); virtual ~mailbox_element();
virtual type_erased_tuple& content(); type_erased_tuple& content() override;
message move_content_to_message() override;
const type_erased_tuple& content() const; const type_erased_tuple& content() const;
...@@ -93,14 +94,24 @@ public: ...@@ -93,14 +94,24 @@ public:
template <class... Us> template <class... Us>
mailbox_element_vals(strong_actor_ptr&& sender, message_id id, mailbox_element_vals(strong_actor_ptr&& sender, message_id id,
forwarding_stack&& stages, Us&&... xs) forwarding_stack&& stages, Us&&... xs)
: mailbox_element(std::move(sender), id, std::move(stages)), : mailbox_element(std::move(sender), id, std::move(stages)),
detail::tuple_vals_impl<type_erased_tuple, Ts...>(std::forward<Us>(xs)...) { detail::tuple_vals_impl<type_erased_tuple, Ts...>(std::forward<Us>(xs)...) {
// nop // nop
} }
type_erased_tuple& content() { type_erased_tuple& content() {
return *this; return *this;
} }
message move_content_to_message() {
message_factory f;
auto& xs = this->data();
return detail::apply_moved_args(f, detail::get_indices(xs), xs);
}
void dispose() noexcept {
this->deref();
}
}; };
/// Provides a view for treating arbitrary data as message element. /// Provides a view for treating arbitrary data as message element.
...@@ -115,9 +126,15 @@ public: ...@@ -115,9 +126,15 @@ public:
// nop // nop
} }
type_erased_tuple& content() { type_erased_tuple& content() override {
return *this; 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);
}
}; };
/// @relates mailbox_element /// @relates mailbox_element
......
...@@ -171,7 +171,7 @@ public: ...@@ -171,7 +171,7 @@ public:
message tmp; message tmp;
intermediate_pseudo_tuple tup{xs.shared()}; intermediate_pseudo_tuple tup{xs.shared()};
if (is_manipulator && tup.shared_access) { if (is_manipulator && tup.shared_access) {
tmp = message::copy_from(&xs); tmp = message::copy(xs);
tup.shared_access = false; tup.shared_access = false;
for (size_t i = 0; i < tmp.size(); ++i) for (size_t i = 0; i < tmp.size(); ++i)
tup[i] = const_cast<void*>(tmp.at(i)); tup[i] = const_cast<void*>(tmp.at(i));
......
...@@ -85,11 +85,8 @@ public: ...@@ -85,11 +85,8 @@ public:
return concat_impl({xs.vals()...}); return concat_impl({xs.vals()...});
} }
/// Creates a new message from a type-erased tuple.
static message from(const type_erased_tuple* ptr);
/// Creates a new message by copying all elements in a type-erased tuple. /// Creates a new message by copying all elements in a type-erased tuple.
static message copy_from(const type_erased_tuple* ptr); static message copy(const type_erased_tuple& xs);
// -- modifiers -------------------------------------------------------------- // -- modifiers --------------------------------------------------------------
...@@ -502,6 +499,13 @@ inline message make_message() { ...@@ -502,6 +499,13 @@ inline message make_message() {
return message{}; return message{};
} }
struct message_factory {
template <class... Ts>
message operator()(Ts&&... xs) const {
return make_message(std::forward<Ts>(xs)...);
}
};
} // namespace caf } // namespace caf
#endif // CAF_MESSAGE_HPP #endif // CAF_MESSAGE_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_MESSAGE_VIEW_HPP
#define CAF_MESSAGE_VIEW_HPP
#include "caf/fwd.hpp"
namespace caf {
/// Represents an object pointing to a `type_erased_tuple` that
/// is convertible to a `message`
class message_view {
public:
virtual ~message_view();
virtual type_erased_tuple& content() = 0;
virtual message move_content_to_message() = 0;
};
} // namespace caf
#endif // CAF_MESSAGE_VIEW_HPP
...@@ -48,21 +48,21 @@ namespace caf { ...@@ -48,21 +48,21 @@ 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.
result<message> reflect(scheduled_actor*, const type_erased_tuple&); result<message> reflect(scheduled_actor*, message_view&);
/// @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.
result<message> reflect_and_quit(scheduled_actor*, const type_erased_tuple&); result<message> reflect_and_quit(scheduled_actor*, message_view&);
/// @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.
result<message> print_and_drop(scheduled_actor*, const type_erased_tuple&); result<message> print_and_drop(scheduled_actor*, message_view&);
/// @relates scheduled_actor /// @relates scheduled_actor
/// Default handler function that simply drops messages. /// Default handler function that simply drops messages.
result<message> drop(scheduled_actor*, const type_erased_tuple&); result<message> drop(scheduled_actor*, message_view&);
/// A cooperatively scheduled, event-based actor implementation. This is the /// A cooperatively scheduled, event-based actor implementation. This is the
/// recommended base class for user-defined actors. /// recommended base class for user-defined actors.
...@@ -77,11 +77,8 @@ public: ...@@ -77,11 +77,8 @@ public:
/// A pointer to a scheduled actor. /// A pointer to a scheduled actor.
using pointer = scheduled_actor*; using pointer = scheduled_actor*;
/// A constant pointer to a `type_erased_tuple`.
using tuple_cref = const type_erased_tuple&;
/// Function object for handling unmatched messages. /// Function object for handling unmatched messages.
using default_handler = std::function<result<message> (pointer, tuple_cref)>; using default_handler = std::function<result<message> (pointer, message_view&)>;
/// 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&)>;
......
...@@ -32,8 +32,8 @@ namespace caf { ...@@ -32,8 +32,8 @@ namespace caf {
/// skipping to the runtime. /// skipping to the runtime.
class skip_t { class skip_t {
public: public:
using fun = std::function<result<message>(scheduled_actor* self, using fun = std::function<result<message> (scheduled_actor* self,
const type_erased_tuple&)>; message_view&)>;
constexpr skip_t() { constexpr skip_t() {
// nop // nop
...@@ -46,8 +46,7 @@ public: ...@@ -46,8 +46,7 @@ public:
operator fun() const; operator fun() const;
private: private:
static result<message> skip_fun_impl(scheduled_actor*, static result<message> skip_fun_impl(scheduled_actor*, message_view&);
const type_erased_tuple&);
}; };
/// Tells the runtime system to skip a message when used as message /// Tells the runtime system to skip a message when used as message
......
...@@ -237,6 +237,8 @@ public: ...@@ -237,6 +237,8 @@ public:
template <size_t X> template <size_t X>
using num_token = std::integral_constant<size_t, X>; using num_token = std::integral_constant<size_t, X>;
using tuple_type = std::tuple<type_erased_value_impl<std::reference_wrapper<Ts>>...>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
type_erased_tuple_view(Ts&... xs) : xs_(xs...) { type_erased_tuple_view(Ts&... xs) : xs_(xs...) {
...@@ -289,6 +291,16 @@ public: ...@@ -289,6 +291,16 @@ public:
return ptrs_[pos]->save(sink); return ptrs_[pos]->save(sink);
} }
// -- member variables access ------------------------------------------------
tuple_type& data() {
return xs_;
}
const tuple_type& data() const {
return xs_;
}
private: private:
// -- pointer "lookup table" utility ----------------------------------------- // -- pointer "lookup table" utility -----------------------------------------
...@@ -309,10 +321,12 @@ private: ...@@ -309,10 +321,12 @@ private:
// -- data members ----------------------------------------------------------- // -- data members -----------------------------------------------------------
std::tuple<type_erased_value_impl<std::reference_wrapper<Ts>>...> xs_; tuple_type xs_;
type_erased_value* ptrs_[sizeof...(Ts) == 0 ? 1 : sizeof...(Ts)]; type_erased_value* ptrs_[sizeof...(Ts) == 0 ? 1 : sizeof...(Ts)];
}; };
template <class... Ts> template <class... Ts>
type_erased_tuple_view<Ts...> make_type_erased_tuple_view(Ts&... xs) { type_erased_tuple_view<Ts...> make_type_erased_tuple_view(Ts&... xs) {
return {xs...}; return {xs...};
......
...@@ -200,6 +200,12 @@ public: ...@@ -200,6 +200,12 @@ public:
return type_erased_value_ptr{new type_erased_value_impl(x_)}; return type_erased_value_ptr{new type_erased_value_impl(x_)};
} }
// -- conversion operators ---------------------------------------------------
operator value_type&() {
return x_;
}
private: private:
// -- address-of-member utility ---------------------------------------------- // -- address-of-member utility ----------------------------------------------
......
...@@ -39,6 +39,7 @@ ...@@ -39,6 +39,7 @@
#include "caf/default_attachable.hpp" #include "caf/default_attachable.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
namespace caf { namespace caf {
......
...@@ -25,6 +25,8 @@ ...@@ -25,6 +25,8 @@
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/detail/disposer.hpp"
namespace caf { namespace caf {
actor_addr actor_control_block::address() { actor_addr actor_control_block::address() {
......
...@@ -54,11 +54,10 @@ namespace { ...@@ -54,11 +54,10 @@ namespace {
void broadcast_dispatch(actor_system&, actor_pool::uplock&, 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 = message::from(&ptr->content()); auto msg = ptr->move_content_to_message();
for (size_t i = 1; i < vec.size(); ++i) for (auto& worker : vec)
vec[i]->enqueue(ptr->sender, ptr->mid, msg, host); worker->enqueue(ptr->sender, ptr->mid, msg, host);
vec.front()->enqueue(std::move(ptr), host);
} }
} // namespace <anonymous> } // namespace <anonymous>
...@@ -67,7 +66,6 @@ actor_pool::policy actor_pool::broadcast() { ...@@ -67,7 +66,6 @@ actor_pool::policy actor_pool::broadcast() {
return broadcast_dispatch; return broadcast_dispatch;
} }
actor_pool::policy actor_pool::random() { actor_pool::policy actor_pool::random() {
struct impl { struct impl {
impl() : rd_() { impl() : rd_() {
...@@ -103,12 +101,6 @@ actor actor_pool::make(execution_unit* eu, policy pol) { ...@@ -103,12 +101,6 @@ actor actor_pool::make(execution_unit* eu, policy pol) {
auto ptr = static_cast<actor_pool*>(actor_cast<abstract_actor*>(res)); auto ptr = static_cast<actor_pool*>(actor_cast<abstract_actor*>(res));
ptr->policy_ = std::move(pol); ptr->policy_ = std::move(pol);
return res; return res;
/*
intrusive_ptr<actor_pool> ptr;
ptr = make_counted<actor_pool>(cfg);
ptr->policy_ = std::move(pol);
return actor_cast<actor>(ptr);
*/
} }
actor actor_pool::make(execution_unit* eu, size_t num_workers, actor actor_pool::make(execution_unit* eu, size_t num_workers,
...@@ -126,7 +118,7 @@ actor actor_pool::make(execution_unit* eu, size_t num_workers, ...@@ -126,7 +118,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->content(), eu)) if (filter(guard, what->sender, what->mid, *what, eu))
return; return;
policy_(home_system(), guard, workers_, what, eu); policy_(home_system(), guard, workers_, what, eu);
} }
...@@ -141,14 +133,15 @@ void actor_pool::on_cleanup() { ...@@ -141,14 +133,15 @@ void actor_pool::on_cleanup() {
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,
const type_erased_tuple& msg, execution_unit* eu) { message_view& mv, execution_unit* eu) {
CAF_LOG_TRACE(CAF_ARG(mid) << CAF_ARG(msg)); auto& content = mv.content();
if (msg.match_elements<exit_msg>()) { CAF_LOG_TRACE(CAF_ARG(mid) << CAF_ARG(content));
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 tmp = msg.get_as<exit_msg>(0).reason; auto em = content.get_as<exit_msg>(0).reason;
if (cleanup(std::move(tmp), eu)) { if (cleanup(std::move(em), eu)) {
auto tmp = message::from(&msg); 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};
...@@ -160,9 +153,9 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard, ...@@ -160,9 +153,9 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
} }
return true; return true;
} }
if (msg.match_elements<down_msg>()) { if (content.match_elements<down_msg>()) {
// remove failed worker from pool // remove failed worker from pool
auto& dm = msg.get_as<down_msg>(0); auto& dm = content.get_as<down_msg>(0);
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard}; upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
auto last = workers_.end(); auto last = workers_.end();
auto i = std::find(workers_.begin(), workers_.end(), dm.source); auto i = std::find(workers_.begin(), workers_.end(), dm.source);
...@@ -176,17 +169,17 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard, ...@@ -176,17 +169,17 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
} }
return true; return true;
} }
if (msg.match_elements<sys_atom, put_atom, actor>()) { if (content.match_elements<sys_atom, put_atom, actor>()) {
auto& worker = msg.get_as<actor>(2); auto& worker = content.get_as<actor>(2);
worker->attach(default_attachable::make_monitor(worker.address(), worker->attach(default_attachable::make_monitor(worker.address(),
address())); address()));
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard}; upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
workers_.push_back(worker); workers_.push_back(worker);
return true; return true;
} }
if (msg.match_elements<sys_atom, delete_atom, actor>()) { if (content.match_elements<sys_atom, delete_atom, actor>()) {
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard}; upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
auto& what = msg.get_as<actor>(2); auto& what = content.get_as<actor>(2);
auto last = workers_.end(); auto last = workers_.end();
auto i = std::find(workers_.begin(), last, what); auto i = std::find(workers_.begin(), last, what);
if (i != last) { if (i != last) {
...@@ -194,7 +187,7 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard, ...@@ -194,7 +187,7 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
} }
return true; return true;
} }
if (msg.match_elements<sys_atom, get_atom>()) { if (content.match_elements<sys_atom, get_atom>()) {
auto cpy = workers_; auto cpy = workers_;
guard.unlock(); guard.unlock();
sender->enqueue(nullptr, mid.response_id(), sender->enqueue(nullptr, mid.response_id(),
......
...@@ -63,8 +63,7 @@ void adapter::enqueue(mailbox_element_ptr x, execution_unit* context) { ...@@ -63,8 +63,7 @@ void adapter::enqueue(mailbox_element_ptr x, execution_unit* context) {
bounce(x, fail_state); bounce(x, fail_state);
return; return;
} }
message tmp{detail::merged_tuple::make(merger, message tmp{detail::merged_tuple::make(merger, x->move_content_to_message())};
message::from(&x->content()))};
decorated->enqueue(make_mailbox_element(std::move(x->sender), x->mid, decorated->enqueue(make_mailbox_element(std::move(x->sender), x->mid,
std::move(x->stages), std::move(tmp)), std::move(x->stages), std::move(tmp)),
context); context);
......
...@@ -131,6 +131,9 @@ namespace { ...@@ -131,6 +131,9 @@ namespace {
class message_sequence { class message_sequence {
public: public:
virtual ~message_sequence() {
// nop
}
virtual bool at_end() = 0; virtual bool at_end() = 0;
virtual bool await_value(bool reset_timeout) = 0; virtual bool await_value(bool reset_timeout) = 0;
virtual mailbox_element& value() = 0; virtual mailbox_element& value() = 0;
...@@ -335,7 +338,7 @@ void blocking_actor::receive_impl(receive_cond& rcc, ...@@ -335,7 +338,7 @@ void blocking_actor::receive_impl(receive_cond& rcc,
default: default:
break; break;
case match_case::no_match: { case match_case::no_match: {
auto sres = bhvr.fallback(&current_element_->content()); auto sres = bhvr.fallback(*current_element_);
// when dealing with response messages, there's either a match // when dealing with response messages, there's either a match
// on the first handler or we produce an error to // on the first handler or we produce an error to
// get a match on the second (error) handler // get a match on the second (error) handler
...@@ -345,7 +348,7 @@ void blocking_actor::receive_impl(receive_cond& rcc, ...@@ -345,7 +348,7 @@ void blocking_actor::receive_impl(receive_cond& rcc,
// invoke again with an unexpected_response error // invoke again with an unexpected_response error
auto& old = *current_element_; auto& old = *current_element_;
auto err = make_error(sec::unexpected_response, auto err = make_error(sec::unexpected_response,
message::from(&old.content())); old.move_content_to_message());
mailbox_element_view<error> tmp{std::move(old.sender), old.mid, mailbox_element_view<error> tmp{std::move(old.sender), old.mid,
std::move(old.stages), err}; std::move(old.stages), err};
current_element_ = &tmp; current_element_ = &tmp;
......
...@@ -30,7 +30,7 @@ blocking_behavior::blocking_behavior(behavior x) : nested(std::move(x)) { ...@@ -30,7 +30,7 @@ blocking_behavior::blocking_behavior(behavior x) : nested(std::move(x)) {
// nop // nop
} }
result<message> blocking_behavior::fallback(const type_erased_tuple*) { result<message> blocking_behavior::fallback(message_view&) {
return skip; return skip;
} }
......
...@@ -68,7 +68,7 @@ void forwarding_actor_proxy::enqueue(mailbox_element_ptr what, ...@@ -68,7 +68,7 @@ void forwarding_actor_proxy::enqueue(mailbox_element_ptr what,
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,
message::from(&what->content()), &what->stages); what->move_content_to_message(), &what->stages);
} }
......
...@@ -156,8 +156,8 @@ public: ...@@ -156,8 +156,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*, const type_erased_tuple& x) -> result<message> { auto fwd = [=](scheduled_actor*, message_view& x) -> result<message> {
send_to_acquaintances(message::from(&x)); send_to_acquaintances(x.move_content_to_message());
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*, const type_erased_tuple& x) -> result<message> { auto fwd = [=](local_actor*, message_view& x) -> result<message> {
group_->send_all_subscribers(current_element_->sender, message::from(&x), group_->send_all_subscribers(current_element_->sender,
context()); x.move_content_to_message(), context());
return message{}; return message{};
}; };
set_default_handler(fwd); set_default_handler(fwd);
......
...@@ -26,22 +26,27 @@ namespace { ...@@ -26,22 +26,27 @@ namespace {
/// Wraps a `message` into a mailbox element. /// Wraps a `message` into a mailbox element.
class mailbox_element_wrapper : public mailbox_element { class mailbox_element_wrapper : public mailbox_element {
public: public:
/// Stores the elements for this mailbox element.
message msg;
mailbox_element_wrapper(strong_actor_ptr&& x0, message_id x1, mailbox_element_wrapper(strong_actor_ptr&& x0, message_id x1,
forwarding_stack&& x2, message&& x3) forwarding_stack&& x2, message&& x3)
: mailbox_element(std::move(x0), x1, std::move(x2)), : mailbox_element(std::move(x0), x1, std::move(x2)),
msg(std::move(x3)) { msg_(std::move(x3)) {
// nop // nop
} }
type_erased_tuple& content() override { type_erased_tuple& content() override {
auto ptr = msg.vals().raw_ptr(); auto ptr = msg_.vals().raw_ptr();
if (ptr) if (ptr)
return *ptr; return *ptr;
return dummy_; return dummy_;
} }
message move_content_to_message() override {
return std::move(msg_);
}
private:
/// Stores the content of this mailbox element.
message msg_;
}; };
} // namespace <anonymous> } // namespace <anonymous>
...@@ -72,6 +77,10 @@ type_erased_tuple& mailbox_element::content() { ...@@ -72,6 +77,10 @@ type_erased_tuple& mailbox_element::content() {
return dummy_; return dummy_;
} }
message mailbox_element::move_content_to_message() {
return {};
}
const type_erased_tuple& mailbox_element::content() const { const type_erased_tuple& mailbox_element::content() const {
return const_cast<mailbox_element*>(this)->content(); return const_cast<mailbox_element*>(this)->content();
} }
......
...@@ -64,23 +64,10 @@ void* message::get_mutable(size_t p) { ...@@ -64,23 +64,10 @@ void* message::get_mutable(size_t p) {
return vals_->get_mutable(p); return vals_->get_mutable(p);
} }
message message::from(const type_erased_tuple* ptr) { message message::copy(const type_erased_tuple& xs) {
CAF_ASSERT(ptr != nullptr);
auto dptr = dynamic_cast<const detail::message_data*>(ptr);
if (dptr) {
data_ptr dp{const_cast<detail::message_data*>(dptr), true};
return message{std::move(dp)};
}
message_builder mb;
for (size_t i = 0; i < ptr->size(); ++i)
mb.emplace(ptr->copy(i));
return mb.move_to_message();
}
message message::copy_from(const type_erased_tuple* ptr) {
message_builder mb; message_builder mb;
for (size_t i = 0; i < ptr->size(); ++i) for (size_t i = 0; i < xs.size(); ++i)
mb.emplace(ptr->copy(i)); mb.emplace(xs.copy(i));
return mb.move_to_message(); return mb.move_to_message();
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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
...@@ -29,28 +29,26 @@ namespace caf { ...@@ -29,28 +29,26 @@ namespace caf {
// -- related free functions --------------------------------------------------- // -- related free functions ---------------------------------------------------
result<message> reflect(scheduled_actor*, const type_erased_tuple& x) { result<message> reflect(scheduled_actor*, message_view& x) {
return message::from(&x); return x.move_content_to_message();
} }
result<message> reflect_and_quit(scheduled_actor* ptr, result<message> reflect_and_quit(scheduled_actor* ptr, message_view& x) {
const type_erased_tuple& x) {
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, x);
} }
result<message> print_and_drop(scheduled_actor* ptr, result<message> print_and_drop(scheduled_actor* ptr, message_view& x) {
const type_erased_tuple& x) { CAF_LOG_WARNING("unexpected message" << CAF_ARG(x.content()));
CAF_LOG_WARNING("unexpected message" << CAF_ARG(x));
aout(ptr) << "*** unexpected message [id: " << ptr->id() aout(ptr) << "*** unexpected message [id: " << ptr->id()
<< ", name: " << ptr->name() << "]: " << ", name: " << ptr->name() << "]: "
<< x.stringify() << x.content().stringify()
<< std::endl; << std::endl;
return sec::unexpected_message; return sec::unexpected_message;
} }
result<message> drop(scheduled_actor*, const type_erased_tuple&) { result<message> drop(scheduled_actor*, message_view&) {
return sec::unexpected_message; return sec::unexpected_message;
} }
...@@ -360,7 +358,7 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) { ...@@ -360,7 +358,7 @@ invoke_message_result scheduled_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(make_error(sec::unexpected_response, auto msg = make_message(make_error(sec::unexpected_response,
message::from(&x.content()))); x.move_content_to_message()));
pr.second(msg); pr.second(msg);
} }
awaited_responses_.pop_front(); awaited_responses_.pop_front();
...@@ -375,7 +373,7 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) { ...@@ -375,7 +373,7 @@ invoke_message_result scheduled_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(make_error(sec::unexpected_response, auto msg = make_message(make_error(sec::unexpected_response,
message::from(&x.content()))); x.move_content_to_message()));
mrh->second(msg); mrh->second(msg);
} }
multiplexed_responses_.erase(mrh); multiplexed_responses_.erase(mrh);
...@@ -408,7 +406,7 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) { ...@@ -408,7 +406,7 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
has_timeout(true); has_timeout(true);
}); });
auto call_default_handler = [&] { auto call_default_handler = [&] {
auto sres = default_handler_(this, x.content()); auto sres = default_handler_(this, x);
switch (sres.flag) { switch (sres.flag) {
default: default:
break; break;
......
...@@ -24,8 +24,7 @@ ...@@ -24,8 +24,7 @@
namespace caf { namespace caf {
result<message> skip_t::skip_fun_impl(scheduled_actor*, result<message> skip_t::skip_fun_impl(scheduled_actor*, message_view&) {
const type_erased_tuple&) {
return skip(); return skip();
} }
......
...@@ -41,8 +41,8 @@ struct splitter_state { ...@@ -41,8 +41,8 @@ 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*, const type_erased_tuple& x) -> result<message> { auto f = [=](local_actor*, message_view& x) -> result<message> {
auto msg = message::from(&x); 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
...@@ -69,46 +69,6 @@ behavior fan_out_fan_in(stateful_actor<splitter_state>* self, ...@@ -69,46 +69,6 @@ behavior fan_out_fan_in(stateful_actor<splitter_state>* self,
return [] { return [] {
// nop // nop
}; };
/*
// TODO: maybe infer some useful timeout or use config parameter?
self->request(actor_cast<actor>(*i), infinite, msg)
.generic_await(
[=](const message& tmp) {
self->state.result += tmp;
if (--self->state.pending == 0)
self->state.rp.deliver(std::move(self->state.result));
},
[=](const error& err) {
self->state.rp.deliver(err);
self->quit();
}
);
};
set_default_handler(f);
return {
others >> [=](const message& msg) {
self->state.rp = self->make_response_promise();
self->state.pending = workers.size();
// request().await() has LIFO ordering
for (auto i = workers.rbegin(); i != workers.rend(); ++i)
// TODO: maybe infer some useful timeout or use config parameter?
self->request(actor_cast<actor>(*i), infinite, msg)
.generic_await(
[=](const message& tmp) {
self->state.result += tmp;
if (--self->state.pending == 0)
self->state.rp.deliver(std::move(self->state.result));
},
[=](const error& err) {
self->state.rp.deliver(err);
self->quit();
}
);
self->unbecome();
}
};
*/
} }
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -144,7 +144,9 @@ CAF_TEST(broadcast_actor_pool) { ...@@ -144,7 +144,9 @@ CAF_TEST(broadcast_actor_pool) {
return actor_pool::make(&context, 5, fixture::spawn_worker, return actor_pool::make(&context, 5, fixture::spawn_worker,
actor_pool::broadcast()); actor_pool::broadcast());
}; };
CAF_CHECK_EQUAL(system.registry().running(), 1u);
auto pool = actor_pool::make(&context, 5, spawn5, actor_pool::broadcast()); auto pool = actor_pool::make(&context, 5, spawn5, actor_pool::broadcast());
CAF_CHECK_EQUAL(system.registry().running(), 32u);
self->send(pool, 1, 2); self->send(pool, 1, 2);
std::vector<int> results; std::vector<int> results;
int i = 0; int i = 0;
......
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