Commit 7b598dff authored by Dominik Charousset's avatar Dominik Charousset

Integrate review feedback

parent e6fe71e9
......@@ -57,7 +57,7 @@ private:
};
template <size_t Index, class... Ts>
const auto& get(const_typed_message_view<Ts...>& xs) {
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()
......
......@@ -64,8 +64,8 @@ public:
rc_.fetch_add(1, std::memory_order_relaxed);
}
/// Decreases reference count by one and calls `request_deletion`
/// when it drops to zero.
/// Decreases the reference count by one and destroys the object when its
/// reference count drops to zero.
void deref() noexcept {
if (unique() || rc_.fetch_sub(1, std::memory_order_acq_rel) == 1) {
this->~message_data();
......
......@@ -48,11 +48,11 @@ struct type_id_sequence_helper<type_id_pair<Begin, End>, Is...> {
Is..., Begin>::type;
};
template <uint16_t Begin, uint16_t End>
using make_type_id_sequence =
typename type_id_sequence_helper<type_id_pair<Begin, End>>::type;
template <class Range>
using make_type_id_sequence = typename type_id_sequence_helper<
type_id_pair<Range::begin, Range::end>>::type;
CAF_CORE_EXPORT void init_global_builtin_meta_objects();
CAF_CORE_EXPORT void init_core_module_meta_objects();
} // namespace caf::detail
......@@ -74,9 +74,9 @@ void init_global_meta_objects_impl(std::integer_sequence<uint16_t, Is...>) {
/// causes undefined behavior.
template <class ProjectIds = void>
void init_global_meta_objects() {
detail::init_global_builtin_meta_objects();
detail::init_core_module_meta_objects();
if constexpr (!std::is_same<ProjectIds, void>::value) {
detail::make_type_id_sequence<ProjectIds::begin, ProjectIds::end> seq;
detail::make_type_id_sequence<ProjectIds> seq;
init_global_meta_objects_impl<ProjectIds>(seq);
}
}
......
......@@ -203,112 +203,112 @@ constexpr type_id_t first_custom_type_id = 200;
}; \
}
CAF_BEGIN_TYPE_ID_BLOCK(builtin, 0)
CAF_BEGIN_TYPE_ID_BLOCK(core_module, 0)
// -- C types
CAF_ADD_TYPE_ID(builtin, (bool) )
CAF_ADD_TYPE_ID(builtin, (double) )
CAF_ADD_TYPE_ID(builtin, (float) )
CAF_ADD_TYPE_ID(builtin, (int16_t))
CAF_ADD_TYPE_ID(builtin, (int32_t))
CAF_ADD_TYPE_ID(builtin, (int64_t))
CAF_ADD_TYPE_ID(builtin, (int8_t))
CAF_ADD_TYPE_ID(builtin, (long double) )
CAF_ADD_TYPE_ID(builtin, (uint16_t))
CAF_ADD_TYPE_ID(builtin, (uint32_t))
CAF_ADD_TYPE_ID(builtin, (uint64_t))
CAF_ADD_TYPE_ID(builtin, (uint8_t))
CAF_ADD_TYPE_ID(core_module, (bool) )
CAF_ADD_TYPE_ID(core_module, (double) )
CAF_ADD_TYPE_ID(core_module, (float) )
CAF_ADD_TYPE_ID(core_module, (int16_t))
CAF_ADD_TYPE_ID(core_module, (int32_t))
CAF_ADD_TYPE_ID(core_module, (int64_t))
CAF_ADD_TYPE_ID(core_module, (int8_t))
CAF_ADD_TYPE_ID(core_module, (long double) )
CAF_ADD_TYPE_ID(core_module, (uint16_t))
CAF_ADD_TYPE_ID(core_module, (uint32_t))
CAF_ADD_TYPE_ID(core_module, (uint64_t))
CAF_ADD_TYPE_ID(core_module, (uint8_t))
// -- STL types
CAF_ADD_TYPE_ID(builtin, (std::string))
CAF_ADD_TYPE_ID(builtin, (std::u16string))
CAF_ADD_TYPE_ID(builtin, (std::u32string))
CAF_ADD_TYPE_ID(builtin, (std::set<std::string>) )
CAF_ADD_TYPE_ID(core_module, (std::string))
CAF_ADD_TYPE_ID(core_module, (std::u16string))
CAF_ADD_TYPE_ID(core_module, (std::u32string))
CAF_ADD_TYPE_ID(core_module, (std::set<std::string>) )
// -- CAF types
CAF_ADD_TYPE_ID(builtin, (caf::actor))
CAF_ADD_TYPE_ID(builtin, (caf::actor_addr))
CAF_ADD_TYPE_ID(builtin, (caf::byte_buffer))
CAF_ADD_TYPE_ID(builtin, (caf::config_value))
CAF_ADD_TYPE_ID(builtin, (caf::dictionary<caf::config_value>) )
CAF_ADD_TYPE_ID(builtin, (caf::down_msg))
CAF_ADD_TYPE_ID(builtin, (caf::downstream_msg))
CAF_ADD_TYPE_ID(builtin, (caf::error))
CAF_ADD_TYPE_ID(builtin, (caf::exit_msg))
CAF_ADD_TYPE_ID(builtin, (caf::group))
CAF_ADD_TYPE_ID(builtin, (caf::group_down_msg))
CAF_ADD_TYPE_ID(builtin, (caf::message))
CAF_ADD_TYPE_ID(builtin, (caf::message_id))
CAF_ADD_TYPE_ID(builtin, (caf::node_id))
CAF_ADD_TYPE_ID(builtin, (caf::open_stream_msg))
CAF_ADD_TYPE_ID(builtin, (caf::strong_actor_ptr))
CAF_ADD_TYPE_ID(builtin, (caf::timeout_msg))
CAF_ADD_TYPE_ID(builtin, (caf::timespan))
CAF_ADD_TYPE_ID(builtin, (caf::timestamp))
CAF_ADD_TYPE_ID(builtin, (caf::unit_t))
CAF_ADD_TYPE_ID(builtin, (caf::upstream_msg))
CAF_ADD_TYPE_ID(builtin, (caf::uri))
CAF_ADD_TYPE_ID(builtin, (caf::weak_actor_ptr))
CAF_ADD_TYPE_ID(builtin, (std::vector<caf::actor>) )
CAF_ADD_TYPE_ID(builtin, (std::vector<caf::actor_addr>) )
CAF_ADD_TYPE_ID(builtin, (std::vector<caf::config_value>) )
CAF_ADD_TYPE_ID(builtin, (std::vector<caf::strong_actor_ptr>) )
CAF_ADD_TYPE_ID(builtin, (std::vector<caf::weak_actor_ptr>) )
CAF_ADD_TYPE_ID(builtin, (std::vector<std::pair<std::string, message>>) );
CAF_ADD_TYPE_ID(core_module, (caf::actor))
CAF_ADD_TYPE_ID(core_module, (caf::actor_addr))
CAF_ADD_TYPE_ID(core_module, (caf::byte_buffer))
CAF_ADD_TYPE_ID(core_module, (caf::config_value))
CAF_ADD_TYPE_ID(core_module, (caf::dictionary<caf::config_value>) )
CAF_ADD_TYPE_ID(core_module, (caf::down_msg))
CAF_ADD_TYPE_ID(core_module, (caf::downstream_msg))
CAF_ADD_TYPE_ID(core_module, (caf::error))
CAF_ADD_TYPE_ID(core_module, (caf::exit_msg))
CAF_ADD_TYPE_ID(core_module, (caf::group))
CAF_ADD_TYPE_ID(core_module, (caf::group_down_msg))
CAF_ADD_TYPE_ID(core_module, (caf::message))
CAF_ADD_TYPE_ID(core_module, (caf::message_id))
CAF_ADD_TYPE_ID(core_module, (caf::node_id))
CAF_ADD_TYPE_ID(core_module, (caf::open_stream_msg))
CAF_ADD_TYPE_ID(core_module, (caf::strong_actor_ptr))
CAF_ADD_TYPE_ID(core_module, (caf::timeout_msg))
CAF_ADD_TYPE_ID(core_module, (caf::timespan))
CAF_ADD_TYPE_ID(core_module, (caf::timestamp))
CAF_ADD_TYPE_ID(core_module, (caf::unit_t))
CAF_ADD_TYPE_ID(core_module, (caf::upstream_msg))
CAF_ADD_TYPE_ID(core_module, (caf::uri))
CAF_ADD_TYPE_ID(core_module, (caf::weak_actor_ptr))
CAF_ADD_TYPE_ID(core_module, (std::vector<caf::actor>) )
CAF_ADD_TYPE_ID(core_module, (std::vector<caf::actor_addr>) )
CAF_ADD_TYPE_ID(core_module, (std::vector<caf::config_value>) )
CAF_ADD_TYPE_ID(core_module, (std::vector<caf::strong_actor_ptr>) )
CAF_ADD_TYPE_ID(core_module, (std::vector<caf::weak_actor_ptr>) )
CAF_ADD_TYPE_ID(core_module, (std::vector<std::pair<std::string, message>>) );
// -- predefined atoms
CAF_ADD_ATOM(builtin, caf, add_atom)
CAF_ADD_ATOM(builtin, caf, close_atom)
CAF_ADD_ATOM(builtin, caf, connect_atom)
CAF_ADD_ATOM(builtin, caf, contact_atom)
CAF_ADD_ATOM(builtin, caf, delete_atom)
CAF_ADD_ATOM(builtin, caf, demonitor_atom)
CAF_ADD_ATOM(builtin, caf, div_atom)
CAF_ADD_ATOM(builtin, caf, flush_atom)
CAF_ADD_ATOM(builtin, caf, forward_atom)
CAF_ADD_ATOM(builtin, caf, get_atom)
CAF_ADD_ATOM(builtin, caf, idle_atom)
CAF_ADD_ATOM(builtin, caf, join_atom)
CAF_ADD_ATOM(builtin, caf, leave_atom)
CAF_ADD_ATOM(builtin, caf, link_atom)
CAF_ADD_ATOM(builtin, caf, migrate_atom)
CAF_ADD_ATOM(builtin, caf, monitor_atom)
CAF_ADD_ATOM(builtin, caf, mul_atom)
CAF_ADD_ATOM(builtin, caf, ok_atom)
CAF_ADD_ATOM(builtin, caf, open_atom)
CAF_ADD_ATOM(builtin, caf, pending_atom)
CAF_ADD_ATOM(builtin, caf, ping_atom)
CAF_ADD_ATOM(builtin, caf, pong_atom)
CAF_ADD_ATOM(builtin, caf, publish_atom)
CAF_ADD_ATOM(builtin, caf, publish_udp_atom)
CAF_ADD_ATOM(builtin, caf, put_atom)
CAF_ADD_ATOM(builtin, caf, receive_atom)
CAF_ADD_ATOM(builtin, caf, redirect_atom)
CAF_ADD_ATOM(builtin, caf, reset_atom)
CAF_ADD_ATOM(builtin, caf, resolve_atom)
CAF_ADD_ATOM(builtin, caf, spawn_atom)
CAF_ADD_ATOM(builtin, caf, stream_atom)
CAF_ADD_ATOM(builtin, caf, sub_atom)
CAF_ADD_ATOM(builtin, caf, subscribe_atom)
CAF_ADD_ATOM(builtin, caf, sys_atom)
CAF_ADD_ATOM(builtin, caf, tick_atom)
CAF_ADD_ATOM(builtin, caf, timeout_atom)
CAF_ADD_ATOM(builtin, caf, unlink_atom)
CAF_ADD_ATOM(builtin, caf, unpublish_atom)
CAF_ADD_ATOM(builtin, caf, unpublish_udp_atom)
CAF_ADD_ATOM(builtin, caf, unsubscribe_atom)
CAF_ADD_ATOM(builtin, caf, update_atom)
CAF_ADD_ATOM(builtin, caf, wait_for_atom)
CAF_END_TYPE_ID_BLOCK(builtin)
CAF_ADD_ATOM(core_module, caf, add_atom)
CAF_ADD_ATOM(core_module, caf, close_atom)
CAF_ADD_ATOM(core_module, caf, connect_atom)
CAF_ADD_ATOM(core_module, caf, contact_atom)
CAF_ADD_ATOM(core_module, caf, delete_atom)
CAF_ADD_ATOM(core_module, caf, demonitor_atom)
CAF_ADD_ATOM(core_module, caf, div_atom)
CAF_ADD_ATOM(core_module, caf, flush_atom)
CAF_ADD_ATOM(core_module, caf, forward_atom)
CAF_ADD_ATOM(core_module, caf, get_atom)
CAF_ADD_ATOM(core_module, caf, idle_atom)
CAF_ADD_ATOM(core_module, caf, join_atom)
CAF_ADD_ATOM(core_module, caf, leave_atom)
CAF_ADD_ATOM(core_module, caf, link_atom)
CAF_ADD_ATOM(core_module, caf, migrate_atom)
CAF_ADD_ATOM(core_module, caf, monitor_atom)
CAF_ADD_ATOM(core_module, caf, mul_atom)
CAF_ADD_ATOM(core_module, caf, ok_atom)
CAF_ADD_ATOM(core_module, caf, open_atom)
CAF_ADD_ATOM(core_module, caf, pending_atom)
CAF_ADD_ATOM(core_module, caf, ping_atom)
CAF_ADD_ATOM(core_module, caf, pong_atom)
CAF_ADD_ATOM(core_module, caf, publish_atom)
CAF_ADD_ATOM(core_module, caf, publish_udp_atom)
CAF_ADD_ATOM(core_module, caf, put_atom)
CAF_ADD_ATOM(core_module, caf, receive_atom)
CAF_ADD_ATOM(core_module, caf, redirect_atom)
CAF_ADD_ATOM(core_module, caf, reset_atom)
CAF_ADD_ATOM(core_module, caf, resolve_atom)
CAF_ADD_ATOM(core_module, caf, spawn_atom)
CAF_ADD_ATOM(core_module, caf, stream_atom)
CAF_ADD_ATOM(core_module, caf, sub_atom)
CAF_ADD_ATOM(core_module, caf, subscribe_atom)
CAF_ADD_ATOM(core_module, caf, sys_atom)
CAF_ADD_ATOM(core_module, caf, tick_atom)
CAF_ADD_ATOM(core_module, caf, timeout_atom)
CAF_ADD_ATOM(core_module, caf, unlink_atom)
CAF_ADD_ATOM(core_module, caf, unpublish_atom)
CAF_ADD_ATOM(core_module, caf, unpublish_udp_atom)
CAF_ADD_ATOM(core_module, caf, unsubscribe_atom)
CAF_ADD_ATOM(core_module, caf, update_atom)
CAF_ADD_ATOM(core_module, caf, wait_for_atom)
CAF_END_TYPE_ID_BLOCK(core_module)
namespace caf::detail {
static constexpr type_id_t io_module_begin = builtin_type_ids::end;
static constexpr type_id_t io_module_begin = core_module_type_ids::end;
static constexpr type_id_t io_module_end = io_module_begin + 19;
......
......@@ -53,7 +53,7 @@ private:
};
template <size_t Index, class... Ts>
auto& get(typed_message_view<Ts...>& x) {
auto& get(typed_message_view<Ts...> x) {
static_assert(Index < sizeof...(Ts));
using type = caf::detail::tl_at_t<caf::detail::type_list<Ts...>, Index>;
return *reinterpret_cast<type*>(x->storage()
......
......@@ -233,7 +233,8 @@ actor_system::actor_system(actor_system_config& cfg)
}
// Make sure meta objects are loaded.
auto gmos = detail::global_meta_objects();
if (gmos.size() < builtin_type_ids::end || gmos[0].type_name == nullptr) {
if (gmos.size() < core_module_type_ids::end
|| gmos[core_module_type_ids::begin].type_name == nullptr) {
CAF_CRITICAL("actor_system created without calling "
"caf::init_global_meta_objects<>() before");
}
......
......@@ -105,18 +105,8 @@ void type_id_list_builder::reserve(size_t new_capacity) {
}
void type_id_list_builder::push_back(type_id_t id) {
if ((size_ + 1) >= reserved_) {
if ((size_ + 1) >= reserved_)
reserve(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;
}
......
......@@ -38,9 +38,9 @@
namespace caf::detail {
void init_global_builtin_meta_objects() {
make_type_id_sequence<builtin_type_ids::begin, builtin_type_ids::end> seq;
init_global_meta_objects_impl<builtin_type_ids>(seq);
void init_core_module_meta_objects() {
make_type_id_sequence<core_module_type_ids> seq;
init_global_meta_objects_impl<core_module_type_ids>(seq);
}
} // namespace caf::detail
......@@ -663,7 +663,7 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
auto bhvr = std::move(mrh->second);
multiplexed_responses_.erase(mrh);
if (!invoke(this, bhvr, x)) {
CAF_LOG_DEBUG("got unexpected_response, invoke as again as error");
CAF_LOG_DEBUG("got unexpected_response, invoke unexpected_response");
auto msg = make_message(
make_error(sec::unexpected_response, std::move(x.payload)));
bhvr(msg);
......
......@@ -253,25 +253,7 @@ using fixture = test_coordinator_fixture<>;
CAF_TEST_FIXTURE_SCOPE(fused_downstream_manager_tests, fixture)
// Currently fails for some bizarre reason related to type ID list.
CAF_TEST_DISABLED(depth_3_pipeline_with_fork) {
CAF_MESSAGE("sanity checks");
{
using detail::strip_and_convert_t;
using int_stream = stream<int32_t>;
using str_stream = stream<std::string>;
auto m1 = make_message(int_stream{});
auto m2 = make_message(str_stream{});
CAF_CHECK_NOT_EQUAL(type_id_v<int_stream>, type_id_v<str_stream>);
CAF_CHECK_NOT_EQUAL(type_id_v<strip_and_convert_t<int_stream>>,
type_id_v<strip_and_convert_t<str_stream>>);
CAF_CHECK_NOT_EQUAL(m1.types(), m2.types());
// CAF_CHECK_NOT_EQUAL(make_type_id_list<int_stream>(),
// make_type_id_list<str_stream>()); auto m1_ = make_message(int_stream{});
// auto m2_ = make_message(str_stream{});
// CAF_CHECK_NOT_EQUAL(m1_.types(), m2_.types());
}
auto src1 = sys.spawn(int_file_reader, 50u);
auto src2 = sys.spawn(string_file_reader, 50u);
auto stg = sys.spawn(stream_multiplexer);
......
......@@ -377,10 +377,6 @@ void middleman::init(actor_system_config& cfg) {
};
auto gfactory = [=]() -> group_module* { return new remote_groups(*this); };
cfg.group_module_factories.emplace_back(gfactory);
// Add I/O-related types.
using ids = io_module_type_ids;
detail::make_type_id_sequence<ids::begin, ids::end> seq;
caf::init_global_meta_objects_impl<ids>(seq);
// Compute and set ID for this network node.
auto this_node = node_id::default_data::local(cfg);
system().node_.swap(this_node);
......
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