Commit 1700bcf8 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Fix copy-on-write semantic for messages

parent 8a47221b
...@@ -156,7 +156,8 @@ actor_factory make_actor_factory(F fun) { ...@@ -156,7 +156,8 @@ actor_factory make_actor_factory(F fun) {
typename ctrait::arg_types>; typename ctrait::arg_types>;
fd f{fun, static_cast<impl*>(x)}; fd f{fun, static_cast<impl*>(x)};
empty_type_erased_tuple dummy_; empty_type_erased_tuple dummy_;
auto& ct = msg.empty() ? dummy_ : const_cast<message&>(msg).content(); auto& ct = msg.empty() ? dummy_
: const_cast<message&>(msg).content();
auto opt = ct.apply(f); auto opt = ct.apply(f);
if (!opt) if (!opt)
return {}; return {};
......
...@@ -122,7 +122,7 @@ public: ...@@ -122,7 +122,7 @@ public:
} }
inline explicit operator bool() const noexcept { inline explicit operator bool() const noexcept {
return static_cast<bool>(ptr_); return ptr_ != nullptr;
} }
inline message_data* get() const noexcept { inline message_data* get() const noexcept {
...@@ -134,6 +134,22 @@ private: ...@@ -134,6 +134,22 @@ private:
intrusive_ptr<message_data> ptr_; intrusive_ptr<message_data> ptr_;
}; };
inline bool operator==(const message_data::cow_ptr& x, std::nullptr_t) {
return x.get() == nullptr;
}
inline bool operator==(std::nullptr_t, const message_data::cow_ptr& x) {
return x.get() == nullptr;
}
inline bool operator!=(const message_data::cow_ptr& x, std::nullptr_t) {
return x.get() != nullptr;
}
inline bool operator!=(std::nullptr_t, const message_data::cow_ptr& x) {
return x.get() != nullptr;
}
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
......
...@@ -23,28 +23,29 @@ ...@@ -23,28 +23,29 @@
#include <sstream> #include <sstream>
#include <type_traits> #include <type_traits>
#include "caf/fwd.hpp"
#include "caf/skip.hpp"
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/none.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/optional.hpp" #include "caf/fwd.hpp"
#include "caf/make_counted.hpp"
#include "caf/index_mapping.hpp" #include "caf/index_mapping.hpp"
#include "caf/make_counted.hpp"
#include "caf/none.hpp"
#include "caf/optional.hpp"
#include "caf/skip.hpp"
#include "caf/type_nr.hpp"
#include "caf/detail/int_list.hpp"
#include "caf/detail/apply_args.hpp" #include "caf/detail/apply_args.hpp"
#include "caf/detail/comparable.hpp" #include "caf/detail/comparable.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/message_data.hpp"
#include "caf/detail/implicit_conversions.hpp" #include "caf/detail/implicit_conversions.hpp"
#include "caf/detail/int_list.hpp"
#include "caf/detail/message_data.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
class message_handler; class message_handler;
/// Describes a fixed-length, copy-on-write, type-erased /// Describes a fixed-length, copy-on-write, type-erased
/// tuple with elements of any type. /// tuple with elements of any type.
class message { class message : public type_erased_tuple {
public: public:
// -- nested types ----------------------------------------------------------- // -- nested types -----------------------------------------------------------
...@@ -74,7 +75,33 @@ public: ...@@ -74,7 +75,33 @@ public:
message& operator=(message&&) noexcept; message& operator=(message&&) noexcept;
explicit message(data_ptr ptr) noexcept; explicit message(data_ptr ptr) noexcept;
~message(); ~message() override;
// -- implementation of type_erased_tuple ------------------------------------
void* get_mutable(size_t p) override;
error load(size_t pos, deserializer& source) override;
size_t size() const noexcept override;
uint32_t type_token() const noexcept override;
rtti_pair type(size_t pos) const noexcept override;
const void* get(size_t pos) const noexcept override;
std::string stringify(size_t pos) const override;
type_erased_value_ptr copy(size_t pos) const override;
error save(size_t pos, serializer& sink) const override;
bool shared() const noexcept override;
error load(deserializer& source) override;
error save(serializer& sink) const override;
// -- factories -------------------------------------------------------------- // -- factories --------------------------------------------------------------
...@@ -92,16 +119,6 @@ public: ...@@ -92,16 +119,6 @@ public:
/// Concatenates `*this` and `x`. /// Concatenates `*this` and `x`.
message& operator+=(const message& x); message& operator+=(const message& x);
/// Returns a mutable pointer to the element at position `p`.
void* get_mutable(size_t p);
/// Returns the value at position `p` as mutable reference of type `T`.
template <class T>
T& get_mutable_as(size_t p) {
CAF_ASSERT(match_element(p, type_nr<T>::value, &typeid(T)));
return *reinterpret_cast<T*>(get_mutable(p));
}
/// Returns `handler(*this)`. /// Returns `handler(*this)`.
optional<message> apply(message_handler handler); optional<message> apply(message_handler handler);
...@@ -212,7 +229,7 @@ public: ...@@ -212,7 +229,7 @@ public:
/// Returns a const pointer to the element at position `p`. /// Returns a const pointer to the element at position `p`.
inline const void* at(size_t p) const noexcept { inline const void* at(size_t p) const noexcept {
CAF_ASSERT(vals_); CAF_ASSERT(vals_ != nullptr);
return vals_->get(p); return vals_->get(p);
} }
...@@ -226,21 +243,7 @@ public: ...@@ -226,21 +243,7 @@ public:
return vals_; return vals_;
} }
/// Returns a type hint for the pattern matching engine.
inline uint32_t type_token() const noexcept {
return vals_ ? vals_->type_token() : 0xFFFFFFFF;
}
/// Returns whether there are more than one references to the content.
inline bool shared() const noexcept {
return vals_ ? vals_->shared() : false;
}
/// Returns the size of this message. /// Returns the size of this message.
inline size_t size() const noexcept {
return vals_ ? vals_->size() : 0;
}
/// Creates a new message from the first n values. /// Creates a new message from the first n values.
inline message take(size_t n) const { inline message take(size_t n) const {
return n >= size() ? *this : drop_right(size() - n); return n >= size() ? *this : drop_right(size() - n);
...@@ -251,71 +254,24 @@ public: ...@@ -251,71 +254,24 @@ public:
return n >= size() ? *this : drop(size() - n); return n >= size() ? *this : drop(size() - n);
} }
/// Returns true if `size() == 0, otherwise false.
inline bool empty() const {
return size() == 0;
}
/// Returns the value at position `p` as const reference of type `T`.
template <class T>
const T& get_as(size_t p) const {
CAF_ASSERT(match_element(p, type_nr<T>::value, &typeid(T)));
return *reinterpret_cast<const T*>(at(p));
}
/// Queries whether the element at position `p` is of type `T`.
template <class T>
bool match_element(size_t p) const noexcept {
return vals_ ? vals_->match_element<T>(p) : false;
}
/// Queries whether the types of this message are `Ts...`.
template <class... Ts>
bool match_elements() const noexcept {
detail::type_list<Ts...> token;
return match_elements(token);
}
/// Queries the run-time type information for the element at position `pos`.
inline std::pair<uint16_t, const std::type_info*>
type(size_t pos) const noexcept {
CAF_ASSERT(vals_ && vals_->size() > pos);
return vals_->type(pos);
}
/// Checks whether the type of the stored value at position `pos`
/// matches type number `n` and run-time type information `p`.
bool match_element(size_t pos, uint16_t n,
const std::type_info* p) const noexcept {
CAF_ASSERT(vals_);
return vals_->matches(pos, n, p);
}
inline bool match_elements(detail::type_list<>) const noexcept {
return !vals_ || vals_->empty();
}
template <class T, class... Ts>
bool match_elements(detail::type_list<T, Ts...>) const noexcept {
return vals_ ? vals_->match_elements<T, Ts...>() : false;
}
/// @cond PRIVATE /// @cond PRIVATE
/// @pre `!empty()` /// @pre `!empty()`
inline type_erased_tuple& content() { inline type_erased_tuple& content() {
CAF_ASSERT(vals_); CAF_ASSERT(vals_ != nullptr);
return *vals_; return *vals_;
} }
inline const type_erased_tuple& content() const { inline const type_erased_tuple& content() const {
CAF_ASSERT(vals_); CAF_ASSERT(vals_ != nullptr);
return *vals_; return *vals_;
} }
/// @endcond /// @endcond
private: private:
// -- private helpers --------------------------------------------------------
template <size_t P> template <size_t P>
static bool match_elements_impl(std::integral_constant<size_t, P>, static bool match_elements_impl(std::integral_constant<size_t, P>,
detail::type_list<>) noexcept { detail::type_list<>) noexcept {
...@@ -334,17 +290,24 @@ private: ...@@ -334,17 +290,24 @@ private:
static message concat_impl(std::initializer_list<data_ptr> xs); static message concat_impl(std::initializer_list<data_ptr> xs);
// -- member functions -------------------------------------------------------
data_ptr vals_; data_ptr vals_;
}; };
/// @relates message // -- nested types -------------------------------------------------------------
error inspect(serializer& sink, message& msg);
/// @relates message /// Stores the result of `message::extract_opts`.
error inspect(deserializer& source, message& msg); struct message::cli_res {
/// Stores the remaining (unmatched) arguments.
/// @relates message message remainder;
std::string to_string(const message& msg); /// Stores the names of all active options.
std::set<std::string> opts;
/// Stores the automatically generated help text.
std::string helptext;
/// Stores errors during option parsing.
std::string error;
};
/// Stores the name of a command line option ("<long name>[,<short name>]") /// Stores the name of a command line option ("<long name>[,<short name>]")
/// along with a description and a callback. /// along with a description and a callback.
...@@ -389,11 +352,7 @@ struct message::cli_arg { ...@@ -389,11 +352,7 @@ struct message::cli_arg {
/// Creates a CLI argument for converting from strings, /// Creates a CLI argument for converting from strings,
/// storing its matched argument in `dest`. /// storing its matched argument in `dest`.
template <class T> template <class T>
cli_arg(typename std::enable_if< cli_arg(std::string nstr, std::string tstr, T& arg)
type_nr<T>::value != 0,
std::string
>::type nstr,
std::string tstr, T& arg)
: name(std::move(nstr)), : name(std::move(nstr)),
text(std::move(tstr)), text(std::move(tstr)),
flag(nullptr) { flag(nullptr) {
...@@ -420,7 +379,7 @@ struct message::cli_arg { ...@@ -420,7 +379,7 @@ struct message::cli_arg {
flag(nullptr) { flag(nullptr) {
fun = [&arg](const std::string& str) -> bool { fun = [&arg](const std::string& str) -> bool {
T x; T x;
std::istringstream iss{ str }; std::istringstream iss{str};
if (iss >> x) { if (iss >> x) {
arg.emplace_back(std::move(x)); arg.emplace_back(std::move(x));
return true; return true;
...@@ -430,17 +389,16 @@ struct message::cli_arg { ...@@ -430,17 +389,16 @@ struct message::cli_arg {
} }
}; };
/// Stores the result of `message::extract_opts`. // -- related non-members ------------------------------------------------------
struct message::cli_res {
/// Stores the remaining (unmatched) arguments. /// @relates message
message remainder; error inspect(serializer& sink, message& msg);
/// Stores the names of all active options.
std::set<std::string> opts; /// @relates message
/// Stores the automatically generated help text. error inspect(deserializer& source, message& msg);
std::string helptext;
/// Stores errors during option parsing. /// @relates message
std::string error; std::string to_string(const message& msg);
};
/// @relates message /// @relates message
inline message operator+(const message& lhs, const message& rhs) { inline message operator+(const message& lhs, const message& rhs) {
......
...@@ -36,6 +36,8 @@ ...@@ -36,6 +36,8 @@
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/response_type.hpp" #include "caf/response_type.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
namespace caf { namespace caf {
namespace mixin { namespace mixin {
......
...@@ -138,14 +138,8 @@ protected: ...@@ -138,14 +138,8 @@ protected:
template <class F> template <class F>
bool handle_system_message(mailbox_element& x, execution_unit* context, bool handle_system_message(mailbox_element& x, execution_unit* context,
bool trap_exit, F& down_msg_handler) { bool trap_exit, F& down_msg_handler) {
auto& content = x.content(); if (x.content().type_token() == make_type_token<down_msg>()) {
if (content.type_token() == make_type_token<down_msg>()) { down_msg_handler(x.content().get_mutable_as<down_msg>(0));
if (content.shared()) {
auto vptr = content.copy(0);
down_msg_handler(vptr->get_mutable_as<down_msg>());
} else {
down_msg_handler(content.get_mutable_as<down_msg>(0));
}
return true; return true;
} }
return handle_system_message(x, context, trap_exit); return handle_system_message(x, context, trap_exit);
......
...@@ -823,6 +823,8 @@ public: ...@@ -823,6 +823,8 @@ public:
} }
*/ */
void push_to_cache(mailbox_element_ptr ptr);
/// @endcond /// @endcond
/// Returns the queue for storing incoming messages. /// Returns the queue for storing incoming messages.
...@@ -865,8 +867,6 @@ protected: ...@@ -865,8 +867,6 @@ protected:
bool handle_stream_msg(mailbox_element& x, behavior* active_behavior); bool handle_stream_msg(mailbox_element& x, behavior* active_behavior);
void push_to_cache(mailbox_element_ptr ptr);
// -- Member Variables ------------------------------------------------------- // -- Member Variables -------------------------------------------------------
// used by both event-based and blocking actors // used by both event-based and blocking actors
......
...@@ -60,7 +60,7 @@ public: ...@@ -60,7 +60,7 @@ public:
// -- modifiers -------------------------------------------------------------- // -- modifiers --------------------------------------------------------------
/// Load the content for the tuple from `source`. /// Load the content for the tuple from `source`.
error load(deserializer& source); virtual error load(deserializer& source);
// -- pure virtual observers ------------------------------------------------- // -- pure virtual observers -------------------------------------------------
...@@ -99,11 +99,12 @@ public: ...@@ -99,11 +99,12 @@ public:
std::string stringify() const; std::string stringify() const;
/// Saves the content of the tuple to `sink`. /// Saves the content of the tuple to `sink`.
error save(serializer& sink) const; virtual error save(serializer& sink) const;
/// Checks whether the type of the stored value at position `pos` /// Checks whether the type of the stored value at position `pos`
/// matches type number `n` and run-time type information `p`. /// matches type number `n` and run-time type information `p`.
bool matches(size_t pos, uint16_t nr, const std::type_info* ptr) const noexcept; bool matches(size_t pos, uint16_t nr,
const std::type_info* ptr) const noexcept;
// -- convenience functions -------------------------------------------------- // -- convenience functions --------------------------------------------------
...@@ -175,9 +176,8 @@ public: ...@@ -175,9 +176,8 @@ public:
/// Returns `true` if the pattern `Ts...` matches the content of this tuple. /// Returns `true` if the pattern `Ts...` matches the content of this tuple.
template <class... Ts> template <class... Ts>
bool match_elements() const noexcept { bool match_elements() const noexcept {
detail::meta_elements<detail::type_list<Ts...>> xs; detail::type_list<Ts...> tk;
return xs.arr.empty() ? empty() return match_elements(tk);
: detail::try_match(*this, &xs.arr[0], sizeof...(Ts));
} }
template <class F> template <class F>
...@@ -190,6 +190,16 @@ public: ...@@ -190,6 +190,16 @@ public:
} }
private: private:
template <class T, class... Ts>
bool match_elements(detail::type_list<T, Ts...>) const noexcept {
detail::meta_elements<detail::type_list<T, Ts...>> xs;
return detail::try_match(*this, &xs.arr[0], 1 + sizeof...(Ts));
}
inline bool match_elements(detail::type_list<>) const noexcept {
return empty();
}
template <class F, class R, class... Ts> template <class F, class R, class... Ts>
optional<R> apply(F& fun, detail::type_list<R>, optional<R> apply(F& fun, detail::type_list<R>,
detail::type_list<Ts...> tk) { detail::type_list<Ts...> tk) {
......
...@@ -50,7 +50,7 @@ public: ...@@ -50,7 +50,7 @@ public:
} }
type_erased_tuple& content() override { type_erased_tuple& content() override {
return msg_.content(); return msg_;
} }
const type_erased_tuple& content() const override { const type_erased_tuple& content() const override {
......
...@@ -56,25 +56,157 @@ message::~message() { ...@@ -56,25 +56,157 @@ message::~message() {
// nop // nop
} }
void message::reset(raw_ptr new_ptr, bool add_ref) noexcept { // -- implementation of type_erased_tuple --------------------------------------
vals_.reset(new_ptr, add_ref);
void* message::get_mutable(size_t p) {
CAF_ASSERT(vals_ != nullptr);
return vals_->get_mutable(p);
} }
void message::swap(message& other) noexcept { error message::load(size_t pos, deserializer& source) {
vals_.swap(other.vals_); printf("load %d\n", (int) pos);
CAF_ASSERT(vals_ != nullptr);
return vals_->load(pos, source);
} }
void* message::get_mutable(size_t p) { size_t message::size() const noexcept {
CAF_ASSERT(vals_); return vals_ != nullptr ? vals_->size() : 0;
return vals_->get_mutable(p); }
uint32_t message::type_token() const noexcept {
return vals_ != nullptr ? vals_->type_token() : 0xFFFFFFFF;
}
message::rtti_pair message::type(size_t pos) const noexcept {
CAF_ASSERT(vals_ != nullptr);
return vals_->type(pos);
}
const void* message::get(size_t pos) const noexcept {
CAF_ASSERT(vals_ != nullptr);
return vals_->get(pos);
}
std::string message::stringify(size_t pos) const {
CAF_ASSERT(vals_ != nullptr);
return vals_->stringify(pos);
}
type_erased_value_ptr message::copy(size_t pos) const {
CAF_ASSERT(vals_ != nullptr);
return vals_->copy(pos);
}
error message::save(size_t pos, serializer& sink) const {
printf("save %d\n", (int) pos);
CAF_ASSERT(vals_ != nullptr);
return vals_->save(pos, sink);
} }
bool message::shared() const noexcept {
return vals_ != nullptr ? vals_->shared() : false;
}
error message::load(deserializer& source) {
if (source.context() == nullptr)
return sec::no_context;
uint16_t zero;
std::string tname;
error err;
err = source.begin_object(zero, tname);
if (err)
return err;
if (zero != 0)
return sec::unknown_type;
if (tname == "@<>") {
vals_.reset();
return none;
}
if (tname.compare(0, 4, "@<>+") != 0)
return sec::unknown_type;
// iterate over concatenated type names
auto eos = tname.end();
auto next = [&](std::string::iterator iter) {
return std::find(iter, eos, '+');
};
auto& types = source.context()->system().types();
auto dmd = make_counted<detail::dynamic_message_data>();
std::string tmp;
std::string::iterator i = next(tname.begin());
++i; // skip first '+' sign
do {
auto n = next(i);
tmp.assign(i, n);
auto ptr = types.make_value(tmp);
if (!ptr)
return make_error(sec::unknown_type, tmp);
err = ptr->load(source);
if (err)
return err;
dmd->append(std::move(ptr));
if (n != eos)
i = n + 1;
else
i = eos;
} while (i != eos);
err = source.end_object();
if (err)
return err;
message result{std::move(dmd)};
swap(result);
return none;
}
error message::save(serializer& sink) const {
if (sink.context() == nullptr)
return sec::no_context;
// build type name
uint16_t zero = 0;
std::string tname = "@<>";
if (empty())
return error::eval([&] { return sink.begin_object(zero, tname); },
[&] { return sink.end_object(); });
auto& types = sink.context()->system().types();
auto n = size();
for (size_t i = 0; i < n; ++i) {
auto rtti = cvals()->type(i);
auto ptr = types.portable_name(rtti);
if (ptr == nullptr) {
std::cerr << "[ERROR]: cannot serialize message because a type was "
"not added to the types list, typeid name: "
<< (rtti.second != nullptr ? rtti.second->name()
: "-not-available-")
<< std::endl;
return make_error(sec::unknown_type, rtti.second != nullptr
? rtti.second->name()
: "-not-available-");
}
tname += '+';
tname += *ptr;
}
auto save_loop = [&]() -> error {
for (size_t i = 0; i < n; ++i) {
auto e = cvals()->save(i, sink);
if (e)
return e;
}
return none;
};
return error::eval([&] { return sink.begin_object(zero, tname); },
[&] { return save_loop(); },
[&] { return sink.end_object(); });
}
// -- factories ----------------------------------------------------------------
message message::copy(const type_erased_tuple& xs) { message message::copy(const type_erased_tuple& xs) {
message_builder mb; message_builder mb;
for (size_t i = 0; i < xs.size(); ++i) for (size_t i = 0; i < xs.size(); ++i)
mb.emplace(xs.copy(i)); mb.emplace(xs.copy(i));
return mb.move_to_message(); return mb.move_to_message();
} }
// -- modifiers ----------------------------------------------------------------
message& message::operator+=(const message& x) { message& message::operator+=(const message& x) {
auto tmp = *this + x; auto tmp = *this + x;
...@@ -82,8 +214,22 @@ message& message::operator+=(const message& x) { ...@@ -82,8 +214,22 @@ message& message::operator+=(const message& x) {
return *this; return *this;
} }
optional<message> message::apply(message_handler handler) {
return handler(*this);
}
void message::swap(message& other) noexcept {
vals_.swap(other.vals_);
}
void message::reset(raw_ptr new_ptr, bool add_ref) noexcept {
vals_.reset(new_ptr, add_ref);
}
// -- observers ----------------------------------------------------------------
message message::drop(size_t n) const { message message::drop(size_t n) const {
CAF_ASSERT(vals_); CAF_ASSERT(vals_ != nullptr);
if (n == 0) if (n == 0)
return *this; return *this;
if (n >= size()) if (n >= size())
...@@ -95,7 +241,7 @@ message message::drop(size_t n) const { ...@@ -95,7 +241,7 @@ message message::drop(size_t n) const {
} }
message message::drop_right(size_t n) const { message message::drop_right(size_t n) const {
CAF_ASSERT(vals_); CAF_ASSERT(vals_ != nullptr);
if (n == 0) { if (n == 0) {
return *this; return *this;
} }
...@@ -117,33 +263,6 @@ message message::slice(size_t pos, size_t n) const { ...@@ -117,33 +263,6 @@ message message::slice(size_t pos, size_t n) const {
return message{detail::decorated_tuple::make(vals_, std::move(mapping))}; return message{detail::decorated_tuple::make(vals_, std::move(mapping))};
} }
optional<message> message::apply(message_handler handler) {
return handler(*this);
}
message message::extract_impl(size_t start, message_handler handler) const {
auto s = size();
for (size_t i = start; i < s; ++i) {
for (size_t n = (s - i) ; n > 0; --n) {
auto next_slice = slice(i, n);
auto res = handler(next_slice);
if (res) {
std::vector<size_t> mapping(s);
std::iota(mapping.begin(), mapping.end(), size_t{0});
auto first = mapping.begin() + static_cast<ptrdiff_t>(i);
auto last = first + static_cast<ptrdiff_t>(n);
mapping.erase(first, last);
if (mapping.empty()) {
return message{};
}
message next{detail::decorated_tuple::make(vals_, std::move(mapping))};
return next.extract_impl(i, handler);
}
}
}
return *this;
}
message message::extract(message_handler handler) const { message message::extract(message_handler handler) const {
return extract_impl(0, std::move(handler)); return extract_impl(0, std::move(handler));
} }
...@@ -325,6 +444,46 @@ message::cli_res message::extract_opts(std::vector<cli_arg> xs, ...@@ -325,6 +444,46 @@ message::cli_res message::extract_opts(std::vector<cli_arg> xs,
return {res, std::move(opts), std::move(helpstr), std::move(error)}; return {res, std::move(opts), std::move(helpstr), std::move(error)};
} }
// -- private helpers ----------------------------------------------------------
message message::extract_impl(size_t start, message_handler handler) const {
auto s = size();
for (size_t i = start; i < s; ++i) {
for (size_t n = (s - i) ; n > 0; --n) {
auto next_slice = slice(i, n);
auto res = handler(next_slice);
if (res) {
std::vector<size_t> mapping(s);
std::iota(mapping.begin(), mapping.end(), size_t{0});
auto first = mapping.begin() + static_cast<ptrdiff_t>(i);
auto last = first + static_cast<ptrdiff_t>(n);
mapping.erase(first, last);
if (mapping.empty()) {
return message{};
}
message next{detail::decorated_tuple::make(vals_, std::move(mapping))};
return next.extract_impl(i, handler);
}
}
}
return *this;
}
message message::concat_impl(std::initializer_list<data_ptr> xs) {
auto not_nullptr = [](const data_ptr& ptr) { return ptr.get() != nullptr; };
switch (std::count_if(xs.begin(), xs.end(), not_nullptr)) {
case 0:
return message{};
case 1:
return message{*std::find_if(xs.begin(), xs.end(), not_nullptr)};
default:
return message{detail::concatenated_tuple::make(xs)};
}
}
// -- nested types -------------------------------------------------------------
message::cli_arg::cli_arg(std::string nstr, std::string tstr) message::cli_arg::cli_arg(std::string nstr, std::string tstr)
: name(std::move(nstr)), : name(std::move(nstr)),
text(std::move(tstr)), text(std::move(tstr)),
...@@ -384,105 +543,7 @@ message::cli_arg::cli_arg(std::string nstr, std::string tstr, ...@@ -384,105 +543,7 @@ message::cli_arg::cli_arg(std::string nstr, std::string tstr,
// nop // nop
} }
message message::concat_impl(std::initializer_list<data_ptr> xs) { // -- related non-members ------------------------------------------------------
auto not_nullptr = [](const data_ptr& ptr) { return ptr.get() != nullptr; };
switch (std::count_if(xs.begin(), xs.end(), not_nullptr)) {
case 0:
return message{};
case 1:
return message{*std::find_if(xs.begin(), xs.end(), not_nullptr)};
default:
return message{detail::concatenated_tuple::make(xs)};
}
}
error inspect(serializer& sink, message& msg) {
if (sink.context() == nullptr)
return sec::no_context;
// build type name
uint16_t zero = 0;
std::string tname = "@<>";
if (msg.empty())
return error::eval([&] { return sink.begin_object(zero, tname); },
[&] { return sink.end_object(); });
auto& types = sink.context()->system().types();
auto n = msg.size();
for (size_t i = 0; i < n; ++i) {
auto rtti = msg.cvals()->type(i);
auto ptr = types.portable_name(rtti);
if (ptr == nullptr) {
std::cerr << "[ERROR]: cannot serialize message because a type was "
"not added to the types list, typeid name: "
<< (rtti.second != nullptr ? rtti.second->name() : "-not-available-")
<< std::endl;
return make_error(sec::unknown_type,
rtti.second != nullptr ? rtti.second->name() : "-not-available-");
}
tname += '+';
tname += *ptr;
}
auto save_loop = [&]() -> error {
for (size_t i = 0; i < n; ++i) {
auto e = msg.cvals()->save(i, sink);
if (e)
return e;
}
return none;
};
return error::eval([&] { return sink.begin_object(zero, tname); },
[&] { return save_loop(); },
[&] { return sink.end_object(); });
}
error inspect(deserializer& source, message& msg) {
if (source.context() == nullptr)
return sec::no_context;
uint16_t zero;
std::string tname;
error err;
err = source.begin_object(zero, tname);
if (err)
return err;
if (zero != 0)
return sec::unknown_type;
if (tname == "@<>") {
msg = message{};
return none;
}
if (tname.compare(0, 4, "@<>+") != 0)
return sec::unknown_type;
// iterate over concatenated type names
auto eos = tname.end();
auto next = [&](std::string::iterator iter) {
return std::find(iter, eos, '+');
};
auto& types = source.context()->system().types();
auto dmd = make_counted<detail::dynamic_message_data>();
std::string tmp;
std::string::iterator i = next(tname.begin());
++i; // skip first '+' sign
do {
auto n = next(i);
tmp.assign(i, n);
auto ptr = types.make_value(tmp);
if (!ptr)
return make_error(sec::unknown_type, tmp);
err = ptr->load(source);
if (err)
return err;
dmd->append(std::move(ptr));
if (n != eos)
i = n + 1;
else
i = eos;
} while (i != eos);
err = source.end_object();
if (err)
return err;
message result{std::move(dmd)};
msg.swap(result);
return none;
}
std::string to_string(const message& msg) { std::string to_string(const message& msg) {
if (msg.empty()) if (msg.empty())
......
...@@ -37,19 +37,19 @@ using namespace caf; ...@@ -37,19 +37,19 @@ using namespace caf;
namespace { namespace {
template <class... Ts> template <class... Ts>
optional<tuple<Ts...>> fetch(type_erased_tuple& x) { optional<tuple<Ts...>> fetch(const type_erased_tuple& x) {
if (!x.match_elements<Ts...>()) if (!x.match_elements<Ts...>())
return none; return none;
return x.get_as_tuple<Ts...>(); return x.get_as_tuple<Ts...>();
} }
template <class... Ts> template <class... Ts>
optional<tuple<Ts...>> fetch(message& x) { optional<tuple<Ts...>> fetch(const message& x) {
return fetch<Ts...>(x.content()); return fetch<Ts...>(x.content());
} }
template <class... Ts> template <class... Ts>
optional<tuple<Ts...>> fetch(mailbox_element& x) { optional<tuple<Ts...>> fetch(const mailbox_element& x) {
return fetch<Ts...>(x.content()); return fetch<Ts...>(x.content());
} }
......
...@@ -86,10 +86,48 @@ struct fixture { ...@@ -86,10 +86,48 @@ struct fixture {
actor_system system{cfg}; actor_system system{cfg};
}; };
struct fail_on_copy {
int value;
fail_on_copy(int x = 0) : value(x) {
// nop
}
fail_on_copy(fail_on_copy&&) = default;
fail_on_copy& operator=(fail_on_copy&&) = default;
fail_on_copy(const fail_on_copy&) {
CAF_FAIL("fail_on_copy: copy constructor called");
}
fail_on_copy& operator=(const fail_on_copy&) {
CAF_FAIL("fail_on_copy: copy assign operator called");
}
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, fail_on_copy& x) {
return f(x.value);
}
} // namespace <anonymous> } // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(message_lifetime_tests, fixture) CAF_TEST_FIXTURE_SCOPE(message_lifetime_tests, fixture)
CAF_TEST(nocopy_in_scoped_actor) {
auto msg = make_message(fail_on_copy{1});
scoped_actor self{system};
self->send(self, msg);
self->receive(
[&](const fail_on_copy& x) {
CAF_CHECK_EQUAL(x.value, 1);
CAF_CHECK_EQUAL(msg.cvals()->get_reference_count(), 2u);
}
);
CAF_CHECK_EQUAL(msg.cvals()->get_reference_count(), 1u);
}
CAF_TEST(message_lifetime_in_scoped_actor) { CAF_TEST(message_lifetime_in_scoped_actor) {
auto msg = make_message(1, 2, 3); auto msg = make_message(1, 2, 3);
scoped_actor self{system}; scoped_actor self{system};
......
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