From 9305ee204391172a43cf913816c74f4d0bf71521 Mon Sep 17 00:00:00 2001 From: tisomi Date: Wed, 13 May 2026 13:36:05 +0200 Subject: [PATCH] Populate type hashes in graph cache names_and_types (#356) Populates the new type_hashes field on rmw_names_and_types_t from the entries in GraphCache. The internal map is widened from std::set to std::map so each (topic, type) pair carries its hash through to the rmw caller, covering both the global get_names_and_types and per-node paths. When the same (topic, type) pair carries different hashes across endpoints, the conflict is detected and a WARN-level log is emitted; the stored hash is replaced with a zero hash so the caller sees an unambiguous "no consensus" value, matching the rmw_zenoh behavior. Adds graph-cache tests covering the populated hashes for both paths. Related: ros2/rmw#356 Signed-off-by: tisomi --- rmw_dds_common/src/graph_cache.cpp | 144 ++++++++++++++--- rmw_dds_common/test/test_graph_cache.cpp | 193 +++++++++++++++++++++++ 2 files changed, 315 insertions(+), 22 deletions(-) diff --git a/rmw_dds_common/src/graph_cache.cpp b/rmw_dds_common/src/graph_cache.cpp index 3fe5255..e90e73b 100644 --- a/rmw_dds_common/src/graph_cache.cpp +++ b/rmw_dds_common/src/graph_cache.cpp @@ -835,7 +835,57 @@ GraphCache::get_servers_info_by_service( endpoints_info); } -using NamesAndTypes = std::map>; +using NamesAndTypesAndHashes = std::map>; + +static bool +_type_hashes_equal(const rosidl_type_hash_t & lhs, const rosidl_type_hash_t & rhs) +{ + if (lhs.version != rhs.version) { + return false; + } + return 0 == std::memcmp(lhs.value, rhs.value, ROSIDL_TYPE_HASH_SIZE); +} + +static void +_insert_type_hash_or_warn( + std::map & types_map, + const std::string & topic_name, + const std::string & type_name, + const rosidl_type_hash_t & hash) +{ + auto result = types_map.insert({type_name, hash}); + if (!result.second && !_type_hashes_equal(result.first->second, hash)) { + RCUTILS_LOG_WARN_NAMED( + log_tag, + "Conflicting type hashes for topic '%s', type '%s'; storing zero hash", + topic_name.c_str(), + type_name.c_str()); + result.first->second = rosidl_get_zero_initialized_type_hash(); + } +} + +// Pick the hash that callers of the graph query APIs should see for a given +// entity. Services are implemented on DDS as a pair of underlying request/ +// reply topics, so a data_reader/data_writer that belongs to a service has +// two hashes available: +// * topic_type_hash: hash of the per-direction message type (Request or +// Reply). Differs between server and client side. +// * service_type_hash: hash of the service descriptor itself. Identical on +// both sides and is the value the user expects. +// When the entity is registered as a service (present in `services`), return +// the service-level hash; otherwise fall back to the topic-level hash. +static inline rosidl_type_hash_t +_select_hash_for_entity( + const rmw_gid_t & gid, + const rosidl_type_hash_t & topic_type_hash, + const GraphCache::EntityGidToServiceInfo & services) +{ + auto it = services.find(gid); + if (it != services.end()) { + return it->second.service_type_hash; + } + return topic_type_hash; +} static void @@ -843,22 +893,34 @@ __get_names_and_types( const GraphCache::EntityGidToInfo & entities, DemangleFunctionT demangle_topic, DemangleFunctionT demangle_type, - NamesAndTypes & topics) + // Lookup map for entities that are part of a service. Passed in so this + // helper can return the service-level hash for service entries while + // continuing to return the topic-level hash for plain topics. + const GraphCache::EntityGidToServiceInfo & services, + NamesAndTypesAndHashes & topics) { assert(nullptr != demangle_topic); assert(nullptr != demangle_type); for (const auto & item : entities) { std::string demangled_topic_name = demangle_topic(item.second.topic_name); - if ("" != demangled_topic_name) { - topics[demangled_topic_name].insert(demangle_type(item.second.topic_type)); + if ("" == demangled_topic_name) { + continue; } + const std::string type_name = demangle_type(item.second.topic_type); + const rosidl_type_hash_t hash = + _select_hash_for_entity(item.first, item.second.topic_type_hash, services); + _insert_type_hash_or_warn( + topics[demangled_topic_name], + demangled_topic_name, + type_name, + hash); } } static rmw_ret_t __populate_rmw_names_and_types( - NamesAndTypes topics, + const NamesAndTypesAndHashes & topics, rcutils_allocator_t * allocator, rmw_names_and_types_t * topic_names_and_types) { @@ -881,26 +943,43 @@ __populate_rmw_names_and_types( } topic_names_and_types->names.data[index] = topic_name; - { - rcutils_ret_t rcutils_ret = rcutils_string_array_init( - &topic_names_and_types->types[index], - item.second.size(), - allocator); - if (RCUTILS_RET_OK != rcutils_ret) { - RMW_SET_ERROR_MSG(rcutils_get_error_string().str); - rmw_ret = rmw_convert_rcutils_ret_to_rmw_ret(rcutils_ret); + rcutils_ret_t rcutils_ret = rcutils_string_array_init( + &topic_names_and_types->types[index], + item.second.size(), + allocator); + if (RCUTILS_RET_OK != rcutils_ret) { + RMW_SET_ERROR_MSG(rcutils_get_error_string().str); + rmw_ret = rmw_convert_rcutils_ret_to_rmw_ret(rcutils_ret); + goto cleanup; + } + + // allocate array of hashes for this topic + if (item.second.size() > 0) { + rosidl_type_hash_t * hashes = static_cast( + allocator->allocate(sizeof(rosidl_type_hash_t) * item.second.size(), allocator->state)); + if (!hashes) { + RMW_SET_ERROR_MSG("failed to allocate memory for type hashes"); + rmw_ret = RMW_RET_BAD_ALLOC; goto cleanup; } + topic_names_and_types->type_hashes[index] = hashes; + } else { + topic_names_and_types->type_hashes[index] = nullptr; } size_t type_index = 0; - for (const auto & type : item.second) { - char * type_name = rcutils_strdup(type.c_str(), *allocator); + for (const auto & type_pair : item.second) { + // type_pair: std::pair + char * type_name = rcutils_strdup(type_pair.first.c_str(), *allocator); if (!type_name) { RMW_SET_ERROR_MSG("failed to allocate memory for type name"); rmw_ret = RMW_RET_BAD_ALLOC; goto cleanup; } topic_names_and_types->types[index].data[type_index] = type_name; + // copy the hash into the hashes array (if allocated) + if (topic_names_and_types->type_hashes[index]) { + topic_names_and_types->type_hashes[index][type_index] = type_pair.second; + } ++type_index; } ++index; @@ -933,18 +1012,22 @@ GraphCache::get_names_and_types( // TODO(ivanpauno): Avoid using an intermediate representation. // We need a way to reallocate `topic_names_and_types.names` and `topic_names_and_types.names`. // Or have a good guess of the size (lower bound), and then shrink. - NamesAndTypes topics; + NamesAndTypesAndHashes topics; { std::lock_guard guard(mutex_); + // Forward data_services_ so __get_names_and_types can return the + // service-level hash for entities that belong to a service. __get_names_and_types( data_readers_, demangle_topic, demangle_type, + data_services_, topics); __get_names_and_types( data_writers_, demangle_topic, demangle_type, + data_services_, topics); } @@ -975,14 +1058,18 @@ __find_node( } static -NamesAndTypes +NamesAndTypesAndHashes __get_names_and_types_from_gids( const GraphCache::EntityGidToInfo & entities_map, const GraphCache::GidSeq & gids, DemangleFunctionT demangle_topic, - DemangleFunctionT demangle_type) + DemangleFunctionT demangle_type, + // Same role as in __get_names_and_types: when an entity is part of a + // service, the by-node graph query must report the service-level hash, + // not the per-direction message hash. + const GraphCache::EntityGidToServiceInfo & services) { - NamesAndTypes topics; + NamesAndTypesAndHashes topics; for (const auto & gid_msg : gids) { rmw_gid_t gid; @@ -995,7 +1082,14 @@ __get_names_and_types_from_gids( if ("" == demangled_topic_name) { continue; } - topics[demangled_topic_name].insert(demangle_type(it->second.topic_type)); + const std::string type_name = demangle_type(it->second.topic_type); + const rosidl_type_hash_t hash = + _select_hash_for_entity(gid, it->second.topic_type_hash, services); + _insert_type_hash_or_warn( + topics[demangled_topic_name], + demangled_topic_name, + type_name, + hash); } return topics; } @@ -1013,6 +1107,9 @@ __get_names_and_types_by_node( DemangleFunctionT demangle_topic, DemangleFunctionT demangle_type, GetEntitiesGidsFuncT get_entities_gids, + // Threaded through to __get_names_and_types_from_gids so service hashes + // are reported correctly on the by-node graph query paths. + const GraphCache::EntityGidToServiceInfo & services, rcutils_allocator_t * allocator, rmw_names_and_types_t * topic_names_and_types) { @@ -1033,11 +1130,12 @@ __get_names_and_types_by_node( return RMW_RET_NODE_NAME_NON_EXISTENT; } - NamesAndTypes topics = __get_names_and_types_from_gids( + NamesAndTypesAndHashes topics = __get_names_and_types_from_gids( entities_map, get_entities_gids(*node_info_ptr), demangle_topic, - demangle_type); + demangle_type, + services); return __populate_rmw_names_and_types( topics, @@ -1070,6 +1168,7 @@ GraphCache::get_writer_names_and_types_by_node( demangle_topic, demangle_type, __get_writers_gids, + data_services_, allocator, topic_names_and_types); } @@ -1099,6 +1198,7 @@ GraphCache::get_reader_names_and_types_by_node( demangle_topic, demangle_type, __get_readers_gids, + data_services_, allocator, topic_names_and_types); } diff --git a/rmw_dds_common/test/test_graph_cache.cpp b/rmw_dds_common/test/test_graph_cache.cpp index c9927ad..ae13d57 100644 --- a/rmw_dds_common/test/test_graph_cache.cpp +++ b/rmw_dds_common/test/test_graph_cache.cpp @@ -24,6 +24,7 @@ #include "osrf_testing_tools_cpp/scope_exit.hpp" #include "rcutils/testing/fault_injection.h" +#include "rosidl_runtime_c/type_hash.h" #include "rmw/qos_profiles.h" #include "rmw/topic_endpoint_info.h" #include "rmw/topic_endpoint_info_array.h" @@ -65,6 +66,7 @@ struct NameAndTypes { std::string name; std::vector types; + std::vector type_hashes = {}; }; void check_names_and_types( @@ -82,6 +84,22 @@ void check_names_and_types( for (size_t j = 0; j < expected_types.size(); j++) { EXPECT_EQ(expected_types[j], types.data[j]); } + if (!item.type_hashes.empty()) { + ASSERT_NE(nullptr, names_and_types.type_hashes); + ASSERT_NE(nullptr, names_and_types.type_hashes[i]); + ASSERT_EQ(item.type_hashes.size(), expected_types.size()); + for (size_t j = 0; j < item.type_hashes.size(); j++) { + EXPECT_EQ( + item.type_hashes[j].version, + names_and_types.type_hashes[i][j].version); + EXPECT_EQ( + 0, + memcmp( + item.type_hashes[j].value, + names_and_types.type_hashes[i][j].value, + ROSIDL_TYPE_HASH_SIZE)); + } + } } } @@ -93,6 +111,17 @@ identity_demangle(const std::string & name) using DemangleFunctionT = GraphCache::DemangleFunctionT; +rosidl_type_hash_t +make_test_hash(uint8_t seed) +{ + rosidl_type_hash_t hash = rosidl_get_zero_initialized_type_hash(); + hash.version = 1; + for (size_t i = 0; i < ROSIDL_TYPE_HASH_SIZE; ++i) { + hash.value[i] = static_cast(seed + i); + } + return hash; +} + void check_results( const GraphCache & graph_cache, @@ -444,6 +473,123 @@ TEST(test_graph_cache, add_remove_entities) check_results_by_topic(graph_cache, "topic4"); } +TEST(test_graph_cache, type_hashes_populated) +{ + GraphCache graph_cache; + + const auto hash_int = make_test_hash(0x10); + const auto hash_str = make_test_hash(0x20); + + EXPECT_TRUE( + graph_cache.add_entity( + gid_from_string("reader1"), + "topic1", + "Str", + hash_str, + gid_from_string("participant1"), + rmw_qos_profile_default, + true)); + + EXPECT_TRUE( + graph_cache.add_entity( + gid_from_string("reader2"), + "topic1", + "Int", + hash_int, + gid_from_string("participant1"), + rmw_qos_profile_default, + true)); + + check_results( + graph_cache, + {}, + { + {"topic1", {"Int", "Str"}, {hash_int, hash_str}}, + }); +} + +// A service is represented on DDS as a pair of request/reply topics, each +// carrying a different message type with its own topic_type_hash. The hash +// the user is meant to see, however, is the hash of the service descriptor +// (passed in here as `service_type_hash`), which is identical on both sides +// of the request/reply pair. This test pins down that the graph query +// returns the service hash for service entries, not the per-direction +// message hash that lives in topic_type_hash. Regression coverage for the +// case where __get_names_and_types previously surfaced the wrong field. +TEST(test_graph_cache, service_type_hash_overrides_topic_hash) +{ + GraphCache graph_cache; + + // Distinct test hashes so a mix-up between the two would be visible. + const auto request_msg_hash = make_test_hash(0x30); // hash of Request type + const auto reply_msg_hash = make_test_hash(0x40); // hash of Reply type + const auto service_hash = make_test_hash(0xA0); // hash of service descriptor + + // The server's reader (for incoming requests) carries the request message + // hash on the underlying topic, but is part of a service so the service + // descriptor hash is also recorded. + EXPECT_TRUE( + graph_cache.add_entity( + gid_from_string("srv_reader"), + "rq/svc1Request", + "ServiceRequest", + request_msg_hash, + gid_from_string("participant1"), + rmw_qos_profile_default, + true, + &service_hash)); + + // The server's writer (for outgoing replies) carries the reply message + // hash on its topic. Same service, so the same service descriptor hash. + EXPECT_TRUE( + graph_cache.add_entity( + gid_from_string("srv_writer"), + "rr/svc1Reply", + "ServiceReply", + reply_msg_hash, + gid_from_string("participant1"), + rmw_qos_profile_default, + false, + &service_hash)); + + // Demangle the DDS request/reply topic names to the same service name so + // the graph cache groups them together as a single service entry. + auto service_demangle = [](const std::string & dds_name) -> std::string { + if (dds_name.rfind("rq/", 0) == 0) {return "/svc1";} + if (dds_name.rfind("rr/", 0) == 0) {return "/svc1";} + return ""; + }; + auto type_demangle = [](const std::string &) -> std::string {return "svc_pkg/srv/Svc1";}; + + rmw_names_and_types_t result = rmw_get_zero_initialized_names_and_types(); + rcutils_allocator_t allocator = rcutils_get_default_allocator(); + ASSERT_EQ( + RMW_RET_OK, + graph_cache.get_names_and_types(service_demangle, type_demangle, &allocator, &result)); + + // Service must appear exactly once and its hash must be the service + // descriptor hash, not either of the per-direction message hashes. + ASSERT_EQ(1u, result.names.size); + EXPECT_STREQ("/svc1", result.names.data[0]); + ASSERT_EQ(1u, result.types[0].size); + EXPECT_STREQ("svc_pkg/srv/Svc1", result.types[0].data[0]); + ASSERT_NE(nullptr, result.type_hashes); + ASSERT_NE(nullptr, result.type_hashes[0]); + EXPECT_EQ(service_hash.version, result.type_hashes[0][0].version); + EXPECT_EQ( + 0, + memcmp(service_hash.value, result.type_hashes[0][0].value, ROSIDL_TYPE_HASH_SIZE)); + // And confirm it is NOT either of the underlying message hashes. + EXPECT_NE( + 0, + memcmp(request_msg_hash.value, result.type_hashes[0][0].value, ROSIDL_TYPE_HASH_SIZE)); + EXPECT_NE( + 0, + memcmp(reply_msg_hash.value, result.type_hashes[0][0].value, ROSIDL_TYPE_HASH_SIZE)); + + ASSERT_EQ(RMW_RET_OK, rmw_names_and_types_fini(&result)); +} + void add_participants( GraphCache & graph_cache, @@ -606,6 +752,53 @@ void dissociate_entities( } } +TEST(test_graph_cache, type_hashes_populated_by_node) +{ + GraphCache graph_cache; + + const auto hash_str = make_test_hash(0x20); + const auto hash_int = make_test_hash(0x10); + + add_participants(graph_cache, {"participant1"}); + add_nodes(graph_cache, {{"participant1", "ns", "node1"}}); + + EXPECT_TRUE( + graph_cache.add_entity( + gid_from_string("reader1"), + "topic1", + "Str", + hash_str, + gid_from_string("participant1"), + rmw_qos_profile_default, + true)); + + EXPECT_TRUE( + graph_cache.add_entity( + gid_from_string("writer1"), + "topic1", + "Int", + hash_int, + gid_from_string("participant1"), + rmw_qos_profile_default, + false)); + + associate_entities( + graph_cache, + { + {"reader1", true, "participant1", "ns", "node1"}, + {"writer1", false, "participant1", "ns", "node1"}, + }); + + check_results_by_node( + graph_cache, "ns", "node1", + { + {"topic1", {"Str"}, {hash_str}}, + }, + { + {"topic1", {"Int"}, {hash_int}}, + }); +} + rmw_dds_common::msg::ParticipantEntitiesInfo get_participant_entities_info_msg(const ParticipantEntitiesInfo & info) {