Commit 77569942 authored by Dominik Charousset's avatar Dominik Charousset

Re-implement message with new flat design

parent e09a6170
......@@ -63,7 +63,6 @@ set(CAF_CORE_SOURCES
src/detail/behavior_impl.cpp
src/detail/behavior_stack.cpp
src/detail/blocking_behavior.cpp
src/detail/dynamic_message_data.cpp
src/detail/fnv_hash.cpp
src/detail/get_mac_addresses.cpp
src/detail/get_process_id.cpp
......@@ -86,6 +85,7 @@ set(CAF_CORE_SOURCES
src/detail/test_actor_clock.cpp
src/detail/thread_safe_actor_clock.cpp
src/detail/tick_emitter.cpp
src/detail/type_id_list_builder.cpp
src/detail/uri_impl.cpp
src/downstream_manager.cpp
src/downstream_manager_base.cpp
......@@ -153,8 +153,6 @@ set(CAF_CORE_SOURCES
src/timestamp.cpp
src/tracing_data.cpp
src/tracing_data_factory.cpp
src/type_erased_tuple.cpp
src/type_erased_value.cpp
src/type_id_list.cpp
src/uri.cpp
src/uri_builder.cpp
......@@ -210,6 +208,7 @@ set(CAF_CORE_TEST_SOURCES
test/detail/ripemd_160.cpp
test/detail/serialized_size.cpp
test/detail/tick_emitter.cpp
test/detail/type_id_list_builder.cpp
test/detail/unique_function.cpp
test/detail/unordered_flat_map.cpp
test/dictionary.cpp
......@@ -268,7 +267,6 @@ set(CAF_CORE_TEST_SOURCES
test/thread_hook.cpp
test/to_string.cpp
test/tracing_data.cpp
test/type_erased_tuple.cpp
test/type_id_list.cpp
test/typed_message_view.cpp
test/typed_response_promise.cpp
......
......@@ -45,82 +45,63 @@ template <class F, class T, class Bhvr, class R, class... Ts>
class fun_decorator<F, T, Bhvr, spawn_mode::function, R,
detail::type_list<Ts...>> {
public:
fun_decorator(F f, T*) : f_(std::move(f)) {
fun_decorator(F f, T*, behavior* bhvr) : f_(std::move(f)), bhvr_(bhvr) {
// nop
}
behavior operator()(Ts... xs) {
detail::type_list<R> token;
return apply(token, xs...);
}
template <class U>
typename std::enable_if<std::is_convertible<U, Bhvr>::value, behavior>::type
apply(detail::type_list<U>, Ts... xs) {
auto bhvr = f_(xs...);
return std::move(bhvr.unbox());
}
template <class U>
typename std::enable_if<!std::is_convertible<U, Bhvr>::value, behavior>::type
apply(detail::type_list<U>, Ts... xs) {
f_(xs...);
return {};
void operator()(Ts... xs) {
if constexpr (std::is_convertible<R, Bhvr>::value) {
auto bhvr = f_(xs...);
*bhvr_ = std::move(bhvr.unbox());
} else {
f_(xs...);
}
}
private:
F f_;
behavior* bhvr_;
};
template <class F, class T, class Bhvr, class R, class... Ts>
class fun_decorator<F, T, Bhvr, spawn_mode::function_with_selfptr, R,
detail::type_list<T*, Ts...>> {
public:
fun_decorator(F f, T* ptr) : f_(std::move(f)), ptr_(ptr) {
fun_decorator(F f, T* ptr, behavior* bhvr)
: f_(std::move(f)), ptr_(ptr), bhvr_(bhvr) {
// nop
}
behavior operator()(Ts... xs) {
detail::type_list<R> token;
return apply(token, xs...);
}
template <class U>
typename std::enable_if<std::is_convertible<U, Bhvr>::value, behavior>::type
apply(detail::type_list<U>, Ts... xs) {
auto bhvr = f_(ptr_, xs...);
return std::move(bhvr.unbox());
}
template <class U>
typename std::enable_if<!std::is_convertible<U, Bhvr>::value, behavior>::type
apply(detail::type_list<U>, Ts... xs) {
f_(ptr_, xs...);
return {};
void operator()(Ts... xs) {
if constexpr (std::is_convertible<R, Bhvr>::value) {
auto bhvr = f_(ptr_, xs...);
*bhvr_ = std::move(bhvr.unbox());
} else {
f_(ptr_, xs...);
}
}
private:
F f_;
T* ptr_;
behavior* bhvr_;
};
template <class Args>
template <spawn_mode Mode, class Args>
struct message_verifier;
template <>
struct message_verifier<detail::type_list<>> {
bool operator()(message& msg, void_mode_token) {
return msg.empty();
template <class... Ts>
struct message_verifier<spawn_mode::function, detail::type_list<Ts...>> {
bool operator()(message& msg) {
return msg.types() == make_type_id_list<Ts...>();
}
};
template <class T, class... Ts>
struct message_verifier<detail::type_list<T, Ts...>> {
bool operator()(message& msg, void_mode_token) {
return msg.match_elements<T, Ts...>();
}
bool operator()(message& msg, selfptr_mode_token) {
return msg.match_elements<Ts...>();
template <class Self, class... Ts>
struct message_verifier<spawn_mode::function_with_selfptr,
detail::type_list<Self*, Ts...>> {
bool operator()(message& msg) {
return msg.types() == make_type_id_list<Ts...>();
}
};
......@@ -131,23 +112,21 @@ actor_factory make_actor_factory(F fun) {
using handle = typename trait::type;
using impl = typename trait::impl;
using behavior_t = typename trait::behavior_type;
spawn_mode_token<trait::mode> tk;
message_verifier<typename trait::arg_types> mv;
if (!mv(msg, tk))
message_verifier<trait::mode, typename trait::arg_types> verify;
if (!verify(msg))
return {};
cfg.init_fun = actor_config::init_fun_type{[=](local_actor* x) -> behavior {
using ctrait = typename detail::get_callable_trait<F>::type;
using fd = fun_decorator<F, impl, behavior_t, trait::mode,
typename ctrait::result_type,
typename ctrait::arg_types>;
fd f{fun, static_cast<impl*>(x)};
empty_type_erased_tuple dummy_;
auto& ct = msg.empty() ? dummy_ : const_cast<message&>(msg).content();
auto opt = ct.apply(f);
if (!opt)
return {};
return std::move(*opt);
}};
cfg.init_fun = actor_config::init_fun_type{
[=](local_actor* x) mutable -> behavior {
using ctrait = typename detail::get_callable_trait<F>::type;
using fd = fun_decorator<F, impl, behavior_t, trait::mode,
typename ctrait::result_type,
typename ctrait::arg_types>;
behavior result;
message_handler f{fd{fun, static_cast<impl*>(x), &result}};
f(msg);
return result;
},
};
handle hdl = cfg.host->system().spawn_class<impl, no_spawn_options>(cfg);
return {actor_cast<strong_actor_ptr>(std::move(hdl)),
cfg.host->system().message_types<handle>()};
......@@ -169,8 +148,8 @@ actor_factory_result dyn_spawn_class(actor_config& cfg, message& msg) {
CAF_ASSERT(cfg.host);
using handle = typename infer_handle_from_class<T>::type;
handle hdl;
dyn_spawn_class_helper<handle, T, Ts...> factory{hdl, cfg};
msg.apply(factory);
message_handler factory{dyn_spawn_class_helper<handle, T, Ts...>{hdl, cfg}};
factory(msg);
return {actor_cast<strong_actor_ptr>(std::move(hdl)),
cfg.host->system().message_types<handle>()};
}
......
......@@ -42,7 +42,6 @@
#include "caf/settings.hpp"
#include "caf/stream.hpp"
#include "caf/thread_hook.hpp"
#include "caf/type_erased_value.hpp"
namespace caf {
......
......@@ -56,6 +56,7 @@
#include "caf/config_value_adaptor_field.hpp"
#include "caf/config_value_field.hpp"
#include "caf/config_value_object_access.hpp"
#include "caf/const_typed_message_view.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/defaults.hpp"
#include "caf/deserializer.hpp"
......@@ -114,6 +115,7 @@
#include "caf/typed_actor_view.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/typed_message_view.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/unit.hpp"
#include "caf/upstream_msg.hpp"
......
......@@ -98,16 +98,6 @@ public:
return impl_ ? impl_->invoke(xs) : none;
}
optional<message> operator()(type_erased_tuple& xs) {
return impl_ ? impl_->invoke(xs) : none;
}
/// Runs this handler with callback.
match_result operator()(detail::invoke_result_visitor& f,
type_erased_tuple& xs) {
return impl_ ? impl_->invoke(f, xs) : match_result::no_match;
}
/// Runs this handler with callback.
match_result operator()(detail::invoke_result_visitor& f, message& xs) {
return impl_ ? impl_->invoke(f, xs) : match_result::no_match;
......
......@@ -18,6 +18,10 @@
#pragma once
#include <utility>
#include "caf/detail/message_data.hpp"
#include "caf/detail/offset_at.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/message.hpp"
......@@ -26,32 +30,57 @@ namespace caf {
template <class... Ts>
class const_typed_message_view {
public:
explicit const_typed_message_view(const type_erased_tuple& msg) noexcept
: ptr_(&msg) {
const_typed_message_view() noexcept : ptr_(nullptr) {
// nop
}
const_typed_message_view() = delete;
explicit const_typed_message_view(const message& msg) noexcept
: ptr_(&msg.cdata()) {
// nop
}
const_typed_message_view(const const_typed_message_view&) noexcept = default;
const_typed_message_view& operator=(const const_typed_message_view&) noexcept
= default;
const type_erased_tuple* operator->() const noexcept {
const detail::message_data* operator->() const noexcept {
return ptr_;
}
explicit operator bool() const noexcept {
return ptr_ != nullptr;
}
private:
const type_erased_tuple* ptr_;
const detail::message_data* ptr_;
};
template <size_t Position, class... Ts>
const auto& get(const const_typed_message_view<Ts...>& x) {
static_assert(Position < sizeof...(Ts));
using types = detail::type_list<Ts...>;
using type = detail::tl_at_t<types, Position>;
return *reinterpret_cast<const type*>(x->get(Position));
template <size_t Index, class... Ts>
const auto& get(const_typed_message_view<Ts...>& xs) {
static_assert(Index < sizeof...(Ts));
using type = caf::detail::tl_at_t<caf::detail::type_list<Ts...>, Index>;
return *reinterpret_cast<const type*>(xs->storage()
+ detail::offset_at<Index, Ts...>);
}
template <class... Ts, size_t... Is>
auto to_tuple(const_typed_message_view<Ts...> xs, std::index_sequence<Is...>) {
return std::make_tuple(get<Is>(xs)...);
}
template <class... Ts>
auto to_tuple(const_typed_message_view<Ts...> xs) {
std::make_index_sequence<sizeof...(Ts)> seq;
return to_tuple(xs, seq);
}
template <class... Ts>
auto make_const_typed_message_view(const message& msg) {
if (msg.types() == make_type_id_list<Ts...>())
return const_typed_message_view<Ts...>{msg};
return const_typed_message_view<Ts...>{};
}
} // namespace caf
......@@ -21,8 +21,8 @@
#include <tuple>
#include "caf/detail/comparable.hpp"
#include "caf/detail/tuple_vals.hpp"
#include "caf/make_copy_on_write.hpp"
#include "caf/ref_counted.hpp"
namespace caf {
......@@ -33,10 +33,27 @@ class cow_tuple : detail::comparable<cow_tuple<Ts...>>,
public:
// -- member types -----------------------------------------------------------
using impl = detail::tuple_vals<Ts...>;
using data_type = std::tuple<Ts...>;
struct impl : ref_counted {
template <class... Us>
impl(Us&&... xs) : data(std::forward<Us>(xs)...) {
// nop
}
impl() = default;
impl(const impl&) = default;
impl& operator=(const impl&) = default;
impl* copy() const {
return new impl{*this};
}
data_type data;
};
// -- constructors, destructors, and assignment operators --------------------
explicit cow_tuple(Ts... xs)
......@@ -60,13 +77,13 @@ public:
/// Returns the managed tuple.
const data_type& data() const noexcept {
return ptr_->data();
return ptr_->data;
}
/// Returns a mutable reference to the managed tuple, guaranteed to have a
/// reference count of 1.
data_type& unshared() {
return ptr_.unshared().data();
return ptr_.unshared().data;
}
/// Returns whether the reference count of the managed object is 1.
......
......@@ -65,16 +65,11 @@ public:
match_result invoke_empty(detail::invoke_result_visitor& f);
virtual match_result
invoke(detail::invoke_result_visitor& f, type_erased_tuple& xs)
virtual match_result invoke(detail::invoke_result_visitor& f, message& xs)
= 0;
match_result invoke(detail::invoke_result_visitor& f, message& xs);
optional<message> invoke(message&);
optional<message> invoke(type_erased_tuple&);
virtual void handle_timeout();
timespan timeout() const noexcept {
......@@ -131,13 +126,13 @@ public:
}
virtual match_result invoke(detail::invoke_result_visitor& f,
type_erased_tuple& xs) override {
message& xs) override {
return invoke_impl(f, xs, std::make_index_sequence<sizeof...(Ts)>{});
}
template <size_t... Is>
match_result invoke_impl(detail::invoke_result_visitor& f,
type_erased_tuple& msg, std::index_sequence<Is...>) {
match_result invoke_impl(detail::invoke_result_visitor& f, message& msg,
std::index_sequence<Is...>) {
auto result = match_result::no_match;
auto dispatch = [&](auto& fun) {
using fun_type = std::decay_t<decltype(fun)>;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <algorithm>
#include <vector>
#include "caf/config.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/message_data.hpp"
#include "caf/detail/tuple_vals.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/ref_counted.hpp"
namespace caf::detail {
class CAF_CORE_EXPORT decorated_tuple : public message_data {
public:
// -- member types -----------------------------------------------------------
using message_data::cow_ptr;
using vector_type = std::vector<size_t>;
// -- constructors, destructors, and assignment operators --------------------
decorated_tuple(cow_ptr&&, vector_type&&);
static cow_ptr make(cow_ptr d, vector_type v);
decorated_tuple& operator=(const decorated_tuple&) = delete;
// -- overridden observers of message_data -----------------------------------
message_data* copy() const override;
// -- overridden modifiers of type_erased_tuple ------------------------------
void* get_mutable(size_t pos) override;
error load(size_t pos, deserializer& source) override;
// -- overridden observers of type_erased_tuple ------------------------------
size_t size() const noexcept override;
type_id_t 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;
// -- inline observers -------------------------------------------------------
inline const cow_ptr& decorated() const {
return decorated_;
}
inline const vector_type& mapping() const {
return mapping_;
}
private:
// -- constructors, destructors, and assignment operators --------------------
decorated_tuple(const decorated_tuple&) = default;
// -- data members -----------------------------------------------------------
cow_ptr decorated_;
vector_type mapping_;
};
} // namespace caf::detail
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <vector>
#include "caf/detail/core_export.hpp"
#include "caf/detail/message_data.hpp"
#include "caf/type_erased_value.hpp"
#include "caf/type_id.hpp"
namespace caf::detail {
class CAF_CORE_EXPORT dynamic_message_data : public message_data {
public:
// -- member types -----------------------------------------------------------
using elements = std::vector<type_erased_value_ptr>;
// -- constructors, destructors, and assignment operators --------------------
dynamic_message_data();
dynamic_message_data(elements&& data);
dynamic_message_data(dynamic_message_data&&) = default;
dynamic_message_data(const dynamic_message_data& other);
dynamic_message_data& operator=(dynamic_message_data&&) = delete;
dynamic_message_data& operator=(const dynamic_message_data&) = delete;
~dynamic_message_data() override;
// -- overridden observers of message_data -----------------------------------
dynamic_message_data* copy() const override;
// -- overridden modifiers of type_erased_tuple ------------------------------
void* get_mutable(size_t pos) override;
error load(size_t pos, deserializer& source) override;
error_code<sec> load(size_t pos, binary_deserializer& source) override;
// -- overridden observers of type_erased_tuple ------------------------------
size_t size() const noexcept override;
type_id_list types() const noexcept override;
type_id_t 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;
error_code<sec> save(size_t pos, binary_serializer& sink) const override;
// -- modifiers --------------------------------------------------------------
void clear();
void append(type_erased_value_ptr x);
private:
// -- data members -----------------------------------------------------------
std::vector<type_id_t> types_;
elements elements_;
};
CAF_CORE_EXPORT void intrusive_ptr_add_ref(const dynamic_message_data*);
CAF_CORE_EXPORT void intrusive_ptr_release(const dynamic_message_data*);
CAF_CORE_EXPORT dynamic_message_data*
intrusive_cow_ptr_unshare(dynamic_message_data*&);
} // namespace caf::detail
......@@ -26,7 +26,6 @@
#include "caf/detail/type_traits.hpp"
#include "caf/expected.hpp"
#include "caf/fwd.hpp"
#include "caf/make_message.hpp"
#include "caf/make_sink_result.hpp"
#include "caf/make_source_result.hpp"
#include "caf/make_stage_result.hpp"
......
......@@ -26,7 +26,7 @@
#include "caf/byte.hpp"
#include "caf/deserializer.hpp"
#include "caf/detail/meta_object.hpp"
#include "caf/detail/type_erased_value_impl.hpp"
#include "caf/detail/padded_size.hpp"
#include "caf/error.hpp"
#include "caf/serializer.hpp"
......@@ -36,8 +36,12 @@ template <class T>
meta_object make_meta_object(const char* type_name) {
return {
type_name,
padded_size_v<T>,
[](void* ptr) noexcept { reinterpret_cast<T*>(ptr)->~T(); },
[](void* ptr) { new (ptr) T(); },
[](const void* src, void* dst) {
new (dst) T(*reinterpret_cast<const T*>(src));
},
[](caf::binary_serializer& sink, const void* ptr) {
return sink(*reinterpret_cast<const T*>(ptr));
},
......@@ -50,10 +54,6 @@ meta_object make_meta_object(const char* type_name) {
[](caf::deserializer& source, void* ptr) {
return source(*reinterpret_cast<T*>(ptr));
},
// Temporary hack.
[]() -> type_erased_value* {
return new detail::type_erased_value_impl<T>;
},
};
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/abstract_actor.hpp"
#include "caf/actor_addr.hpp"
#include "caf/attachable.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/message.hpp"
namespace caf::detail {
class CAF_CORE_EXPORT merged_tuple : public message_data {
public:
// -- member types -----------------------------------------------------------
using message_data::cow_ptr;
using data_type = std::vector<cow_ptr>;
using mapping_type = std::vector<std::pair<size_t, size_t>>;
// -- constructors, destructors, and assignment operators --------------------
static cow_ptr make(message x, message y);
merged_tuple(data_type xs, mapping_type ys);
merged_tuple& operator=(const merged_tuple&) = delete;
// -- overridden observers of message_data -----------------------------------
merged_tuple* copy() const override;
// -- overridden modifiers of type_erased_tuple ------------------------------
void* get_mutable(size_t pos) override;
error load(size_t pos, deserializer& source) override;
// -- overridden observers of type_erased_tuple ------------------------------
size_t size() const noexcept override;
type_id_t 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;
// -- observers --------------------------------------------------------------
const mapping_type& mapping() const;
private:
// -- constructors, destructors, and assignment operators --------------------
merged_tuple(const merged_tuple&) = default;
// -- data members -----------------------------------------------------------
data_type data_;
mapping_type mapping_;
};
} // namespace caf::detail
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Copyright 2011-2020 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 *
......@@ -18,67 +18,60 @@
#pragma once
#include <algorithm>
#include <vector>
#include <memory>
#include <new>
#include "caf/byte.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/decorated_tuple.hpp"
#include "caf/detail/padded_size.hpp"
namespace caf::detail {
class CAF_CORE_EXPORT concatenated_tuple : public message_data {
/// Wraps a value for either copying or moving it into a pre-allocated storage
/// later.
class CAF_CORE_EXPORT message_builder_element {
public:
// -- member types -----------------------------------------------------------
using message_data::cow_ptr;
using vector_type = std::vector<cow_ptr>;
// -- constructors, destructors, and assignment operators --------------------
concatenated_tuple(std::initializer_list<cow_ptr> xs);
static cow_ptr make(std::initializer_list<cow_ptr> xs);
concatenated_tuple(const concatenated_tuple&) = default;
concatenated_tuple& operator=(const concatenated_tuple&) = delete;
// -- overridden observers of message_data -----------------------------------
concatenated_tuple* copy() const override;
// -- overridden modifiers of type_erased_tuple ------------------------------
void* get_mutable(size_t pos) override;
error load(size_t pos, deserializer& source) override;
// -- overridden observers of type_erased_tuple ------------------------------
size_t size() const noexcept override;
type_id_t 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;
virtual ~message_builder_element() noexcept;
/// Uses placement new to create a copy of the wrapped value at given memory
/// region.
/// @returns the past-the-end pointer of the object, i.e., the first byte for
/// the *next* object.
virtual byte* copy_init(byte* storage) const = 0;
/// Uses placement new to move the wrapped value to given memory region.
/// @returns the past-the-end pointer of the object, i.e., the first byte for
/// the *next* object.
virtual byte* move_init(byte* storage) = 0;
};
// -- element access ---------------------------------------------------------
template <class T>
class message_builder_element_impl : public message_builder_element {
public:
message_builder_element_impl(T value) : value_(std::move(value)) {
// nop
}
std::pair<message_data*, size_t> select(size_t pos);
byte* copy_init(byte* storage) const override {
new (storage) T(value_);
return storage + padded_size_v<T>;
}
std::pair<const message_data*, size_t> select(size_t pos) const;
byte* move_init(byte* storage) override {
new (storage) T(std::move(value_));
return storage + padded_size_v<T>;
}
private:
// -- data members -----------------------------------------------------------
vector_type data_;
size_t size_;
T value_;
};
using message_builder_element_ptr = std::unique_ptr<message_builder_element>;
template <class T>
auto make_message_builder_element(T&& x) {
using impl = message_builder_element_impl<std::decay_t<T>>;
return message_builder_element_ptr{new impl(std::forward<T>(x))};
}
} // namespace caf::detail
......@@ -18,44 +18,143 @@
#pragma once
#include <iterator>
#include <string>
#include <typeinfo>
#include <atomic>
#include <cstdlib>
#include "caf/byte.hpp"
#include "caf/config.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/implicit_conversions.hpp"
#include "caf/detail/padded_size.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive_cow_ptr.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/ref_counted.hpp"
#include "caf/type_erased_tuple.hpp"
#include "caf/type_id_list.hpp"
#ifdef CAF_CLANG
# pragma clang diagnostic push
# pragma clang diagnostic ignored "-Wc99-extensions"
#elif defined(CAF_GCC)
# pragma GCC diagnostic push
# pragma GCC diagnostic ignored "-Wc99-extensions"
#endif
namespace caf::detail {
class CAF_CORE_EXPORT message_data : public ref_counted,
public type_erased_tuple {
/// Container for storing an arbitrary number of message elements.
class CAF_CORE_EXPORT message_data {
public:
// -- nested types -----------------------------------------------------------
// -- constructors, destructors, and assignment operators --------------------
using cow_ptr = intrusive_cow_ptr<message_data>;
message_data() = delete;
// -- constructors, destructors, and assignment operators --------------------
message_data(const message_data&) = delete;
message_data& operator=(const message_data&) = delete;
/// Constructs the message data object *without* constructing any element.
explicit message_data(type_id_list types);
~message_data() noexcept;
message_data* copy() const;
// -- reference counting -----------------------------------------------------
/// Increases reference count by one.
void ref() const noexcept {
rc_.fetch_add(1, std::memory_order_relaxed);
}
/// Decreases reference count by one and calls `request_deletion`
/// when it drops to zero.
void deref() noexcept {
if (unique() || rc_.fetch_sub(1, std::memory_order_acq_rel) == 1) {
this->~message_data();
free(const_cast<message_data*>(this));
}
}
// -- properties -------------------------------------------------------------
/// Queries whether there is exactly one reference to this data.
bool unique() const noexcept {
return rc_ == 1;
}
message_data() = default;
message_data(const message_data&) = default;
/// Returns the current number of references to this data.
size_t get_reference_count() const noexcept {
return rc_.load();
}
~message_data() override;
/// Returns the memory region for storing the message elements.
byte* storage() noexcept {
return storage_;
}
// -- pure virtual observers -------------------------------------------------
/// @copydoc storage
const byte* storage() const noexcept {
return storage_;
}
virtual message_data* copy() const = 0;
/// Returns the type IDs of the message elements.
auto types() const noexcept {
return types_;
}
// -- observers --------------------------------------------------------------
/// Returns the number of elements.
auto size() const noexcept {
return types_.size();
}
using type_erased_tuple::copy;
/// Returns the memory location for the object at given index.
/// @pre `index < size()`
byte* at(size_t index) noexcept;
bool shared() const noexcept override;
/// @copydoc at
const byte* at(size_t index) const noexcept;
caf::error save(caf::serializer& sink) const;
caf::error save(caf::binary_serializer& sink) const;
caf::error load(caf::deserializer& source);
caf::error load(caf::binary_deserializer& source);
private:
mutable std::atomic<size_t> rc_;
type_id_list types_;
byte storage_[];
};
// -- related non-members ------------------------------------------------------
/// @relates message_data
inline void intrusive_ptr_add_ref(const message_data* ptr) {
ptr->ref();
}
/// @relates message_data
inline void intrusive_ptr_release(message_data* ptr) {
ptr->deref();
}
inline void message_data_init(byte*) {
// nop
}
template <class T, class... Ts>
void message_data_init(byte* storage, T&& x, Ts&&... xs) {
// TODO: exception safety: if any constructor throws, we need to unwind the
// stack here and call destructors.
using type = strip_and_convert_t<T>;
new (storage) strip_and_convert_t<T>(std::forward<T>(x));
message_data_init(storage + padded_size_v<type>, std::forward<Ts>(xs)...);
}
} // namespace caf::detail
#ifdef CAF_CLANG
# pragma clang diagnostic pop
#elif defined(CAF_GCC)
# pragma GCC diagnostic pop
#endif
......@@ -32,6 +32,10 @@ struct meta_object {
/// Stores a human-readable representation of the type's name.
const char* type_name = nullptr;
/// Stores how many Bytes objects of this type require, including padding for
/// aligning to `max_align_t`.
size_t padded_size;
/// Calls the destructor for given object.
void (*destroy)(void*) noexcept;
......@@ -39,6 +43,10 @@ struct meta_object {
/// constructor.
void (*default_construct)(void*);
/// Creates a new object at given memory location by calling the copy
/// constructor.
void (*copy_construct)(const void*, void*);
/// Applies an object to a binary serializer.
error_code<sec> (*save_binary)(caf::binary_serializer&, const void*);
......@@ -50,11 +58,24 @@ struct meta_object {
/// Applies an object to a generic deserializer.
caf::error (*load)(caf::deserializer&, void*);
// Temporary hack until re-implementing caf::message.
type_erased_value* (*make)();
};
/// Convenience function for calling `meta.save(sink, obj)`.
CAF_CORE_EXPORT caf::error save(const meta_object& meta, caf::serializer& sink,
const void* obj);
/// Convenience function for calling `meta.save_binary(sink, obj)`.
CAF_CORE_EXPORT caf::error_code<sec>
save(const meta_object& meta, caf::binary_serializer& sink, const void* obj);
/// Convenience function for calling `meta.load(source, obj)`.
CAF_CORE_EXPORT caf::error load(const meta_object& meta,
caf::deserializer& source, void* obj);
/// Convenience function for calling `meta.load_binary(source, obj)`.
CAF_CORE_EXPORT caf::error_code<sec>
load(const meta_object& meta, caf::binary_deserializer& source, void* obj);
/// Returns the global storage for all meta objects. The ::type_id of an object
/// is the index for accessing the corresonding meta object.
CAF_CORE_EXPORT span<const meta_object> global_meta_objects();
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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/padded_size.hpp"
namespace caf::detail {
template <size_t Remaining, class T, class... Ts>
struct offset_at_helper {
static constexpr size_t value = offset_at_helper<Remaining - 1, Ts...>::value
+ padded_size_v<T>;
};
template <class T, class... Ts>
struct offset_at_helper<0, T, Ts...> {
static constexpr size_t value = 0;
};
template <size_t Index, class... Ts>
constexpr size_t offset_at = offset_at_helper<Index, Ts...>::value;
} // namespace caf::detail
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Copyright 2011-2020 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 *
......@@ -19,19 +19,16 @@
#pragma once
#include <cstddef>
#include <cstdint>
#include <tuple>
#include <typeinfo>
#include "caf/detail/type_erased_tuple_view.hpp"
#include "caf/type_erased_tuple.hpp"
namespace caf {
namespace detail {
/// @relates type_erased_tuple
template <class... Ts>
detail::type_erased_tuple_view<Ts...> make_type_erased_tuple_view(Ts&... xs) {
return {xs...};
}
/// Calculates the size for `T` including padding for aligning to `max_align_t`.
template <class T>
constexpr size_t padded_size_v
= ((sizeof(T) / alignof(max_align_t))
+ static_cast<size_t>(sizeof(T) % alignof(max_align_t) != 0))
* alignof(max_align_t);
} // namespace detail
} // namespace caf
......@@ -26,7 +26,7 @@ namespace caf::detail {
template <class... Ts>
class param_message_view {
public:
explicit param_message_view(type_erased_tuple& msg) noexcept : ptr_(&msg) {
explicit param_message_view(const message& msg) noexcept : ptr_(&msg.data()) {
// nop
}
......@@ -37,21 +37,20 @@ public:
param_message_view& operator=(const param_message_view&) noexcept
= default;
const type_erased_tuple* operator->() const noexcept {
const detail::message_data* operator->() const noexcept {
return ptr_;
}
private:
const type_erased_tuple* ptr_;
bool shared_;
const detail::message_data* ptr_;
};
template <size_t Position, class... Ts>
auto get(const param_message_view<Ts...>& x) {
static_assert(Position < sizeof...(Ts));
using types = detail::type_list<Ts...>;
using type = detail::tl_at_t<types, Position>;
return param<type>{x->get(Position), x->shared()};
template <size_t Index, class... Ts>
auto get(const param_message_view<Ts...>& xs) {
static_assert(Index < sizeof...(Ts));
using type = caf::detail::tl_at_t<caf::detail::type_list<Ts...>, Index>;
return param<type>{xs->storage() + detail::offset_at<Index, Ts...>,
!xs->unique()};
}
} // namespace caf::detail
......@@ -30,7 +30,6 @@
#include "caf/detail/type_traits.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/make_message.hpp"
#include "caf/message.hpp"
#include "caf/none.hpp"
#include "caf/parser_state.hpp"
......
......@@ -50,8 +50,8 @@ public:
void handle(inbound_path*, downstream_msg::batch& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
using vec_type = std::vector<input_type>;
if (x.xs.match_elements<vec_type>()) {
driver_.process(x.xs.get_mutable_as<vec_type>(0));
if (auto view = make_typed_message_view<vec_type>(x.xs)) {
driver_.process(get<0>(view));
return;
}
CAF_LOG_ERROR("received unexpected batch type (dropped)");
......
......@@ -61,9 +61,9 @@ public:
void handle(inbound_path*, downstream_msg::batch& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
using vec_type = std::vector<input_type>;
if (x.xs.match_elements<vec_type>()) {
if (auto view = make_typed_message_view<vec_type>(x.xs)) {
downstream<output_type> ds{this->out_.buf()};
driver_.process(ds, x.xs.get_mutable_as<vec_type>(0));
driver_.process(ds, get<0>(view));
return;
}
CAF_LOG_ERROR("received unexpected batch type (dropped)");
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <stdexcept>
#include <tuple>
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/deserializer.hpp"
#include "caf/detail/message_data.hpp"
#include "caf/detail/safe_equal.hpp"
#include "caf/detail/stringification_inspector.hpp"
#include "caf/detail/try_serialize.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/make_type_erased_value.hpp"
#include "caf/serializer.hpp"
#include "caf/type_id_list.hpp"
#define CAF_TUPLE_VALS_DISPATCH(x) \
case x: \
return tuple_inspect_delegate<x, sizeof...(Ts) - 1>(data_, f)
namespace caf::detail {
// avoids triggering static asserts when using CAF_TUPLE_VALS_DISPATCH
template <size_t X, size_t Max, class T, class F>
auto tuple_inspect_delegate(T& data, F& f) -> decltype(f(std::get<Max>(data))) {
return f(std::get<(X < Max ? X : Max)>(data));
}
template <size_t X, size_t N>
struct tup_ptr_access_pos {
constexpr tup_ptr_access_pos() {
// nop
}
};
template <size_t X, size_t N>
constexpr tup_ptr_access_pos<X + 1, N> next(tup_ptr_access_pos<X, N>) {
return {};
}
struct void_ptr_access {
template <class T>
void* operator()(T& x) const noexcept {
return &x;
}
};
template <class Base, class... Ts>
class tuple_vals_impl : public Base {
public:
// -- static invariants ------------------------------------------------------
static_assert(sizeof...(Ts) > 0, "tuple_vals is not allowed to be empty");
// -- member types -----------------------------------------------------------
using super = message_data;
using data_type = std::tuple<Ts...>;
// -- friend functions -------------------------------------------------------
template <class Inspector>
friend typename Inspector::result_type
inspect(Inspector& f, tuple_vals_impl& x) {
return apply_args(f, get_indices(x.data_), x.data_);
}
tuple_vals_impl(const tuple_vals_impl&) = default;
template <class... Us>
tuple_vals_impl(Us&&... xs)
: types_(make_type_id_list<Ts...>()), data_(std::forward<Us>(xs)...) {
// nop
}
data_type& data() {
return data_;
}
const data_type& data() const {
return data_;
}
size_t size() const noexcept override {
return sizeof...(Ts);
}
const void* get(size_t pos) const noexcept override {
CAF_ASSERT(pos < size());
void_ptr_access f;
return mptr()->dispatch(pos, f);
}
void* get_mutable(size_t pos) override {
CAF_ASSERT(pos < size());
void_ptr_access f;
return this->dispatch(pos, f);
}
std::string stringify(size_t pos) const override {
std::string result;
stringification_inspector f{result};
mptr()->dispatch(pos, f);
return result;
}
using Base::copy;
type_erased_value_ptr copy(size_t pos) const override {
type_erased_value_factory f;
return mptr()->dispatch(pos, f);
}
error load(size_t pos, deserializer& source) override {
return dispatch(pos, source);
}
error_code<sec> load(size_t pos, binary_deserializer& source) override {
return dispatch(pos, source);
}
type_id_list types() const noexcept override {
return types_;
}
type_id_t type(size_t pos) const noexcept override {
return types_[pos];
}
error save(size_t pos, serializer& sink) const override {
return mptr()->dispatch(pos, sink);
}
error_code<sec> save(size_t pos, binary_serializer& sink) const override {
return mptr()->dispatch(pos, sink);
}
private:
template <class F>
auto dispatch(size_t pos, F& f) -> decltype(f(std::declval<int&>())) {
CAF_ASSERT(pos < sizeof...(Ts));
switch (pos) {
CAF_TUPLE_VALS_DISPATCH(0);
CAF_TUPLE_VALS_DISPATCH(1);
CAF_TUPLE_VALS_DISPATCH(2);
CAF_TUPLE_VALS_DISPATCH(3);
CAF_TUPLE_VALS_DISPATCH(4);
CAF_TUPLE_VALS_DISPATCH(5);
CAF_TUPLE_VALS_DISPATCH(6);
CAF_TUPLE_VALS_DISPATCH(7);
CAF_TUPLE_VALS_DISPATCH(8);
CAF_TUPLE_VALS_DISPATCH(9);
default:
// fall back to recursive dispatch function
static constexpr size_t max_pos = sizeof...(Ts) - 1;
tup_ptr_access_pos<(10 < max_pos ? 10 : max_pos), max_pos> first;
return rec_dispatch(pos, f, first);
}
}
template <class F, size_t N>
auto rec_dispatch(size_t, F& f, tup_ptr_access_pos<N, N>)
-> decltype(f(std::declval<int&>())) {
return tuple_inspect_delegate<N, N>(data_, f);
}
template <class F, size_t X, size_t N>
auto rec_dispatch(size_t pos, F& f, tup_ptr_access_pos<X, N> token)
-> decltype(f(std::declval<int&>())) {
return pos == X ? tuple_inspect_delegate<X, N>(data_, f)
: rec_dispatch(pos, f, next(token));
}
tuple_vals_impl* mptr() const {
return const_cast<tuple_vals_impl*>(this);
}
type_id_list types_;
data_type data_;
};
template <class... Ts>
class tuple_vals : public tuple_vals_impl<message_data, Ts...> {
public:
static_assert(sizeof...(Ts) > 0, "tuple_vals is not allowed to be empty");
using super = tuple_vals_impl<message_data, Ts...>;
using super::super;
using super::copy;
tuple_vals* copy() const override {
return new tuple_vals(*this);
}
};
} // namespace caf::detail
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <cstddef>
#include <cstdint>
#include <functional>
#include <tuple>
#include <typeinfo>
#include "caf/detail/apply_args.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/type_erased_tuple.hpp"
#include "caf/type_erased_value.hpp"
#include "caf/type_id_list.hpp"
namespace caf::detail {
template <class... Ts>
class type_erased_tuple_view : public type_erased_tuple {
public:
// -- member types -----------------------------------------------------------
template <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 --------------------
type_erased_tuple_view(Ts&... xs) : xs_(xs...) {
init();
}
type_erased_tuple_view(const type_erased_tuple_view& other)
: type_erased_tuple(), xs_(other.xs_) {
init();
}
type_erased_tuple_view& operator=(type_erased_tuple_view&&) = delete;
type_erased_tuple_view& operator=(const type_erased_tuple_view&) = delete;
// -- overridden modifiers ---------------------------------------------------
void* get_mutable(size_t pos) override {
return ptrs_[pos]->get_mutable();
}
error load(size_t pos, deserializer& source) override {
return ptrs_[pos]->load(source);
}
error_code<sec> load(size_t pos, binary_deserializer& source) override {
return ptrs_[pos]->load(source);
}
// -- overridden observers ---------------------------------------------------
size_t size() const noexcept override {
return sizeof...(Ts);
}
type_id_list types() const noexcept override {
return make_type_id_list<Ts...>();
}
type_id_t type(size_t pos) const noexcept override {
auto xs = types();
return xs[pos];
}
const void* get(size_t pos) const noexcept override {
return ptrs_[pos]->get();
}
std::string stringify(size_t pos) const override {
return ptrs_[pos]->stringify();
}
type_erased_value_ptr copy(size_t pos) const override {
return ptrs_[pos]->copy();
}
error save(size_t pos, serializer& sink) const override {
return ptrs_[pos]->save(sink);
}
error_code<sec> save(size_t pos, binary_serializer& sink) const override {
return ptrs_[pos]->save(sink);
}
// -- member variables access ------------------------------------------------
tuple_type& data() {
return xs_;
}
const tuple_type& data() const {
return xs_;
}
private:
// -- pointer "lookup table" utility -----------------------------------------
template <size_t N>
void init(num_token<N>, num_token<N>) {
// end of recursion
}
template <size_t P, size_t N>
void init(num_token<P>, num_token<N> last) {
ptrs_[P] = &std::get<P>(xs_);
init(num_token<P + 1>{}, last);
}
void init() {
init(num_token<0>{}, num_token<sizeof...(Ts)>{});
}
// -- data members -----------------------------------------------------------
tuple_type xs_;
type_erased_value* ptrs_[sizeof...(Ts) == 0 ? 1 : sizeof...(Ts)];
};
} // namespace caf::detail
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <cstdint>
#include <functional>
#include <typeinfo>
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/deserializer.hpp"
#include "caf/detail/safe_equal.hpp"
#include "caf/detail/try_serialize.hpp"
#include "caf/error.hpp"
#include "caf/serializer.hpp"
#include "caf/type_erased_value.hpp"
#include "caf/type_id.hpp"
namespace caf::detail {
/// @relates type_erased_value
/// Default implementation for single type-erased values.
template <class T>
class type_erased_value_impl : public type_erased_value {
public:
// -- member types -----------------------------------------------------------
using value_type = typename detail::strip_reference_wrapper<T>::type;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
type_erased_value_impl(Ts&&... xs) : x_(std::forward<Ts>(xs)...) {
// nop
}
template <class U, size_t N,
class = typename std::enable_if<std::is_same<T, U[N]>::value>::type>
type_erased_value_impl(const U (&ys)[N]) {
array_copy(x_, ys);
}
template <class U, size_t N,
class = typename std::enable_if<std::is_same<T, U[N]>::value>::type>
type_erased_value_impl(const U(&&ys)[N]) {
array_copy(x_, ys);
}
type_erased_value_impl(type_erased_value_impl&& other)
: type_erased_value_impl(std::move(other.x_)) {
// nop
}
type_erased_value_impl(const type_erased_value_impl& other)
: type_erased_value_impl(other.x_) {
// nop
}
// -- overridden modifiers ---------------------------------------------------
void* get_mutable() override {
return addr_of(x_);
}
error load(deserializer& source) override {
return source(*addr_of(x_));
}
error_code<sec> load(binary_deserializer& source) override {
return source(*addr_of(x_));
}
// -- overridden observers ---------------------------------------------------
type_id_t type() const override {
return type_id_v<value_type>;
}
const void* get() const override {
// const is restored when returning from the function
return addr_of(const_cast<T&>(x_));
}
error save(serializer& sink) const override {
return sink(*addr_of(const_cast<T&>(x_)));
}
error_code<sec> save(binary_serializer& sink) const override {
return sink(*addr_of(const_cast<T&>(x_)));
}
std::string stringify() const override {
return deep_to_string(x_);
}
type_erased_value_ptr copy() const override {
return type_erased_value_ptr{new type_erased_value_impl(x_)};
}
// -- conversion operators ---------------------------------------------------
operator value_type&() {
return x_;
}
operator const value_type&() const {
return x_;
}
private:
// -- address-of-member utility ----------------------------------------------
template <class U>
static inline U* addr_of(const U& x) {
return const_cast<U*>(&x);
}
template <class U, size_t S>
static inline U* addr_of(const U (&x)[S]) {
return const_cast<U*>(static_cast<const U*>(x));
}
template <class U>
static inline U* addr_of(std::reference_wrapper<U>& x) {
return &x.get();
}
template <class U>
static inline U* addr_of(const std::reference_wrapper<U>& x) {
return &x.get();
}
// -- array copying utility --------------------------------------------------
template <class U, size_t Len>
static void array_copy_impl(U (&x)[Len], const U (&y)[Len], std::true_type) {
for (size_t i = 0; i < Len; ++i)
array_copy(x[i], y[i]);
}
template <class U, size_t Len>
static void array_copy_impl(U (&x)[Len], const U (&y)[Len], std::false_type) {
std::copy(y, y + Len, x);
}
template <class U, size_t Len>
static void array_copy(U (&x)[Len], const U (&y)[Len]) {
std::integral_constant<bool, std::is_array<U>::value> token;
array_copy_impl(x, y, token);
}
// -- data members -----------------------------------------------------------
T x_;
};
} // namespace caf::detail
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Copyright 2011-2020 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 *
......@@ -18,32 +18,43 @@
#pragma once
#include <cstdint>
#include <functional>
#include <typeinfo>
#include "caf/detail/type_erased_value_impl.hpp"
#include "caf/type_erased_value.hpp"
namespace caf {
/// @relates type_erased_value
/// Creates a type-erased value of type `T` from `xs`.
template <class T, class... Ts>
type_erased_value_ptr make_type_erased_value(Ts&&... xs) {
type_erased_value_ptr result;
result.reset(new detail::type_erased_value_impl<T>(std::forward<Ts>(xs)...));
return result;
}
/// @relates type_erased_value
/// Converts values to type-erased values.
struct type_erased_value_factory {
template <class T>
type_erased_value_ptr operator()(T&& x) const {
return make_type_erased_value<typename std::decay<T>::type>(
std::forward<T>(x));
#include <cstdlib>
#include "caf/fwd.hpp"
#include "caf/type_id.hpp"
namespace caf::detail {
class type_id_list_builder {
public:
static constexpr size_t block_size = 8;
type_id_list_builder();
~type_id_list_builder();
void reserve(size_t new_capacity);
void push_back(type_id_t id);
/// Returns the number of elements currenty stored in the array.
size_t size() const noexcept;
/// @pre `index < size()`
type_id_t operator[](size_t index) const noexcept;
/// Convertes the internal buffer to a ::type_id_list and returns it.
/// @pre `push_back` was called at least once
type_id_list to_list() noexcept;
void clear() noexcept {
size_ = 0;
}
private:
size_t size_;
size_t reserved_;
type_id_t* storage_;
};
} // namespace caf
} // namespace caf::detail
......@@ -21,7 +21,7 @@
#include <deque>
#include <vector>
#include "caf/make_message.hpp"
#include "caf/message.hpp"
namespace caf {
......
......@@ -27,6 +27,7 @@
#include "caf/error_category.hpp"
#include "caf/error_code.hpp"
#include "caf/fwd.hpp"
#include "caf/message.hpp"
#include "caf/meta/load_callback.hpp"
#include "caf/meta/omittable_if_empty.hpp"
#include "caf/meta/save_callback.hpp"
......@@ -97,7 +98,8 @@ public:
}
template <class Enum>
error(error_code<Enum> code) : error(code.value()) {
error(error_code<Enum> code)
: error(static_cast<uint8_t>(code.value()), error_category<Enum>::value) {
// nop
}
......
......@@ -69,7 +69,8 @@ struct downstream_manager_selector {
template <class T, class... Ts>
downstream_manager* operator()(const message& msg, T& x, Ts&... xs) {
if (msg.match_element<stream<typename T::value_type>>(0))
if (msg.size() > 1
&& msg.type_at(0) == type_id_v<stream<typename T::value_type>>)
return &x;
return (*this)(msg, xs...);
}
......
......@@ -139,8 +139,8 @@ class stream_manager;
class string_view;
class tracing_data;
class tracing_data_factory;
class type_erased_tuple;
class type_erased_value;
class type_id_list;
class type_id_list_builder;
class uri;
class uri_builder;
......@@ -199,9 +199,6 @@ using type_id_t = uint16_t;
/// @relates actor_system_config
CAF_CORE_EXPORT const settings& content(const actor_system_config&);
template <class T, class... Ts>
message make_message(T&& x, Ts&&... xs);
// -- intrusive containers -----------------------------------------------------
namespace intrusive {
......@@ -264,9 +261,6 @@ class manager;
namespace detail {
template <class>
class type_erased_value_impl;
template <class>
class stream_distribution_tree;
......@@ -309,6 +303,5 @@ using stream_manager_ptr = intrusive_ptr<stream_manager>;
using mailbox_element_ptr = std::unique_ptr<mailbox_element>;
using tracing_data_ptr = std::unique_ptr<tracing_data>;
using type_erased_value_ptr = std::unique_ptr<type_erased_value>;
} // namespace caf
......@@ -22,22 +22,13 @@
#include <memory>
#include "caf/actor_control_block.hpp"
#include "caf/config.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/tuple_vals.hpp"
#include "caf/detail/type_erased_tuple_view.hpp"
#include "caf/extend.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/make_message.hpp"
#include "caf/memory_managed.hpp"
#include "caf/message.hpp"
#include "caf/message_id.hpp"
#include "caf/meta/omittable_if_empty.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/ref_counted.hpp"
#include "caf/tracing_data.hpp"
#include "caf/type_erased_tuple.hpp"
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 <string>
#include <tuple>
#include <type_traits>
#include "caf/allowed_unsafe_message_type.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/deserializer.hpp"
#include "caf/detail/inspect.hpp"
#include "caf/detail/tuple_vals.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/message.hpp"
#include "caf/serializer.hpp"
namespace caf {
/// Unboxes atom constants, i.e., converts `atom_constant<V>` to `V`.
/// @relates message
template <class T, int IsPlaceholderRes = std::is_placeholder<T>::value>
struct unbox_message_element {
using type = index_mapping;
};
template <class T>
struct unbox_message_element<T, 0> {
using type = T;
};
template <>
struct unbox_message_element<actor_control_block*, 0> {
using type = strong_actor_ptr;
};
///
template <class T>
struct is_serializable_or_whitelisted {
static constexpr bool value
= std::is_arithmetic<T>::value //
|| std::is_empty<T>::value //
|| std::is_enum<T>::value //
|| detail::is_stl_tuple_type<T>::value //
|| detail::is_map_like<T>::value //
|| detail::is_list_like<T>::value
|| (detail::is_inspectable<binary_serializer, T>::value
&& detail::is_inspectable<binary_deserializer, T>::value
&& detail::is_inspectable<serializer, T>::value
&& detail::is_inspectable<deserializer, T>::value)
|| allowed_unsafe_message_type<T>::value;
};
template <>
struct is_serializable_or_whitelisted<byte> : std::true_type {};
template <>
struct is_serializable_or_whitelisted<std::string> : std::true_type {};
template <>
struct is_serializable_or_whitelisted<std::u16string> : std::true_type {};
template <>
struct is_serializable_or_whitelisted<std::u32string> : std::true_type {};
/// Returns a new `message` containing the values `(x, xs...)`.
/// @relates message
template <class T, class... Ts>
message make_message(T&& x, Ts&&... xs) {
if constexpr (sizeof...(Ts) == 0
&& std::is_same<message, std::decay_t<T>>::value) {
return std::forward<T>(x);
} else {
using namespace detail;
// clang-format off
static_assert(is_complete<type_id<strip_and_convert_t<T>>>
&& (is_complete<type_id<strip_and_convert_t<Ts>>> && ...),
"at least one type has no ID, "
"did you forgot to announce it via CAF_ADD_TYPE_ID?");
// clang-format on
using storage
= tuple_vals<strip_and_convert_t<T>, strip_and_convert_t<Ts>...>;
auto ptr = make_counted<storage>(std::forward<T>(x),
std::forward<Ts>(xs)...);
return message{detail::message_data::cow_ptr{std::move(ptr)}};
}
}
/// Returns an empty `message`.
/// @relates message
inline message make_message() {
return message{};
}
struct message_factory {
template <class... Ts>
message operator()(Ts&&... xs) const {
return make_message(std::forward<Ts>(xs)...);
}
};
/// Converts the tuple `xs` to a message.
template <class... Ts>
message make_message_from_tuple(std::tuple<Ts...> xs) {
message_factory f;
return detail::apply_moved_args(f, detail::get_indices(xs), xs);
}
} // namespace caf
......@@ -22,163 +22,196 @@
#include <tuple>
#include <type_traits>
#include "caf/config.hpp"
#include "caf/detail/apply_args.hpp"
#include "caf/detail/comparable.hpp"
#include "caf/detail/core_export.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"
#include "caf/detail/padded_size.hpp"
#include "caf/fwd.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/intrusive_cow_ptr.hpp"
namespace caf {
class message_handler;
/// Describes a fixed-length, copy-on-write, type-erased
/// tuple with elements of any type.
class CAF_CORE_EXPORT message : public type_erased_tuple {
class CAF_CORE_EXPORT message {
public:
// -- member types -----------------------------------------------------------
/// Raw pointer to content.
using raw_ptr = detail::message_data*;
/// Copy-on-write pointer to content.
using data_ptr = detail::message_data::cow_ptr;
using data_ptr = intrusive_cow_ptr<detail::message_data>;
// -- constructors, destructors, and assignment operators --------------------
message() noexcept = default;
message(none_t) noexcept;
message(const message&) noexcept = default;
message& operator=(const message&) noexcept = default;
explicit message(data_ptr data) noexcept : data_(std::move(data)) {
// nop
}
message(message&&) noexcept;
message& operator=(message&&) noexcept;
explicit message(data_ptr ptr) noexcept;
message() noexcept = default;
~message() override;
message(message&) noexcept = default;
// -- implementation of type_erased_tuple ------------------------------------
message(const message&) noexcept = default;
void* get_mutable(size_t p) override;
message& operator=(message&) noexcept = default;
error load(size_t pos, deserializer& source) override;
message& operator=(const message&) noexcept = default;
error_code<sec> load(size_t pos, binary_deserializer& source) override;
// -- properties -------------------------------------------------------------
size_t size() const noexcept override;
auto types() const noexcept {
return data_ ? data_->types() : make_type_id_list();
}
type_id_list types() const noexcept override;
size_t size() const noexcept {
return types().size();
}
type_id_t type(size_t pos) const noexcept override;
size_t empty() const noexcept {
return size() == 0;
}
const void* get(size_t pos) const noexcept override;
template <class... Ts>
bool match_elements() const noexcept {
return types() == make_type_id_list<Ts...>();
}
std::string stringify(size_t pos) const override;
/// @private
detail::message_data& data() {
return data_.unshared();
}
type_erased_value_ptr copy(size_t pos) const override;
/// @private
const detail::message_data& data() const noexcept {
return *data_;
}
error save(size_t pos, serializer& sink) const override;
/// @private
const detail::message_data& cdata() const noexcept {
return *data_;
}
error_code<sec> save(size_t pos, binary_serializer& sink) const override;
/// @private
detail::message_data* ptr() noexcept {
return data_.unshared_ptr();
}
bool shared() const noexcept override;
/// @private
const detail::message_data* ptr() const noexcept {
return data_.get();
}
error load(deserializer& source) override;
/// @private
const detail::message_data* cptr() const noexcept {
return data_.get();
}
error_code<sec> load(binary_deserializer& source) override;
constexpr operator bool() const {
return static_cast<bool>(data_);
}
error save(serializer& sink) const override;
constexpr bool operator!() const {
return !data_;
}
error_code<sec> save(binary_serializer& sink) const override;
// -- serialization ----------------------------------------------------------
// -- factories --------------------------------------------------------------
error save(serializer& sink) const;
/// Creates a new message by copying all elements in a type-erased tuple.
static message copy(const type_erased_tuple& xs);
error_code<sec> save(binary_serializer& sink) const;
// -- modifiers --------------------------------------------------------------
error load(deserializer& source);
/// Returns `handler(*this)`.
optional<message> apply(message_handler handler);
error_code<sec> load(binary_deserializer& source);
/// Forces the message to copy its content if there are more than
/// one references to the content.
inline void force_unshare() {
vals_.unshare();
}
// -- element access ---------------------------------------------------------
/// Returns a mutable reference to the content. Callers are responsible
/// for unsharing content if necessary.
inline data_ptr& vals() {
return vals_;
/// Returns the type ID of the element at `index`.
/// @pre `index < size()`
type_id_t type_at(size_t index) const noexcept {
auto xs = types();
return xs[index];
}
/// Exchanges content of `this` and `other`.
void swap(message& other) noexcept;
/// Assigns new content.
void reset(raw_ptr new_ptr = nullptr, bool add_ref = true) noexcept;
// -- inline observers -------------------------------------------------------
/// Returns a const pointer to the element at position `p`.
inline const void* at(size_t p) const noexcept {
CAF_ASSERT(vals_ != nullptr);
return vals_->get(p);
/// Returns whether the element at `index` is of type `T`.
/// @pre `index < size()`
template <class T>
bool match_element(size_t index) const noexcept {
return type_at(index) == type_id_v<T>;
}
/// Returns a reference to the content.
inline const data_ptr& vals() const noexcept {
return vals_;
/// @pre `index < size()`
/// @pre `match_element<T>(index)`
template <class T>
const T& get_as(size_t index) const noexcept {
CAF_ASSERT(type_at(index) == type_id_v<T>);
return *reinterpret_cast<const T*>(data_->at(index));
}
/// Returns a reference to the content.
inline const data_ptr& cvals() const noexcept {
return vals_;
/// @pre `index < size()`
/// @pre `match_element<T>(index)`
template <class T>
T& get_mutable_as(size_t index) noexcept {
CAF_ASSERT(type_at(index) == type_id_v<T>);
return *reinterpret_cast<T*>(data_.unshared().at(index));
}
/// @cond PRIVATE
// -- modifiers --------------------------------------------------------------
/// @pre `!empty()`
inline type_erased_tuple& content() {
CAF_ASSERT(vals_ != nullptr);
return vals_.unshared();
void swap(message& other) noexcept {
data_.swap(other.data_);
}
inline const type_erased_tuple& content() const {
CAF_ASSERT(vals_ != nullptr);
return *vals_;
void reset(detail::message_data* new_ptr = nullptr,
bool add_ref = true) noexcept {
data_.reset(new_ptr, add_ref);
}
/// Serializes the content of `x` as if `x` was an instance of `message`. The
/// resulting output of `sink` can then be used to deserialize a `message`
/// even if the serialized object had a different type.
static error save(serializer& sink, const type_erased_tuple& x);
static error_code<sec>
save(binary_serializer& sink, const type_erased_tuple& x);
/// @endcond
private:
data_ptr vals_;
data_ptr data_;
};
// -- related non-members ------------------------------------------------------
/// @relates message
CAF_CORE_EXPORT error inspect(serializer& sink, message& msg);
inline message make_message() {
return {};
}
/// @relates message
template <class... Ts>
message make_message(Ts&&... xs) {
using namespace detail;
static_assert((!std::is_pointer<strip_and_convert_t<Ts>>::value && ...));
static constexpr size_t data_size = sizeof(message_data)
+ (padded_size_v<std::decay_t<Ts>> + ...);
auto types = make_type_id_list<strip_and_convert_t<Ts>...>();
auto vptr = malloc(data_size);
if (vptr == nullptr)
throw std::bad_alloc();
auto raw_ptr = new (vptr) message_data(types);
intrusive_cow_ptr<message_data> ptr{};
message_data_init(raw_ptr->storage(), std::forward<Ts>(xs)...);
return message{std::move(ptr)};
}
template <class Tuple, size_t... Is>
message make_message_from_tuple(Tuple&& xs, std::index_sequence<Is...>) {
return make_message(std::get<Is>(std::forward<Tuple>(xs))...);
}
template <class Tuple>
message make_message_from_tuple(Tuple&& xs) {
using tuple_type = std::decay_t<Tuple>;
std::make_index_sequence<std::tuple_size<tuple_type>::value> seq;
return make_message_from_tuple(std::forward<Tuple>(xs), seq);
}
/// @relates message
CAF_CORE_EXPORT error inspect(serializer& sink, const message& msg);
/// @relates message
CAF_CORE_EXPORT error_code<sec> inspect(binary_serializer& sink, message& msg);
CAF_CORE_EXPORT error_code<sec> inspect(binary_serializer& sink,
const message& msg);
/// @relates message
CAF_CORE_EXPORT error inspect(deserializer& source, message& msg);
......
......@@ -21,10 +21,11 @@
#include <vector>
#include "caf/detail/core_export.hpp"
#include "caf/detail/implicit_conversions.hpp"
#include "caf/detail/message_builder_element.hpp"
#include "caf/detail/type_id_list_builder.hpp"
#include "caf/fwd.hpp"
#include "caf/make_message.hpp"
#include "caf/message.hpp"
#include "caf/type_erased_value.hpp"
namespace caf {
......@@ -34,16 +35,15 @@ class CAF_CORE_EXPORT message_builder {
public:
friend class message;
message_builder() = default;
message_builder(const message_builder&) = delete;
message_builder& operator=(const message_builder&) = delete;
message_builder();
~message_builder();
message_builder& operator=(const message_builder&) = delete;
/// Creates a new instance and immediately calls `append(first, last)`.
template <class Iter>
message_builder(Iter first, Iter last) {
init();
append(first, last);
}
......@@ -58,35 +58,23 @@ public:
/// Adds `x` to the elements of the buffer.
template <class T>
message_builder& append(T&& x) {
using type =
typename unbox_message_element<typename detail::implicit_conversions<
typename std::decay<T>::type>::type>::type;
return emplace(make_type_erased_value<type>(std::forward<T>(x)));
}
inline message_builder& append_all() {
using value_type = detail::strip_and_convert<T>;
static_assert(detail::sendable<value_type>);
types_.push_back(type_id_v<value_type>);
elements_.emplace_back(make_message_builder_element(std::forward<T>(x)));
return *this;
}
template <class T, class... Ts>
message_builder& append_all(T&& x, Ts&&... xs) {
append(std::forward<T>(x));
return append_all(std::forward<Ts>(xs)...);
}
template <size_t N, class... Ts>
message_builder&
append_tuple(std::integral_constant<size_t, N>,
std::integral_constant<size_t, N>, std::tuple<Ts...>&) {
template <class... Ts>
message_builder& append_all(Ts&&... xs) {
(append(std::forward<Ts>(xs)), ...);
return *this;
}
template <size_t I, size_t N, class... Ts>
message_builder&
append_tuple(std::integral_constant<size_t, I>,
std::integral_constant<size_t, N> e, std::tuple<Ts...>& xs) {
append(std::move(std::get<I>(xs)));
return append_tuple(std::integral_constant<size_t, I + 1>{}, e, xs);
template <class Tuple, size_t... Is>
message_builder& append_tuple(Tuple& xs, std::index_sequence<Is...>) {
(append(std::get<Is>(xs)), ...);
return *this;
}
template <class... Ts>
......@@ -105,24 +93,22 @@ public:
/// is undefined behavior (dereferencing a `nullptr`)
message move_to_message();
/// @copydoc message::apply
optional<message> apply(message_handler handler);
/// Removes all elements from the buffer.
void clear();
void clear() noexcept;
/// Returns whether the buffer is empty.
bool empty() const;
bool empty() const noexcept {
return elements_.empty();
}
/// Returns the number of elements in the buffer.
size_t size() const;
size_t size() const noexcept {
return elements_.size();
}
private:
void init();
message_builder& emplace(type_erased_value_ptr);
intrusive_cow_ptr<detail::dynamic_message_data> data_;
detail::type_id_list_builder types_;
std::vector<detail::message_builder_element_ptr> elements_;
};
} // namespace caf
......@@ -90,17 +90,6 @@ public:
return (impl_) ? impl_->invoke(arg) : none;
}
/// Runs this handler and returns its (optional) result.
optional<message> operator()(type_erased_tuple& xs) {
return impl_ ? impl_->invoke(xs) : none;
}
/// Runs this handler with callback.
match_result operator()(detail::invoke_result_visitor& f,
type_erased_tuple& xs) {
return impl_ ? impl_->invoke(f, xs) : match_result::no_match;
}
/// Runs this handler with callback.
match_result operator()(detail::invoke_result_visitor& f, message& xs) {
return impl_ ? impl_->invoke(f, xs) : match_result::no_match;
......
......@@ -35,6 +35,8 @@
#include "caf/detail/functor_attachable.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/system_messages.hpp"
#include "caf/typed_message_view.hpp"
namespace caf {
......@@ -119,7 +121,7 @@ protected:
****************************************************************************/
// precondition: `mtx_` is acquired
inline void attach_impl(attachable_ptr& ptr) {
void attach_impl(attachable_ptr& ptr) {
ptr->next.swap(attachables_head_);
attachables_head_.swap(ptr);
}
......@@ -138,8 +140,8 @@ protected:
template <class F>
bool handle_system_message(mailbox_element& x, execution_unit* context,
bool trap_exit, F& down_msg_handler) {
if (x.content().match_elements<down_msg>()) {
down_msg_handler(x.content().get_mutable_as<down_msg>(0));
if (auto view = make_typed_message_view<down_msg>(x.payload)) {
down_msg_handler(get<0>(view));
return true;
}
return handle_system_message(x, context, trap_exit);
......
......@@ -30,8 +30,8 @@ template <class T>
class param {
public:
enum flag {
shared_access, // x_ lives in a shared type_erased_tuple
exclusive_access, // x_ lives in an unshared type_erased_tuple
shared_access, // x_ lives in a shared message
exclusive_access, // x_ lives in an unshared message
private_access, // x_ is a copy of the original value
};
......
......@@ -344,10 +344,9 @@ public:
/// Sets a custom handler for unexpected messages.
template <class F>
typename std::enable_if<std::is_convertible<
F, std::function<result<message>(type_erased_tuple&)>>::value>::type
F, std::function<result<message>(message&)>>::value>::type
set_default_handler(F fun) {
default_handler_
= [=](scheduled_actor*, const type_erased_tuple& xs) { return fun(xs); };
default_handler_ = [=](scheduled_actor*, message& xs) { return fun(xs); };
}
/// Sets a custom handler for error messages.
......@@ -550,8 +549,8 @@ public:
Finalize fin = {}, policy::arg<DownstreamManager> token = {}) {
CAF_IGNORE_UNUSED(token);
CAF_ASSERT(current_mailbox_element() != nullptr);
CAF_ASSERT(
current_mailbox_element()->content().match_elements<open_stream_msg>());
CAF_ASSERT(current_mailbox_element()->content().types()
== make_type_id_list<open_stream_msg>());
using output_type = typename stream_stage_trait_t<Fun>::output;
using state_type = typename stream_stage_trait_t<Fun>::state;
static_assert(
......
......@@ -64,9 +64,10 @@ public:
const T& peek() {
auto ptr = next_job<scheduled_actor>().mailbox().peek();
CAF_ASSERT(ptr != nullptr);
if (!ptr->content().match_elements<T>())
auto view = make_typed_message_view<T>(ptr->content());
if (!view)
CAF_RAISE_ERROR("Mailbox element does not match T.");
return ptr->content().get_as<T>(0);
return get<0>(view);
}
/// Puts `x` at the front of the queue unless it cannot be found in the queue.
......
......@@ -19,7 +19,6 @@
#pragma once
#include "caf/detail/type_traits.hpp"
#include "caf/make_message.hpp"
#include "caf/message.hpp"
#include "caf/stream_sink.hpp"
......
......@@ -30,7 +30,6 @@
#include "caf/downstream_msg.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/make_message.hpp"
#include "caf/message_builder.hpp"
#include "caf/ref_counted.hpp"
#include "caf/stream.hpp"
......
......@@ -23,7 +23,6 @@
#include <vector>
#include "caf/fwd.hpp"
#include "caf/make_message.hpp"
#include "caf/message.hpp"
#include "caf/stream_sink.hpp"
......
......@@ -18,7 +18,6 @@
#pragma once
#include "caf/make_message.hpp"
#include "caf/message.hpp"
#include "caf/stream_sink.hpp"
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <cstddef>
#include <cstdint>
#include <tuple>
#include <typeinfo>
#include <utility>
#include "caf/detail/apply_args.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/pseudo_tuple.hpp"
#include "caf/error_code.hpp"
#include "caf/fwd.hpp"
#include "caf/optional.hpp"
#include "caf/type_erased_value.hpp"
namespace caf {
/// Represents a tuple of type-erased values.
class CAF_CORE_EXPORT type_erased_tuple {
public:
// -- constructors, destructors, and assignment operators --------------------
type_erased_tuple() = default;
type_erased_tuple(const type_erased_tuple&) = default;
type_erased_tuple& operator=(const type_erased_tuple&) = default;
virtual ~type_erased_tuple();
// -- pure virtual modifiers -------------------------------------------------
/// Returns a mutable pointer to the element at position `pos`.
virtual void* get_mutable(size_t pos) = 0;
/// Load the content for the element at position `pos` from `source`.
virtual error load(size_t pos, deserializer& source) = 0;
/// Load the content for the element at position `pos` from `source`.
virtual error_code<sec> load(size_t pos, binary_deserializer& source) = 0;
// -- modifiers --------------------------------------------------------------
/// Load the content for the tuple from `source`.
virtual error load(deserializer& source);
/// Load the content for the tuple from `source`.
virtual error_code<sec> load(binary_deserializer& source);
// -- pure virtual observers -------------------------------------------------
/// Returns the size of this tuple.
virtual size_t size() const noexcept = 0;
/// Returns the type IDs for this tuple.
virtual type_id_list types() const noexcept = 0;
/// Returns the type ID for the element at position `pos`.
virtual type_id_t type(size_t pos) const noexcept = 0;
/// Returns the element at position `pos`.
virtual const void* get(size_t pos) const noexcept = 0;
/// Returns a string representation of the element at position `pos`.
virtual std::string stringify(size_t pos) const = 0;
/// Returns a copy of the element at position `pos`.
virtual type_erased_value_ptr copy(size_t pos) const = 0;
/// Saves the element at position `pos` to `sink`.
virtual error save(size_t pos, serializer& sink) const = 0;
/// Saves the element at position `pos` to `sink`.
virtual error_code<sec> save(size_t pos, binary_serializer& sink) const = 0;
// -- observers --------------------------------------------------------------
/// Returns whether multiple references to this tuple exist.
/// The default implementation returns false.
virtual bool shared() const noexcept;
/// Returns `size() == 0`.
bool empty() const;
/// Returns a string representation of the tuple.
std::string stringify() const;
/// Saves the content of the tuple to `sink`.
virtual error save(serializer& sink) const;
/// Saves the content of the tuple to `sink`.
virtual error_code<sec> save(binary_serializer& sink) const;
/// @private
type_erased_tuple* copy() const;
// -- convenience functions --------------------------------------------------
/// Convenience function for `*reinterpret_cast<const T*>(get())`.
template <class T>
const T& get_as(size_t pos) const {
return *reinterpret_cast<const T*>(get(pos));
}
template <class T, size_t Pos>
struct typed_index {};
template <class T, size_t Pos>
static constexpr typed_index<T, Pos> make_typed_index() {
return {};
}
template <class T, size_t Pos>
const T& get_as(typed_index<T, Pos>) const {
return *reinterpret_cast<const T*>(get(Pos));
}
template <class... Ts, long... Is>
std::tuple<const Ts&...>
get_as_tuple(detail::type_list<Ts...>, detail::int_list<Is...>) const {
return std::tuple<const Ts&...>{get_as<Ts>(Is)...};
// return get_as<Ts>(Is)...;//(make_typed_index<Ts, Is>())...;
}
template <class... Ts>
std::tuple<const Ts&...> get_as_tuple() const {
return get_as_tuple(detail::type_list<Ts...>{},
typename detail::il_range<0, sizeof...(Ts)>::type{});
}
/// Convenience function for `*reinterpret_cast<T*>(get_mutable())`.
template <class T>
T& get_mutable_as(size_t pos) {
return *reinterpret_cast<T*>(get_mutable(pos));
}
/// Convenience function for moving a value out of the tuple if it is
/// unshared. Returns a copy otherwise.
template <class T>
T move_if_unshared(size_t pos) {
if (shared())
return get_as<T>(pos);
return std::move(get_mutable_as<T>(pos));
}
/// Returns `true` if the element at `pos` matches `T`.
template <class T>
bool match_element(size_t pos) const noexcept {
CAF_ASSERT(pos < size());
return type(pos) == type_id_v<T>;
}
/// Returns `true` if the pattern `Ts...` matches the content of this tuple.
template <class... Ts>
bool match_elements() const noexcept {
return make_type_id_list<Ts...>() == types();
}
template <class... Ts>
bool match_elements(detail::type_list<Ts...>) const noexcept {
return make_type_id_list<Ts...>() == types();
}
template <class F>
auto apply(F fun)
-> optional<typename detail::get_callable_trait<F>::result_type> {
using trait = typename detail::get_callable_trait<F>::type;
detail::type_list<typename trait::result_type> result_token;
typename trait::arg_types args_token;
return apply(fun, result_token, args_token);
}
private:
template <class F, class R, class... Ts>
optional<R> apply(F& fun, detail::type_list<R>, detail::type_list<Ts...> tk) {
if (!match_elements<Ts...>())
return none;
detail::pseudo_tuple<typename std::decay<Ts>::type...> xs{*this};
return detail::apply_args(fun, detail::get_indices(tk), xs);
}
template <class F, class... Ts>
optional<void>
apply(F& fun, detail::type_list<void>, detail::type_list<Ts...> tk) {
if (!match_elements<std::decay_t<Ts>...>())
return none;
detail::pseudo_tuple<std::decay_t<Ts>...> xs{*this};
detail::apply_args(fun, detail::get_indices(tk), xs);
return unit;
}
};
/// @relates type_erased_tuple
CAF_CORE_EXPORT error inspect(serializer& sink, const type_erased_tuple& x);
/// @relates type_erased_tuple
CAF_CORE_EXPORT error inspect(deserializer& source, type_erased_tuple& x);
/// @relates type_erased_tuple
inline auto inspect(binary_serializer& sink, const type_erased_tuple& x) {
return x.save(sink);
}
/// @relates type_erased_tuple
inline auto inspect(binary_deserializer& source, type_erased_tuple& x) {
return x.load(source);
}
/// @relates type_erased_tuple
inline std::string to_string(const type_erased_tuple& x) {
return x.stringify();
}
/// @relates type_erased_tuple
/// Dummy objects representing empty tuples.
class CAF_CORE_EXPORT empty_type_erased_tuple : public type_erased_tuple {
public:
empty_type_erased_tuple() = default;
~empty_type_erased_tuple() override;
void* get_mutable(size_t pos) override;
error load(size_t pos, deserializer& source) override;
error_code<sec> load(size_t pos, binary_deserializer& source) override;
size_t size() const noexcept override;
type_id_list types() const noexcept override;
type_id_t 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;
error_code<sec> save(size_t pos, binary_serializer& sink) const override;
};
} // 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 <cstdint>
#include <functional>
#include <typeinfo>
#include "caf/detail/core_export.hpp"
#include "caf/error_code.hpp"
#include "caf/fwd.hpp"
#include "caf/type_id.hpp"
namespace caf {
/// Represents a single type-erased value.
class CAF_CORE_EXPORT type_erased_value {
public:
// -- constructors, destructors, and assignment operators --------------------
virtual ~type_erased_value();
// -- pure virtual modifiers -------------------------------------------------
/// Returns a mutable pointer to the stored value.
virtual void* get_mutable() = 0;
/// Load the content for the stored value from `source`.
virtual error load(deserializer& source) = 0;
/// Load the content for the stored value from `source`.
virtual error_code<sec> load(binary_deserializer& source) = 0;
// -- pure virtual observers -------------------------------------------------
/// Returns the type number and type information object for the stored value.
virtual type_id_t type() const = 0;
/// Returns a pointer to the stored value.
virtual const void* get() const = 0;
/// Saves the content of the stored value to `sink`.
virtual error save(serializer& sink) const = 0;
/// Saves the content of the stored value to `sink`.
virtual error_code<sec> save(binary_serializer& sink) const = 0;
/// Converts the stored value to a string.
virtual std::string stringify() const = 0;
/// Returns a copy of the stored value.
virtual type_erased_value_ptr copy() const = 0;
// -- convenience functions --------------------------------------------------
/// Convenience function for `reinterpret_cast<const T*>(get())`.
template <class T>
const T& get_as() const {
return *reinterpret_cast<const T*>(get());
}
/// Convenience function for `reinterpret_cast<T*>(get_mutable())`.
template <class T>
T& get_mutable_as() {
return *reinterpret_cast<T*>(get_mutable());
}
};
/// @relates type_erased_value_impl
CAF_CORE_EXPORT error inspect(serializer& f, const type_erased_value& x);
/// @relates type_erased_value_impl
CAF_CORE_EXPORT error inspect(deserializer& f, type_erased_value& x);
/// @relates type_erased_value_impl
inline auto inspect(binary_serializer& f, const type_erased_value& x) {
return x.save(f);
}
/// @relates type_erased_value_impl
inline auto inspect(binary_deserializer& f, type_erased_value& x) {
return x.load(f);
}
/// @relates type_erased_value_impl
inline auto to_string(const type_erased_value& x) {
return x.stringify();
}
} // namespace caf
......@@ -69,6 +69,16 @@ public:
return memcmp(data_, other.data_, (size() + 1) * sizeof(type_id_t));
}
/// Returns an iterator to the first type ID.
pointer begin() const noexcept {
return data_ + 1;
}
/// Returns the past-the-end iterator.
pointer end() const noexcept {
return begin() + size();
}
private:
pointer data_;
};
......
......@@ -63,9 +63,6 @@ public:
template <class>
friend class data_processor;
template <class>
friend class detail::type_erased_value_impl;
template <class...>
friend class typed_actor;
......
......@@ -18,6 +18,8 @@
#pragma once
#include "caf/detail/message_data.hpp"
#include "caf/detail/offset_at.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/message.hpp"
......@@ -26,48 +28,43 @@ namespace caf {
template <class... Ts>
class typed_message_view {
public:
explicit typed_message_view(message& msg)
: ptr_(msg.vals().unshared_ptr()), owning_(false) {
typed_message_view() noexcept : ptr_(nullptr) {
// nop
}
explicit typed_message_view(type_erased_tuple& msg) {
if (msg.shared()) {
ptr_ = msg.copy();
owning_ = true;
} else {
ptr_ = &msg;
owning_ = false;
}
explicit typed_message_view(message& msg) : ptr_(&msg.data()) {
// nop
}
typed_message_view() = delete;
typed_message_view(const typed_message_view&) noexcept = default;
typed_message_view& operator=(const typed_message_view&) noexcept = default;
~typed_message_view() {
if (owning_)
delete ptr_;
detail::message_data* operator->() noexcept {
return ptr_;
}
type_erased_tuple* operator->() noexcept {
return ptr_;
explicit operator bool() const noexcept {
return ptr_ != nullptr;
}
private:
type_erased_tuple* ptr_;
// Temporary hack until redesigning caf::message.
bool owning_;
detail::message_data* ptr_;
};
template <size_t Position, class... Ts>
template <size_t Index, class... Ts>
auto& get(typed_message_view<Ts...>& x) {
static_assert(Position < sizeof...(Ts));
using types = detail::type_list<Ts...>;
using type = detail::tl_at_t<types, Position>;
return *reinterpret_cast<type*>(x->get_mutable(Position));
static_assert(Index < sizeof...(Ts));
using type = caf::detail::tl_at_t<caf::detail::type_list<Ts...>, Index>;
return *reinterpret_cast<type*>(x->storage()
+ detail::offset_at<Index, Ts...>);
}
template <class... Ts>
auto make_typed_message_view(message& msg) {
if (msg.types() == make_type_id_list<Ts...>())
return typed_message_view<Ts...>{msg};
return typed_message_view<Ts...>{};
}
} // namespace caf
......@@ -147,11 +147,11 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
const strong_actor_ptr& sender, message_id mid,
message& content, execution_unit* eu) {
CAF_LOG_TRACE(CAF_ARG(mid) << CAF_ARG(content));
if (content.match_elements<exit_msg>()) {
if (auto view = make_const_typed_message_view<exit_msg>(content)) {
// acquire second mutex as well
std::vector<actor> workers;
auto em = content.get_as<exit_msg>(0).reason;
if (cleanup(std::move(em), eu)) {
auto reason = get<0>(view).reason;
if (cleanup(std::move(reason), eu)) {
// send exit messages *always* to all workers and clear vector afterwards
// but first swap workers_ out of the critical section
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
......@@ -163,9 +163,9 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
}
return true;
}
if (content.match_elements<down_msg>()) {
if (auto view = make_const_typed_message_view<down_msg>(content)) {
// remove failed worker from pool
auto& dm = content.get_as<down_msg>(0);
const auto& dm = get<0>(view);
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
auto last = workers_.end();
auto i = std::find(workers_.begin(), workers_.end(), dm.source);
......@@ -179,17 +179,19 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
}
return true;
}
if (content.match_elements<sys_atom, put_atom, actor>()) {
auto& worker = content.get_as<actor>(2);
if (auto view
= make_const_typed_message_view<sys_atom, put_atom, actor>(content)) {
const auto& worker = get<2>(view);
worker->attach(default_attachable::make_monitor(worker.address(),
address()));
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
workers_.push_back(worker);
return true;
}
if (content.match_elements<sys_atom, delete_atom, actor>()) {
if (auto view
= make_const_typed_message_view<sys_atom, delete_atom, actor>(content)) {
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
auto& what = content.get_as<actor>(2);
auto& what = get<2>(view);
auto last = workers_.end();
auto i = std::find(workers_.begin(), last, what);
if (i != last) {
......
......@@ -181,8 +181,8 @@ blocking_actor::mailbox_visitor::operator()(mailbox_element& x) {
return intrusive::task_result::skip;
}
// Automatically unlink from actors after receiving an exit.
if (x.content().match_elements<exit_msg>())
self->unlink_from(x.content().get_as<exit_msg>(0).source);
if (auto view = make_const_typed_message_view<exit_msg>(x.content()))
self->unlink_from(get<0>(view).source);
// Blocking actors can nest receives => push/pop `current_element_`
auto prev_element = self->current_element_;
self->current_element_ = &x;
......
......@@ -16,11 +16,10 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include <utility>
#include "caf/detail/behavior_impl.hpp"
#include "caf/make_type_erased_tuple_view.hpp"
#include <utility>
#include "caf/message_handler.hpp"
namespace caf::detail {
......@@ -29,8 +28,7 @@ namespace {
class combinator final : public behavior_impl {
public:
match_result invoke(detail::invoke_result_visitor& f,
type_erased_tuple& xs) override {
match_result invoke(detail::invoke_result_visitor& f, message& xs) override {
auto x = first->invoke(f, xs);
return x == match_result::no_match ? second->invoke(f, xs) : x;
}
......@@ -87,7 +85,7 @@ behavior_impl::behavior_impl(timespan tout) : timeout_(tout) {
}
match_result behavior_impl::invoke_empty(detail::invoke_result_visitor& f) {
auto xs = make_type_erased_tuple_view();
message xs;
return invoke(f, xs);
}
......@@ -96,24 +94,16 @@ optional<message> behavior_impl::invoke(message& xs) {
// the following const-cast is safe, because invoke() is aware of
// copy-on-write and does not modify x if it's shared
if (!xs.empty())
invoke(f, *const_cast<message_data*>(xs.cvals().get()));
invoke(f, xs);
else
invoke_empty(f);
return std::move(f.value);
}
optional<message> behavior_impl::invoke(type_erased_tuple& xs) {
maybe_message_visitor f;
invoke(f, xs);
return std::move(f.value);
}
match_result behavior_impl::invoke(detail::invoke_result_visitor& f,
message& xs) {
// the following const-cast is safe, because invoke() is aware of
// copy-on-write and does not modify x if it's shared
if (!xs.empty())
return invoke(f, *const_cast<message_data*>(xs.cvals().get()));
return invoke(f, xs);
return invoke_empty(f);
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/detail/concatenated_tuple.hpp"
#include <numeric>
#include "caf/make_counted.hpp"
#include "caf/message.hpp"
#include "caf/raise_error.hpp"
namespace caf::detail {
concatenated_tuple::concatenated_tuple(std::initializer_list<cow_ptr> xs) {
for (auto& x : xs) {
if (x) {
auto dptr = dynamic_cast<const concatenated_tuple*>(x.get());
if (dptr != nullptr) {
auto& vec = dptr->data_;
data_.insert(data_.end(), vec.begin(), vec.end());
} else {
data_.push_back(std::move(x));
}
}
}
auto acc_size
= [](size_t tmp, const cow_ptr& val) { return tmp + val->size(); };
size_ = std::accumulate(data_.begin(), data_.end(), size_t{0}, acc_size);
}
auto concatenated_tuple::make(std::initializer_list<cow_ptr> xs) -> cow_ptr {
return cow_ptr{make_counted<concatenated_tuple>(xs)};
}
concatenated_tuple* concatenated_tuple::copy() const {
return new concatenated_tuple(*this);
}
void* concatenated_tuple::get_mutable(size_t pos) {
CAF_ASSERT(pos < size());
auto selected = select(pos);
return selected.first->get_mutable(selected.second);
}
error concatenated_tuple::load(size_t pos, deserializer& source) {
CAF_ASSERT(pos < size());
auto selected = select(pos);
return selected.first->load(selected.second, source);
}
size_t concatenated_tuple::size() const noexcept {
return size_;
}
rtti_pair concatenated_tuple::type(size_t pos) const noexcept {
CAF_ASSERT(pos < size());
auto selected = select(pos);
return selected.first->type(selected.second);
}
const void* concatenated_tuple::get(size_t pos) const noexcept {
CAF_ASSERT(pos < size());
auto selected = select(pos);
return selected.first->get(selected.second);
}
std::string concatenated_tuple::stringify(size_t pos) const {
CAF_ASSERT(pos < size());
auto selected = select(pos);
return selected.first->stringify(selected.second);
}
type_erased_value_ptr concatenated_tuple::copy(size_t pos) const {
CAF_ASSERT(pos < size());
auto selected = select(pos);
return selected.first->copy(selected.second);
}
error concatenated_tuple::save(size_t pos, serializer& sink) const {
CAF_ASSERT(pos < size());
auto selected = select(pos);
return selected.first->save(selected.second, sink);
}
std::pair<message_data*, size_t> concatenated_tuple::select(size_t pos) {
auto idx = pos;
for (auto& m : data_) {
auto s = m->size();
if (idx >= s)
idx -= s;
else
return {m.unshared_ptr(), idx};
}
CAF_RAISE_ERROR(std::out_of_range, "concatenated_tuple::select out of range");
}
std::pair<const message_data*, size_t>
concatenated_tuple::select(size_t pos) const {
auto idx = pos;
for (const auto& m : data_) {
auto s = m->size();
if (idx >= s)
idx -= s;
else
return {m.get(), idx};
}
CAF_RAISE_ERROR(std::out_of_range, "concatenated_tuple::select out of range");
}
} // namespace caf::detail
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/detail/dynamic_message_data.hpp"
#include "caf/error.hpp"
#include "caf/intrusive_cow_ptr.hpp"
#include "caf/make_counted.hpp"
namespace caf::detail {
dynamic_message_data::dynamic_message_data() {
types_.push_back(0);
}
dynamic_message_data::dynamic_message_data(elements&& data)
: elements_(std::move(data)) {
types_.reserve(elements_.size() + 1);
types_.emplace_back(static_cast<type_id_t>(elements_.size()));
for (const auto& e : elements_)
types_.emplace_back(e->type());
}
dynamic_message_data::dynamic_message_data(const dynamic_message_data& other)
: detail::message_data(other), types_(other.types_) {
for (auto& e : other.elements_)
elements_.emplace_back(e->copy());
}
dynamic_message_data::~dynamic_message_data() {
// nop
}
dynamic_message_data* dynamic_message_data::copy() const {
return new dynamic_message_data(*this);
}
void* dynamic_message_data::get_mutable(size_t pos) {
CAF_ASSERT(pos < size());
return elements_[pos]->get_mutable();
}
error dynamic_message_data::load(size_t pos, deserializer& source) {
CAF_ASSERT(pos < size());
return elements_[pos]->load(source);
}
error_code<sec>
dynamic_message_data::load(size_t pos, binary_deserializer& source) {
CAF_ASSERT(pos < size());
return elements_[pos]->load(source);
}
size_t dynamic_message_data::size() const noexcept {
return elements_.size();
}
type_id_list dynamic_message_data::types() const noexcept {
return type_id_list{types_.data()};
}
type_id_t dynamic_message_data::type(size_t pos) const noexcept {
CAF_ASSERT(pos < size());
return elements_[pos]->type();
}
const void* dynamic_message_data::get(size_t pos) const noexcept {
CAF_ASSERT(pos < size());
return elements_[pos]->get();
}
std::string dynamic_message_data::stringify(size_t pos) const {
CAF_ASSERT(pos < size());
return elements_[pos]->stringify();
}
type_erased_value_ptr dynamic_message_data::copy(size_t pos) const {
CAF_ASSERT(pos < size());
return elements_[pos]->copy();
}
error dynamic_message_data::save(size_t pos, serializer& sink) const {
CAF_ASSERT(pos < size());
return elements_[pos]->save(sink);
}
error_code<sec>
dynamic_message_data::save(size_t pos, binary_serializer& sink) const {
CAF_ASSERT(pos < size());
return elements_[pos]->save(sink);
}
void dynamic_message_data::clear() {
elements_.clear();
types_.clear();
types_.push_back(0);
}
void dynamic_message_data::append(type_erased_value_ptr x) {
types_[0] += 1;
types_.emplace_back(x->type());
elements_.emplace_back(std::move(x));
}
void intrusive_ptr_add_ref(const dynamic_message_data* ptr) {
intrusive_ptr_add_ref(static_cast<const ref_counted*>(ptr));
}
void intrusive_ptr_release(const dynamic_message_data* ptr) {
intrusive_ptr_release(static_cast<const ref_counted*>(ptr));
}
dynamic_message_data* intrusive_cow_ptr_unshare(dynamic_message_data*& ptr) {
return default_intrusive_cow_ptr_unshare(ptr);
}
} // namespace caf::detail
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/detail/merged_tuple.hpp"
#include "caf/index_mapping.hpp"
#include "caf/system_messages.hpp"
#include "caf/detail/disposer.hpp"
namespace caf::detail {
merged_tuple::cow_ptr merged_tuple::make(message x, message y) {
data_type data{x.vals(), y.vals()};
mapping_type mapping;
auto s = x.size();
for (size_t i = 0; i < s; ++i) {
if (x.match_element<index_mapping>(i))
mapping.emplace_back(1, x.get_as<index_mapping>(i).value - 1);
else
mapping.emplace_back(0, i);
}
return cow_ptr{
make_counted<merged_tuple>(std::move(data), std::move(mapping))};
}
merged_tuple::merged_tuple(data_type xs, mapping_type ys)
: data_(std::move(xs)), mapping_(std::move(ys)) {
CAF_ASSERT(!data_.empty());
CAF_ASSERT(!mapping_.empty());
}
merged_tuple* merged_tuple::copy() const {
return new merged_tuple(data_, mapping_);
}
void* merged_tuple::get_mutable(size_t pos) {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first].unshared().get_mutable(p.second);
}
error merged_tuple::load(size_t pos, deserializer& source) {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first].unshared().load(p.second, source);
}
size_t merged_tuple::size() const noexcept {
return mapping_.size();
}
rtti_pair merged_tuple::type(size_t pos) const noexcept {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->type(p.second);
}
const void* merged_tuple::get(size_t pos) const noexcept {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->get(p.second);
}
std::string merged_tuple::stringify(size_t pos) const {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->stringify(p.second);
}
type_erased_value_ptr merged_tuple::copy(size_t pos) const {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->copy(p.second);
}
error merged_tuple::save(size_t pos, serializer& sink) const {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->save(p.second, sink);
}
const merged_tuple::mapping_type& merged_tuple::mapping() const {
return mapping_;
}
} // namespace caf::detail
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Copyright 2011-2020 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 *
......@@ -16,22 +16,12 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/type_erased_value.hpp"
#include "caf/detail/message_builder_element.hpp"
#include "caf/error.hpp"
namespace caf::detail {
namespace caf {
type_erased_value::~type_erased_value() {
message_builder_element::~message_builder_element() {
// nop
}
error inspect(serializer& f, const type_erased_value& x) {
return x.save(f);
}
error inspect(deserializer& f, type_erased_value& x) {
return x.load(f);
}
} // namespace caf
} // namespace caf::detail
......@@ -19,15 +19,123 @@
#include "caf/detail/message_data.hpp"
#include <cstring>
#include <numeric>
#include "caf/detail/meta_object.hpp"
#include "caf/error.hpp"
#include "caf/error_code.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
namespace caf::detail {
message_data::~message_data() {
message_data::message_data(type_id_list types)
: rc_(1), types_(std::move(types)) {
// nop
}
bool message_data::shared() const noexcept {
return !unique();
message_data::~message_data() noexcept {
// TODO: we unconditionally destroy all objects without some way of telling
// whether these objects were constructed in the first place.
auto gmos = global_meta_objects();
auto ptr = storage();
// TODO: C++ usually destroys members in reverse order.
for (auto id : types_) {
auto& meta = gmos[id];
meta.destroy(ptr);
ptr += meta.padded_size;
}
}
message_data* message_data::copy() const {
auto gmos = global_meta_objects();
auto add = [](size_t interim, const meta_object& meta) noexcept {
return interim + meta.padded_size;
};
auto storage_size = std::accumulate(gmos.begin(), gmos.end(), size_t{0}, add);
auto total_size = sizeof(message_data) + storage_size;
auto vptr = malloc(total_size);
if (vptr == nullptr)
throw std::bad_alloc();
auto ptr = new (vptr) message_data(types_);
auto src = storage();
auto dst = ptr->storage();
for (auto id : types_) {
auto& meta = gmos[id];
// TODO: exception handling.
meta.copy_construct(src, dst);
src += meta.padded_size;
dst += meta.padded_size;
}
return ptr;
}
byte* message_data::at(size_t index) noexcept {
if (index == 0)
return storage();
auto gmos = global_meta_objects();
auto ptr = storage();
for (size_t i = 0; i < index; ++i)
ptr += gmos[types_[i]].padded_size;
return ptr;
}
const byte* message_data::at(size_t index) const noexcept {
if (index == 0)
return storage();
auto gmos = global_meta_objects();
auto ptr = storage();
for (size_t i = 0; i < index; ++i)
ptr += gmos[types_[i]].padded_size;
return ptr;
}
caf::error message_data::save(caf::serializer& sink) const {
auto gmos = global_meta_objects();
auto ptr = storage();
for (auto id : types_) {
auto& meta = gmos[id];
if (auto err = meta.save(sink, ptr))
return err;
ptr += meta.padded_size;
}
return none;
}
caf::error message_data::save(caf::binary_serializer& sink) const {
auto gmos = global_meta_objects();
auto ptr = storage();
for (auto id : types_) {
auto& meta = gmos[id];
if (auto err = meta.save_binary(sink, ptr))
return err;
ptr += meta.padded_size;
}
return none;
}
caf::error message_data::load(caf::deserializer& source) {
auto gmos = global_meta_objects();
auto ptr = storage();
for (auto id : types_) {
auto& meta = gmos[id];
if (auto err = meta.load(source, ptr))
return err;
ptr += meta.padded_size;
}
return none;
}
caf::error message_data::load(caf::binary_deserializer& source) {
auto gmos = global_meta_objects();
auto ptr = storage();
for (auto id : types_) {
auto& meta = gmos[id];
if (auto err = meta.load_binary(source, ptr))
return err;
ptr += meta.padded_size;
}
return none;
}
} // namespace caf::detail
......@@ -24,6 +24,8 @@
#include <cstring>
#include "caf/config.hpp"
#include "caf/error.hpp"
#include "caf/error_code.hpp"
#include "caf/span.hpp"
namespace caf::detail {
......@@ -51,6 +53,25 @@ struct meta_objects_cleanup {
} // namespace
caf::error save(const meta_object& meta, caf::serializer& sink,
const void* obj) {
return meta.save(sink, obj);
}
caf::error_code<sec> save(const meta_object& meta, caf::binary_serializer& sink,
const void* obj) {
return meta.save_binary(sink, obj);
}
caf::error load(const meta_object& meta, caf::deserializer& source, void* obj) {
return meta.load(source, obj);
}
caf::error_code<sec> load(const meta_object& meta,
caf::binary_deserializer& source, void* obj) {
return meta.load_binary(source, obj);
}
span<const meta_object> global_meta_objects() {
return {meta_objects, meta_objects_size};
}
......
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Copyright 2011-2020 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 *
......@@ -16,72 +16,70 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/detail/decorated_tuple.hpp"
#include "caf/detail/type_id_list_builder.hpp"
#include "caf/make_counted.hpp"
#include "caf/config.hpp"
#include "caf/type_id_list.hpp"
namespace caf::detail {
decorated_tuple::decorated_tuple(cow_ptr&& d, vector_type&& v)
: decorated_(std::move(d)), mapping_(std::move(v)) {
CAF_ASSERT(mapping_.empty()
|| *(std::max_element(mapping_.begin(), mapping_.end()))
< static_cast<const cow_ptr&>(decorated_)->size());
type_id_list_builder::type_id_list_builder()
: size_(0), reserved_(0), storage_(nullptr) {
// nop
}
decorated_tuple::cow_ptr decorated_tuple::make(cow_ptr d, vector_type v) {
auto ptr = dynamic_cast<const decorated_tuple*>(d.get());
if (ptr != nullptr) {
d = ptr->decorated();
auto& pmap = ptr->mapping();
for (auto& i : v)
i = pmap[i];
}
auto res = make_counted<decorated_tuple>(std::move(d), std::move(v));
return decorated_tuple::cow_ptr{res};
}
message_data* decorated_tuple::copy() const {
return new decorated_tuple(*this);
}
void* decorated_tuple::get_mutable(size_t pos) {
CAF_ASSERT(pos < size());
return decorated_.unshared().get_mutable(mapping_[pos]);
type_id_list_builder::~type_id_list_builder() {
free(storage_);
}
error decorated_tuple::load(size_t pos, deserializer& source) {
CAF_ASSERT(pos < size());
return decorated_.unshared().load(mapping_[pos], source);
void type_id_list_builder::reserve(size_t new_capacity) {
if (reserved_ >= new_capacity)
return;
reserved_ = new_capacity;
auto ptr = realloc(storage_, reserved_ * sizeof(type_id_t));
if (ptr == nullptr)
throw std::bad_alloc();
}
size_t decorated_tuple::size() const noexcept {
return mapping_.size();
}
rtti_pair decorated_tuple::type(size_t pos) const noexcept {
CAF_ASSERT(pos < size());
return decorated_->type(mapping_[pos]);
void type_id_list_builder::push_back(type_id_t id) {
if ((size_ + 1) >= reserved_) {
reserved_ += block_size;
auto ptr = realloc(storage_, reserved_ * sizeof(type_id_t));
if (ptr == nullptr)
throw std::bad_alloc();
storage_ = reinterpret_cast<type_id_t*>(ptr);
// Add the dummy for later inserting the size on first push_back.
if (size_ == 0)
storage_[0] = 0;
}
storage_[++size_] = id;
}
const void* decorated_tuple::get(size_t pos) const noexcept {
CAF_ASSERT(pos < size());
return decorated_->get(mapping_[pos]);
size_t type_id_list_builder::size() const noexcept {
// Index 0 is reserved for storing the (final) size, i.e., does not contain a
// type ID.
return size_ > 0 ? size_ - 1 : 0;
}
std::string decorated_tuple::stringify(size_t pos) const {
CAF_ASSERT(pos < size());
return decorated_->stringify(mapping_[pos]);
type_id_t type_id_list_builder::operator[](size_t index) const noexcept{
CAF_ASSERT(index<size());
return storage_[index + 1];
}
type_erased_value_ptr decorated_tuple::copy(size_t pos) const {
CAF_ASSERT(pos < size());
return decorated_->copy(mapping_[pos]);
type_id_list type_id_list_builder::to_list() noexcept {
// TODO: implement me!
return make_type_id_list();
}
error decorated_tuple::save(size_t pos, serializer& sink) const {
CAF_ASSERT(pos < size());
return decorated_->save(mapping_[pos], sink);
}
// type_id_list type_id_list_builder::copy_to_list() {
// storage_[0]
// = static_cast<type_id_t>(size_ & type_id_list::dynamically_allocated_flag);
// auto ptr = malloc((size_ + 1) * sizeof(type_id_t));
// if (ptr == nullptr)
// throw std::bad_alloc(
// "type_id_list_builder::copy_to_list failed to allocate memory");
// memcpy(ptr, storage_, (size_ + 1) * sizeof(type_id_t));
// return type_id_list{reinterpret_cast<type_id_t*>(ptr)};
// }
} // namespace caf::detail
......@@ -75,7 +75,7 @@ error& error::operator=(const error& x) {
}
error::error(uint8_t x, uint8_t y)
: data_(x != 0 ? new data{x, y, none} : nullptr) {
: data_(x != 0 ? new data{x, y, message{}} : nullptr) {
// nop
}
......
......@@ -18,10 +18,14 @@
#include "caf/group.hpp"
#include "caf/message.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/deserializer.hpp"
#include "caf/group_manager.hpp"
#include "caf/message.hpp"
#include "caf/serializer.hpp"
namespace caf {
......
......@@ -23,7 +23,6 @@
#include "caf/detail/network_order.hpp"
#include "caf/detail/parser/read_ipv4_address.hpp"
#include "caf/error.hpp"
#include "caf/make_message.hpp"
#include "caf/message.hpp"
#include "caf/parser_state.hpp"
#include "caf/pec.hpp"
......
......@@ -25,7 +25,6 @@
#include "caf/detail/parser/read_ipv6_address.hpp"
#include "caf/error.hpp"
#include "caf/ipv4_address.hpp"
#include "caf/make_message.hpp"
#include "caf/message.hpp"
#include "caf/parser_state.hpp"
#include "caf/pec.hpp"
......
......@@ -21,9 +21,11 @@
#include <utility>
#include "caf/actor_system.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/deserializer.hpp"
#include "caf/detail/dynamic_message_data.hpp"
#include "caf/detail/meta_object.hpp"
#include "caf/detail/type_id_list_builder.hpp"
#include "caf/message_builder.hpp"
#include "caf/message_handler.hpp"
#include "caf/serializer.hpp"
......@@ -31,207 +33,116 @@
namespace caf {
message::message(none_t) noexcept {
// nop
}
message::message(message&& other) noexcept : vals_(std::move(other.vals_)) {
// nop
}
message::message(data_ptr ptr) noexcept : vals_(std::move(ptr)) {
// nop
}
message& message::operator=(message&& other) noexcept {
vals_.swap(other.vals_);
return *this;
}
message::~message() {
// nop
}
// -- implementation of type_erased_tuple --------------------------------------
void* message::get_mutable(size_t p) {
CAF_ASSERT(vals_ != nullptr);
return vals_.unshared().get_mutable(p);
}
error message::load(size_t pos, deserializer& source) {
CAF_ASSERT(vals_ != nullptr);
return vals_.unshared().load(pos, source);
}
error_code<sec> message::load(size_t pos, binary_deserializer& source) {
CAF_ASSERT(vals_ != nullptr);
return vals_.unshared().load(pos, source);
}
size_t message::size() const noexcept {
return vals_ != nullptr ? vals_->size() : 0;
}
type_id_list message::types() const noexcept {
return vals_ != nullptr ? vals_->types() : make_type_id_list();
}
type_id_t 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 {
CAF_ASSERT(vals_ != nullptr);
return vals_->save(pos, sink);
}
error_code<sec> message::save(size_t pos, binary_serializer& sink) const {
CAF_ASSERT(vals_ != nullptr);
return vals_->save(pos, sink);
}
bool message::shared() const noexcept {
return vals_ != nullptr ? vals_->shared() : false;
}
namespace {
template <class Deserializer>
typename Deserializer::result_type
load_vals(Deserializer& source, message::data_ptr& vals) {
// Fetch size.
uint16_t num_elements = 0;
if (auto err = source(num_elements))
load_data(Deserializer& source, message::data_ptr& data) {
uint16_t ids_size = 0;
if (auto err = source.apply(ids_size))
return err;
// Short-circuit empty tuples.
if (num_elements == 0) {
vals.reset();
return {};
if (ids_size == 0) {
data.reset();
return caf::none;
}
// Fetch the meta information (type list).
// TODO: use a stack-allocated buffer for short type lists.
auto meta = detail::global_meta_objects();
std::vector<type_id_t> meta_info;
meta_info.resize(num_elements + 1);
meta_info[0] = num_elements;
for (size_t index = 1; index < (num_elements + 1); ++index) {
uint16_t& id = meta_info[index];
if (auto err = source(id))
detail::type_id_list_builder ids;
ids.reserve(ids_size + 1); // +1 for the prefixed size
for (size_t i = 0; i < ids_size; ++i) {
type_id_t id = 0;
if (auto err = source.apply(id))
return err;
if (id >= meta.size() || meta[id].type_name == nullptr) {
CAF_LOG_ERROR("unknown type ID" << id
<< "while deserializing a message of size"
<< num_elements);
ids.push_back(id);
}
auto gmos = detail::global_meta_objects();
size_t data_size = 0;
for (size_t i = 0; i < ids_size; ++i) {
if (i >= gmos.size())
return sec::unknown_type;
}
auto& mo = gmos[ids[i]];
if (mo.type_name == nullptr)
return sec::unknown_type;
data_size += mo.padded_size;
}
// Fetch the content of the message.
auto result = make_counted<detail::dynamic_message_data>();
for (size_t index = 1; index < (num_elements + 1); ++index) {
uint16_t id = meta_info[index];
std::unique_ptr<type_erased_value> element;
element.reset(meta[id].make());
if (auto err = element->load(source))
auto vptr = malloc(sizeof(detail::message_data) + data_size);
auto ptr = new (vptr) detail::message_data(ids.to_list());
auto pos = ptr->storage();
for (auto i = 1; i <= ids_size; ++i) {
auto& meta = gmos[ids[i]];
meta.default_construct(pos);
if (auto err = load(meta, source, pos)) {
auto rpos = pos;
for (auto j = i; j > 0; --j) {
auto& jmeta = gmos[ids[j]];
jmeta.destroy(rpos);
rpos -= jmeta.padded_size;
}
ptr->~message_data();
free(vptr);
return err;
result->append(std::move(element));
}
pos += meta.padded_size;
}
vals = detail::message_data::cow_ptr{std::move(result)};
return none;
data.reset(ptr, false);
return caf::none;
}
} // namespace
error message::load(deserializer& source) {
return load_vals(source, vals_);
return load_data(source, data_);
}
error_code<sec> message::load(binary_deserializer& source) {
return load_vals(source, vals_);
return load_data(source, data_);
}
namespace {
template <class Serializer>
typename Serializer::result_type
save_tuple(Serializer& sink, const type_erased_tuple& x) {
save_data(Serializer& sink, const message::data_ptr& data) {
// Short-circuit empty tuples.
if (x.empty()) {
if (data == nullptr) {
uint16_t zero = 0;
return sink(zero);
}
// Write type information.
auto* ids = x.types().data();
for (size_t index = 0; index < ids[0] + 1; ++index)
if (auto err = sink(ids[index]))
auto type_ids = data->types();
auto type_ids_size = static_cast<uint16_t>(type_ids.size());
if (auto err = sink(type_ids_size))
return err;
for (auto id : type_ids)
if (auto err = sink(id))
return err;
// Write the payload.
for (size_t index = 0; index < x.size(); ++index)
if (auto err = x.save(index, sink))
// Write elements.
auto gmos = detail::global_meta_objects();
auto storage = data->storage();
for (auto id : type_ids) {
auto& meta = gmos[id];
if (auto err = save(meta, sink, storage))
return err;
storage += meta.padded_size;
}
return {};
}
} // namespace
error message::save(serializer& sink, const type_erased_tuple& x) {
return save_tuple(sink, x);
}
error_code<sec>
message::save(binary_serializer& sink, const type_erased_tuple& x) {
return save_tuple(sink, x);
}
error message::save(serializer& sink) const {
return save_tuple(sink, *this);
return save_data(sink, data_);
}
error_code<sec> message::save(binary_serializer& sink) const {
return save_tuple(sink, *this);
}
// -- factories ----------------------------------------------------------------
message message::copy(const type_erased_tuple& xs) {
message_builder mb;
for (size_t i = 0; i < xs.size(); ++i)
mb.emplace(xs.copy(i));
return mb.move_to_message();
}
// -- modifiers ----------------------------------------------------------------
optional<message> message::apply(message_handler handler) {
return handler(*this);
return save_data(sink, data_);
}
void message::swap(message& other) noexcept {
vals_.swap(other.vals_);
}
// -- related non-members ------------------------------------------------------
void message::reset(raw_ptr new_ptr, bool add_ref) noexcept {
vals_.reset(new_ptr, add_ref);
error inspect(serializer& sink, const message& msg) {
return msg.save(sink);
}
error inspect(serializer& sink, message& msg) {
error_code<sec> inspect(binary_serializer& sink, const message& msg) {
return msg.save(sink);
}
......@@ -239,25 +150,13 @@ error inspect(deserializer& source, message& msg) {
return msg.load(source);
}
error_code<sec> inspect(binary_serializer& sink, message& msg) {
return msg.save(sink);
}
error_code<sec> inspect(binary_deserializer& source, message& msg) {
return msg.load(source);
}
std::string to_string(const message& msg) {
if (msg.empty())
return "<empty-message>";
std::string str = "(";
str += msg.cvals()->stringify(0);
for (size_t i = 1; i < msg.size(); ++i) {
str += ", ";
str += msg.cvals()->stringify(i);
}
str += ")";
return str;
std::string to_string(const message& ) {
// TODO: implement me
return "";
}
} // namespace caf
......@@ -18,65 +18,21 @@
#include "caf/message_builder.hpp"
#include <vector>
#include "caf/detail/dynamic_message_data.hpp"
#include "caf/make_copy_on_write.hpp"
#include "caf/message_handler.hpp"
namespace caf {
message_builder::message_builder() {
init();
}
message_builder::~message_builder() {
// nop
}
void message_builder::init() {
// this should really be done by delegating
// constructors, but we want to support
// some compilers without that feature...
data_ = make_copy_on_write<detail::dynamic_message_data>();
}
void message_builder::clear() {
if (data_->unique())
data_.unshared().clear();
else
init();
}
size_t message_builder::size() const {
return data_->size();
}
bool message_builder::empty() const {
return size() == 0;
}
message_builder& message_builder::emplace(type_erased_value_ptr x) {
data_.unshared().append(std::move(x));
return *this;
void message_builder::clear() noexcept {
types_.clear();
elements_.clear();
}
message message_builder::to_message() const {
return message{data_};
// TODO: implement me
return {};
}
message message_builder::move_to_message() {
return message{std::move(data_)};
}
optional<message> message_builder::apply(message_handler handler) {
// Avoid detaching of data_ by moving the data to a message object,
// calling message::apply and moving the data back.
auto msg = move_to_message();
auto res = msg.apply(std::move(handler));
data_.reset(static_cast<detail::dynamic_message_data*>(msg.vals().release()),
false);
return res;
// TODO: implement me
return {};
}
} // namespace caf
......@@ -25,6 +25,7 @@
#include "caf/message_handler.hpp"
#include "caf/sec.hpp"
#include "caf/system_messages.hpp"
#include "caf/typed_message_view.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
......@@ -220,28 +221,33 @@ bool monitorable_actor::handle_system_message(mailbox_element& x,
execution_unit* ctx,
bool trap_exit) {
auto& msg = x.content();
if (!trap_exit && msg.size() == 1 && msg.match_element<exit_msg>(0)) {
// exits for non-normal exit reasons
auto& em = msg.get_mutable_as<exit_msg>(0);
if (em.reason)
cleanup(std::move(em.reason), ctx);
return true;
if (!trap_exit) {
if (auto view = make_typed_message_view<exit_msg>(msg)) {
// exits for non-normal exit reasons
auto& em = get<0>(view);
if (em.reason)
cleanup(std::move(em.reason), ctx);
return true;
}
}
if (msg.size() > 1 && msg.match_element<sys_atom>(0)) {
if (!x.sender)
return true;
error err;
mailbox_element_ptr res;
msg.apply([&](sys_atom, get_atom, std::string& what) {
CAF_LOG_TRACE(CAF_ARG(what));
if (what != "info") {
err = sec::unsupported_sys_key;
return;
}
res = make_mailbox_element(ctrl(), x.mid.response_id(), {}, ok_atom_v,
std::move(what), strong_actor_ptr{ctrl()},
name());
});
message_handler f{
[&](sys_atom, get_atom, std::string& what) {
CAF_LOG_TRACE(CAF_ARG(what));
if (what != "info") {
err = sec::unsupported_sys_key;
return;
}
res = make_mailbox_element(ctrl(), x.mid.response_id(), {}, ok_atom_v,
std::move(what), strong_actor_ptr{ctrl()},
name());
},
};
f(msg);
if (!res && !err)
err = sec::unsupported_sys_message;
if (err && x.mid.is_request())
......
......@@ -580,8 +580,8 @@ scheduled_actor::categorize(mailbox_element& x) {
}
return message_category::internal;
}
if (content.match_elements<exit_msg>()) {
auto em = content.move_if_unshared<exit_msg>(0);
if (auto view = make_typed_message_view<exit_msg>(content)) {
auto& em = get<0>(view);
// make sure to get rid of attachables if they're no longer needed
unlink_from(em.source);
// exit_reason::kill is always fatal and also aborts streams.
......@@ -604,13 +604,13 @@ scheduled_actor::categorize(mailbox_element& x) {
}
return message_category::internal;
}
if (content.match_elements<down_msg>()) {
auto dm = content.move_if_unshared<down_msg>(0);
if (auto view = make_typed_message_view<down_msg>(content)) {
auto& dm = get<0>(view);
call_handler(down_handler_, this, dm);
return message_category::internal;
}
if (content.match_elements<error>()) {
auto err = content.move_if_unshared<error>(0);
if (auto view = make_typed_message_view<error>(content)) {
auto& err = get<0>(view);
call_handler(error_handler_, this, err);
return message_category::internal;
}
......
......@@ -21,6 +21,8 @@
#include <cstdint>
#include "caf/actor_system.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/deserializer.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp"
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/type_erased_tuple.hpp"
#include <memory>
#include "caf/config.hpp"
#include "caf/detail/dynamic_message_data.hpp"
#include "caf/error.hpp"
#include "caf/raise_error.hpp"
namespace {
caf::type_id_t empty_type_list[] = {0};
} // namespace
namespace caf {
type_erased_tuple::~type_erased_tuple() {
// nop
}
error type_erased_tuple::load(deserializer& source) {
for (size_t i = 0; i < size(); ++i)
if (auto err = load(i, source))
return err;
return none;
}
error_code<sec> type_erased_tuple::load(binary_deserializer& source) {
for (size_t i = 0; i < size(); ++i)
if (auto err = load(i, source))
return err;
return none;
}
bool type_erased_tuple::shared() const noexcept {
return false;
}
bool type_erased_tuple::empty() const {
return size() == 0;
}
std::string type_erased_tuple::stringify() const {
if (size() == 0)
return "()";
std::string result = "(";
result += stringify(0);
for (size_t i = 1; i < size(); ++i) {
result += ", ";
result += stringify(i);
}
result += ')';
return result;
}
error type_erased_tuple::save(serializer& sink) const {
for (size_t i = 0; i < size(); ++i) {
auto e = save(i, sink);
if (e)
return e;
}
return none;
}
error_code<sec> type_erased_tuple::save(binary_serializer& sink) const {
for (size_t i = 0; i < size(); ++i)
save(i, sink);
return none;
}
type_erased_tuple* type_erased_tuple::copy() const {
// Temporary hack until redesigning caf::message.
auto ptr = std::make_unique<detail::dynamic_message_data>();
for (size_t i = 0; i < size(); ++i)
ptr->append(copy(i));
return ptr.release();
}
empty_type_erased_tuple::~empty_type_erased_tuple() {
// nop
}
void* empty_type_erased_tuple::get_mutable(size_t) {
CAF_RAISE_ERROR("empty_type_erased_tuple::get_mutable");
}
error empty_type_erased_tuple::load(size_t, deserializer&) {
CAF_RAISE_ERROR("empty_type_erased_tuple::get_mutable");
}
error_code<sec> empty_type_erased_tuple::load(size_t, binary_deserializer&) {
CAF_RAISE_ERROR("empty_type_erased_tuple::load");
}
size_t empty_type_erased_tuple::size() const noexcept {
return 0;
}
type_id_list empty_type_erased_tuple::types() const noexcept {
return type_id_list{empty_type_list};
}
type_id_t empty_type_erased_tuple::type(size_t) const noexcept {
CAF_CRITICAL("empty_type_erased_tuple::type");
}
const void* empty_type_erased_tuple::get(size_t) const noexcept {
CAF_CRITICAL("empty_type_erased_tuple::get");
}
std::string empty_type_erased_tuple::stringify(size_t) const {
CAF_RAISE_ERROR("empty_type_erased_tuple::stringify");
}
type_erased_value_ptr empty_type_erased_tuple::copy(size_t) const {
CAF_RAISE_ERROR("empty_type_erased_tuple::copy");
}
error empty_type_erased_tuple::save(size_t, serializer&) const {
CAF_RAISE_ERROR("empty_type_erased_tuple::save");
}
error_code<sec>
empty_type_erased_tuple::save(size_t, binary_serializer&) const {
CAF_RAISE_ERROR("empty_type_erased_tuple::save");
}
error inspect(serializer& f, const type_erased_tuple& x) {
return x.save(f);
}
error inspect(deserializer& f, type_erased_tuple& x) {
return x.load(f);
}
} // namespace caf
......@@ -30,7 +30,6 @@
#include "caf/message_handler.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/make_type_erased_tuple_view.hpp"
using namespace caf;
......
......@@ -22,7 +22,6 @@
#include "caf/test/dsl.hpp"
#include "caf/make_message.hpp"
#include "caf/message.hpp"
using namespace caf;
......@@ -30,10 +29,10 @@ using namespace caf;
CAF_TEST(const message views never detach their content) {
auto msg1 = make_message(1, 2, 3, "four");
auto msg2 = msg1;
CAF_REQUIRE(msg1.cvals().get() == msg2.cvals().get());
CAF_REQUIRE(msg1.cptr() == msg2.cptr());
CAF_REQUIRE(msg1.match_elements<int, int, int, std::string>());
const_typed_message_view<int, int, int, std::string> view{msg1};
CAF_REQUIRE(msg1.cvals().get() == msg2.cvals().get());
CAF_REQUIRE(msg1.cptr() == msg2.cptr());
}
CAF_TEST(const message views allow access via get) {
......
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Copyright 2011-2020 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 *
......@@ -16,21 +16,26 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#define CAF_SUITE detail.message_builder_element
#include <functional>
#include "caf/detail/message_builder_element.hpp"
#include "caf/detail/type_erased_value_impl.hpp"
#include "caf/type_erased_value.hpp"
#include "caf/test/dsl.hpp"
namespace caf {
using namespace caf;
/// @relates type_erased_value
/// Creates a type-erased view for `x`.
template <class T>
detail::type_erased_value_impl<std::reference_wrapper<T>>
make_type_erased_view(T& x) {
return {std::ref(x)};
namespace {
struct fixture {
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(message_builder_element_tests, fixture)
CAF_TEST(todo) {
// implement me
}
} // namespace caf
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Copyright 2011-2020 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 *
......@@ -16,23 +16,26 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE detail.type_id_list_builder
#define CAF_SUITE type_erased_tuple
#include "caf/test/unit_test.hpp"
#include "caf/detail/type_id_list_builder.hpp"
#include "caf/all.hpp"
#include "caf/make_type_erased_tuple_view.hpp"
#include "caf/test/dsl.hpp"
using namespace std;
using namespace caf;
CAF_TEST(get_as_tuple) {
int x = 1;
int y = 2;
int z = 3;
auto tup = make_type_erased_tuple_view(x, y, z);
auto xs = tup.get_as_tuple<int, int, int>();
CAF_CHECK_EQUAL(xs, std::make_tuple(1, 2, 3));
namespace {
struct fixture {
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(type_id_list_builder_tests, fixture)
CAF_TEST(todo) {
// implement me
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -36,8 +36,6 @@
#include "caf/detail/meta_object.hpp"
#include "caf/detail/stringification_inspector.hpp"
#include "caf/init_global_meta_objects.hpp"
#include "caf/make_type_erased_value.hpp"
#include "caf/type_erased_value.hpp"
namespace {
......
......@@ -36,16 +36,11 @@ using namespace caf;
namespace {
template <class... Ts>
optional<tuple<Ts...>> fetch(const type_erased_tuple& x) {
if (!x.match_elements<Ts...>())
return none;
return x.get_as_tuple<Ts...>();
}
template <class... Ts>
optional<tuple<Ts...>> fetch(const message& x) {
return fetch<Ts...>(x.content());
if (auto view = make_const_typed_message_view<Ts...>(x))
return to_tuple(view);
return none;
}
template <class... Ts>
......
......@@ -71,18 +71,6 @@ using namespace caf;
using namespace std::literals::string_literals;
CAF_TEST(apply) {
auto f1 = [] {
CAF_ERROR("f1 invoked!");
};
auto f2 = [](int i) {
CAF_CHECK_EQUAL(i, 42);
};
auto m = make_message(42);
m.apply(f1);
m.apply(f2);
}
namespace {
struct s1 {
......
......@@ -133,10 +133,10 @@ CAF_TEST(nocopy_in_scoped_actor) {
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.data().get_reference_count(), 2u);
}
);
CAF_CHECK_EQUAL(msg.cvals()->get_reference_count(), 1u);
CAF_CHECK_EQUAL(msg.data().get_reference_count(), 1u);
}
CAF_TEST(message_lifetime_in_scoped_actor) {
......@@ -147,16 +147,16 @@ CAF_TEST(message_lifetime_in_scoped_actor) {
CAF_CHECK_EQUAL(a, 1);
CAF_CHECK_EQUAL(b, 2);
CAF_CHECK_EQUAL(c, 3);
CAF_CHECK_EQUAL(msg.cvals()->get_reference_count(), 2u);
CAF_CHECK_EQUAL(msg.cdata().get_reference_count(), 2u);
}
);
CAF_CHECK_EQUAL(msg.cvals()->get_reference_count(), 1u);
CAF_CHECK_EQUAL(msg.cdata().get_reference_count(), 1u);
msg = make_message(42);
self->send(self, msg);
CAF_CHECK_EQUAL(msg.cvals()->get_reference_count(), 2u);
CAF_CHECK_EQUAL(msg.cdata().get_reference_count(), 2u);
self->receive(
[&](int& value) {
CAF_CHECK_NOT_EQUAL(&value, msg.at(0));
CAF_CHECK_NOT_EQUAL(&value, msg.cdata().at(0));
value = 10;
}
);
......
......@@ -54,8 +54,6 @@
#include "caf/detail/safe_equal.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/make_type_erased_tuple_view.hpp"
#include "caf/make_type_erased_view.hpp"
#include "caf/message.hpp"
#include "caf/message_handler.hpp"
#include "caf/proxy_registry.hpp"
......@@ -356,37 +354,6 @@ CAF_TEST(multiple_messages) {
CAF_CHECK(is_message(m2).equal(i32, i64, ts, te, str, rs));
}
CAF_TEST(type_erased_value) {
using caf::detail::type_erased_value_impl;
auto buf = serialize(str);
type_erased_value_ptr ptr{new type_erased_value_impl<std::string>};
binary_deserializer source{sys, buf};
ptr->load(source);
CAF_CHECK_EQUAL(str, *reinterpret_cast<const std::string*>(ptr->get()));
}
CAF_TEST(type_erased_view) {
auto str_view = make_type_erased_view(str);
auto buf = serialize(str_view);
std::string res;
deserialize(buf, res);
CAF_CHECK_EQUAL(str, res);
}
CAF_TEST(type_erased_tuple) {
auto tview = make_type_erased_tuple_view(str, i32);
CAF_CHECK_EQUAL(to_string(tview), deep_to_string(std::make_tuple(str, i32)));
auto buf = serialize(tview);
CAF_REQUIRE(!buf.empty());
std::string tmp1;
int32_t tmp2;
deserialize(buf, tmp1, tmp2);
CAF_CHECK_EQUAL(tmp1, str);
CAF_CHECK_EQUAL(tmp2, i32);
deserialize(buf, tview);
CAF_CHECK_EQUAL(to_string(tview), deep_to_string(std::make_tuple(str, i32)));
}
CAF_TEST(long_sequences) {
byte_buffer data;
binary_serializer sink{nullptr, data};
......
......@@ -22,7 +22,6 @@
#include "caf/test/dsl.hpp"
#include "caf/make_message.hpp"
#include "caf/message.hpp"
using namespace caf;
......@@ -30,10 +29,10 @@ using namespace caf;
CAF_TEST(message views detach their content) {
auto msg1 = make_message(1, 2, 3, "four");
auto msg2 = msg1;
CAF_REQUIRE(msg1.cvals().get() == msg2.cvals().get());
CAF_REQUIRE(msg1.cptr() == msg2.cptr());
CAF_REQUIRE(msg1.match_elements<int, int, int, std::string>());
typed_message_view<int, int, int, std::string> view{msg1};
CAF_REQUIRE(msg1.cvals().get() != msg2.cvals().get());
CAF_REQUIRE(msg1.cptr() != msg2.cptr());
}
CAF_TEST(message views allow access via get) {
......
......@@ -24,6 +24,7 @@
#include "caf/actor_proxy.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/config.hpp"
#include "caf/const_typed_message_view.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/execution_unit.hpp"
......@@ -103,16 +104,18 @@ public:
}
// Intercept link messages. Forwarding actor proxies signalize linking
// by sending link_atom/unlink_atom message with src == dest.
if (msg.match_elements<link_atom, strong_actor_ptr>()) {
const auto& ptr = msg.get_as<strong_actor_ptr>(1);
if (auto view
= make_const_typed_message_view<link_atom, strong_actor_ptr>(msg)) {
const auto& ptr = get<1>(view);
if (ptr != nullptr)
static_cast<actor_proxy*>(ptr->get())->add_link(dst->get());
else
CAF_LOG_WARNING("received link message with invalid target");
return;
}
if (msg.match_elements<unlink_atom, strong_actor_ptr>()) {
const auto& ptr = msg.get_as<strong_actor_ptr>(1);
if (auto view
= make_const_typed_message_view<unlink_atom, strong_actor_ptr>(msg)) {
const auto& ptr = get<1>(view);
if (ptr != nullptr)
static_cast<actor_proxy*>(ptr->get())->remove_link(dst->get());
else
......
......@@ -55,27 +55,30 @@ connection_helper(stateful_actor<connection_helper_state>* self, actor b) {
CAF_LOG_DEBUG("received requested config:" << CAF_ARG(msg));
// whatever happens, we are done afterwards
self->quit();
msg.apply({[&](uint16_t port, network::address_listing& addresses) {
if (item == "basp.default-connectivity-tcp") {
auto& mx = self->system().middleman().backend();
for (auto& kvp : addresses) {
for (auto& addr : kvp.second) {
auto hdl = mx.new_tcp_scribe(addr, port);
if (hdl) {
// gotcha! send scribe to our BASP broker
// to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
self->send(b, connect_atom_v, *hdl, port);
return;
message_handler f{
[&](uint16_t port, network::address_listing& addresses) {
if (item == "basp.default-connectivity-tcp") {
auto& mx = self->system().middleman().backend();
for (auto& kvp : addresses) {
for (auto& addr : kvp.second) {
auto hdl = mx.new_tcp_scribe(addr, port);
if (hdl) {
// gotcha! send scribe to our BASP broker
// to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
self->send(b, connect_atom_v, *hdl, port);
return;
}
}
}
CAF_LOG_INFO("could not connect to node directly");
} else {
CAF_LOG_INFO("aborted direct connection attempt, unknown item: "
<< CAF_ARG(item));
}
CAF_LOG_INFO("could not connect to node directly");
} else {
CAF_LOG_INFO("aborted direct connection attempt, unknown item: "
<< CAF_ARG(item));
}
}});
},
};
f(msg);
},
after(autoconnect_timeout) >>
[=] {
......
......@@ -233,9 +233,11 @@ private:
template <class... Ts>
caf::optional<std::tuple<Ts...>> default_extract(caf_handle x) {
auto ptr = x->peek_at_next_mailbox_element();
if (ptr == nullptr || !ptr->content().template match_elements<Ts...>())
if (ptr == nullptr)
return caf::none;
return ptr->content().template get_as_tuple<Ts...>();
if (auto view = caf::make_const_typed_message_view<Ts...>(ptr->content()))
return to_tuple(view);
return caf::none;
}
/// @private
......
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