2026-09-05

rmw_dds_common 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/ros2/rmw_dds_common
版本：1.6.0（Humble），子包 1 个，构建类型 ament_cmake，语言 C++，许可证 Apache 2.0，质量等级 QL1。

rmw_dds_common 是 ROS 2 中 DDS 系 RMW 实现的共享 C++ 库：提供 ROS Graph 缓存、跨进程节点发现消息、QoS 兼容性检查 与 GID/安全/时间 等工具。它不直接调用 DDS API，也不实现 rmw_* 符号；各 rmw_*_cpp 在 init/graph 路径中嵌入并驱动本库。

设计背景：ROS 2 Node Discovery (design #250)

1. 总体认识

1.1 核心职责

能力	说明
GraphCache	维护 Participant / Node / DataReader / DataWriter 关系与 QoS
发现消息	定义 `ParticipantEntitiesInfo` 等 msg，在 `ros_discovery_info` topic 传播
Context 模板	`rmw_dds_common::Context` 汇总 graph 所需 RMW 句柄与线程字段
QoS 工具	`qos_profile_check_compatible` 实现 RMW QoS 兼容 API
GID 工具	`rmw_gid_t` 比较、转换、调试输出
SROS2 辅助	从 enclave 目录收集 DDS Security 证书文件路径
时间转换	`clamp_rmw_time_to_dds_time` 适配 DDS 32 位 Time/Duration

1.2 在 ROS 2 栈中的位置

消费者	使用方式
rmw_cyclonedds_cpp	`rmw_context_impl_t::common` 嵌入 `Context`；graph 全走 GraphCache
rmw_connextdds	同上 + `graph_cache.hpp` 钩子
rmw_fastrtps_cpp	主要使用 `qos_profile_check_compatible`；graph 仍有自有逻辑
rcl	间接通过 RMW graph API（`rmw_get_node_names` 等）

关键区分：本库是 库（library），不是 RMW 实现；链接 librmw_dds_common.so 不会提供 rmw_init。

2. 包结构与构建

rmw_dds_common/
└── rmw_dds_common/
    ├── msg/                          # rosidl 消息定义
    │   ├── Gid.msg
    │   ├── NodeEntitiesInfo.msg
    │   └── ParticipantEntitiesInfo.msg
    ├── include/rmw_dds_common/       # 6 个头文件
    ├── src/                          # 5 个 .cpp（~1646 行）
    ├── test/                         # gmock + benchmark
    └── docs/FEATURES.md

组件	行数（约）	说明
`graph_cache.cpp`	1069	GraphCache 全部逻辑
`graph_cache.hpp`	566	API 与内部类型
`qos.cpp`	377	QoS 兼容性
`gid_utils.cpp`	75	GID 转换
`security.cpp`	64	SROS2 文件查找
`time_utils.cpp`	61	时间截断

2.1 依赖（`package.xml`）

rmw_dds_common
├── rmw                    # rmw_gid_t、qos、topic_endpoint_info 等
├── rcutils                # 分配器、日志、字符串
├── rcpputils              # 文件系统（security）
├── rosidl_runtime_cpp     # 生成消息 C++ 类型
└── rosidl_default_generators

2.2 构建产物

rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/Gid.msg"
  "msg/NodeEntitiesInfo.msg"
  "msg/ParticipantEntitiesInfo.msg")

add_library(${PROJECT_NAME}_library SHARED
  src/gid_utils.cpp
  src/graph_cache.cpp
  src/qos.cpp
  src/security.cpp
  src/time_utils.cpp)
# OUTPUT_NAME → librmw_dds_common.so

库 PUBLIC 链接 rmw::rmw 与生成的 rosidl_typesupport_cpp 目标。

3. 发现消息（rosidl）

3.1 消息层次

ParticipantEntitiesInfo
├── Gid gid                              # DDS Participant 的 rmw_gid
└── NodeEntitiesInfo[] node_entities_info_seq
        ├── string node_namespace
        ├── string node_name
        ├── Gid[] reader_gid_seq         # 该节点拥有的 DataReader
        └── Gid[] writer_gid_seq         # 该节点拥有的 DataWriter

1 2	# Gid.msg — 与 rmw_gid_t.data 对齐（24 字节） char[24] data

3.2 传播路径（由各 RMW 实现 wiring）

本地 rmw_create_node / create_publisher
  → GraphCache::add_node / associate_writer
  → 返回 ParticipantEntitiesInfo
  → rmw_publish("ros_discovery_info", msg)

远端 rmw_take("ros_discovery_info")
  → GraphCache::update_participant_entities(msg)
  → 合并 node ↔ endpoint GID 映射

DDS builtin topic（DCPSPublication/Subscription/Participant）则通过 add_reader / add_writer 填充 topic 名、类型名、QoS，与 ROS 层 node 关联正交——GraphCache 将两路信息合并后回答 graph 查询。

4. GraphCache 数据模型

4.1 三张内部表

class GraphCache {
  EntityGidToInfo data_writers_;      // gid → EntityInfo
  EntityGidToInfo data_readers_;
  ParticipantToNodesMap participants_; // gid → ParticipantInfo
  std::mutex mutex_;
};

结构	字段	含义
EntityInfo	`topic_name`, `topic_type`, `participant_gid`, `qos`	单个 DataReader/Writer 的 DDS 发现信息
ParticipantInfo	`node_entities_info_seq`, `enclave`	某 Participant 下的 ROS 节点及 endpoint GID 列表

EntityGidToInfo 使用 Compare_rmw_gid_t 作为 map 键（按 24 字节 lexicographical 比较）。

4.2 更新 API 分组

graph_cache.hpp 将方法分为四组：

分组	方法	数据来源
dds_discovery_api	`add_reader/writer`, `remove_*`, `add_entity`	DCPS builtin / 远端 DDS 发现
common_api	`add_participant`, `remove_participant`	Participant 出现/消失
ros_discovery_api	`update_participant_entities`	`ros_discovery_info` 消息
local_api	`add_node`, `remove_node`, `associate_`, `dissociate_`	本进程创建/销毁实体；返回 `ParticipantEntitiesInfo` 供 publish
introspection_api	`get_node_names`, `get_names_and_types`, `get_*_info_by_topic` 等	只读查询

4.3 变更回调

1
2
3

graph_cache.set_on_change_callback([]() {
  rmw_trigger_guard_condition(graph_guard_condition);
});

RMW 实现注册回调，在 graph 变化时唤醒 rcl wait set（用于 ros2 topic list 等刷新）。

5. GraphCache 工作流程

5.1 本地节点创建（典型时序）

5.2 本地 Publisher 创建

create_publisher()
  → GraphCache::add_writer(writer_gid, topic, type, participant_gid, qos)  // DDS 侧信息
  → GraphCache::associate_writer(writer_gid, participant_gid, node_name, ns) // 挂到 node
  → 返回 ParticipantEntitiesInfo → publish

5.3 远端更新

1
2
3

take(ros_discovery_info)
  → GraphCache::update_participant_entities(msg)
  → 覆盖 participants_[gid].node_entities_info_seq

5.4 introspection 查询

get_names_and_types：遍历 data_writers_/data_readers_，经 demangle 回调 还原 ROS topic/type 名，聚合为 rmw_names_and_types_t。
get_node_names：遍历 participants_ 中所有 NodeEntitiesInfo。
get_writers/readers_info_by_topic：按 DDS topic 名过滤 entity，填充 rmw_topic_endpoint_info_t（含 node 名、namespace、GID、QoS）。

Demangle 由 RMW 实现注入（DemangleFunctionT），本库不硬编码 rt/ 前缀规则。

6. Context 结构

namespace rmw_dds_common {
struct Context {
  rmw_gid_t gid;                        // 本 Participant 的 GID
  rmw_publisher_t * pub;                // 发布 ParticipantEntitiesInfo
  rmw_subscription_t * sub;             // 订阅 ros_discovery_info
  GraphCache graph_cache;
  std::mutex node_update_mutex;         // update + publish 互斥
  std::thread listener_thread;          // 由 RMW 创建，非本库
  std::atomic_bool thread_is_running;
  rmw_guard_condition_t * listener_thread_gc;
  rmw_guard_condition_t * graph_guard_condition;
};
}

本库只定义结构体，不创建线程或 DDS 实体。rmw_cyclonedds_cpp / rmw_connextdds 在 context init 时：

创建 internal pub/sub on ros_discovery_info
启动 listener/discovery 线程处理 builtin + ParticipantEntitiesInfo
将 graph_guard_condition 注册到 GraphCache 回调

7. GID 工具（`gid_utils`）

API	作用
`Compare_rmw_gid_t`	`std::map<rmw_gid_t, ...>` 排序键
`operator==`	24 字节 memcmp
`operator<<`	十六进制调试打印
`convert_gid_to_msg` / `convert_msg_to_gid`	`rmw_gid_t` ↔ `msg/Gid`

1
2
3

void convert_gid_to_msg(const rmw_gid_t * gid, msg::Gid * msg_gid) {
  std::memcpy(&msg_gid->data, gid->data, RMW_GID_STORAGE_SIZE);
}

implementation_identifier 字段不进入 msg，仅在 rmw_gid_t 全结构中使用。

8. QoS 兼容性（`qos.cpp`）

实现 rmw_qos_profile_check_compatible 的共享逻辑，被 Cyclone/Connext/FastRTPS RMW 转发调用。

8.1 判定级别

结果	含义
`RMW_QOS_COMPATIBILITY_OK`	可通信
`RMW_QOS_COMPATIBILITY_WARNING`	可能有问题（如 depth 不足）
`RMW_QOS_COMPATIBILITY_ERROR`	确定不兼容

8.2 主要 ERROR 规则（节选）

条件	原因
Pub BEST_EFFORT + Sub RELIABLE	可靠订阅收不到尽力 pub
Pub VOLATILE + Sub TRANSIENT_LOCAL	晚加入订阅者收不到历史
Sub deadline 更严而 Pub 无/更松	deadline 契约不满足
Pub AUTOMATIC liveliness + Sub MANUAL_BY_TOPIC	liveliness 策略冲突
Sub lease 短于 Pub lease	lease 不匹配

WARNING 包括 history depth 小于订阅需求等。reason 缓冲区可选，用 _append_to_buffer 拼接人类可读说明。

9. 安全工具（`security.cpp`）

bool get_security_files(
  const std::string & prefix,
  const std::string & secure_root,
  std::unordered_map<std::string, std::string> & result);

在 secure_root（SROS2 enclave 目录）查找必需文件：

键	文件名
IDENTITY_CA	identity_ca.cert.pem
CERTIFICATE	cert.pem
PRIVATE_KEY	key.pem
PERMISSIONS_CA	permissions_ca.cert.pem
GOVERNANCE	governance.p7s
PERMISSIONS	permissions.p7s

可选：CRL → crl.pem。缺失任一必需文件则返回 false 并清空 result。RMW 将 result 映射到 DDS Security QoS property。

10. 时间工具（`time_utils.cpp`）

DDS IDL Duration_t / Time_t 在 C 绑定中为 32 位 sec/nsec。clamp_rmw_time_to_dds_time：

将 nsec 归一化到 < 1s
若总时长超过 INT_MAX 秒，截断到 INT_MAX sec + (10^9-1) nsec 并打 debug 日志

各 RMW 在 ROS QoS → DDS QoS 转换时调用，避免静默溢出。

11. 线程安全与锁

锁	位置	用途
`GraphCache::mutex_`	GraphCache 内部	所有 cache 读写
`Context::node_update_mutex`	RMW 层使用	`add_node` + `rmw_publish` 原子性

GraphCache 方法自身已加锁；RMW 在 publish 发现消息前额外持有 node_update_mutex，防止交错更新导致远端 cache 状态不一致（见 Cyclone rmw_create_node 注释）。

12. 测试

测试	文件	覆盖
`test_graph_cache`	`test/test_graph_cache.cpp`	add/remove/associate、introspection、并发
`test_gid_utils`	`test/test_gid_utils.cpp`	GID 比较与转换
`test_qos`	`test/test_qos.cpp`	QoS 兼容边界
`test_time_utils`	`test/test_time_utils.cpp`	时间截断
`test_security`	`test/test_security.cpp`	enclave 文件查找
`benchmark_graph_cache`	`test/benchmark/`	GraphCache 性能

启用 RCUTILS_ENABLE_FAULT_INJECTION 测试分配失败路径。

13. RMW 实现集成要点

以 rmw_cyclonedds_cpp 为例（Connext 结构类似）：

struct rmw_context_impl_s {
  rmw_dds_common::Context common;
  // ... DDS handles ...
};

// init 时
common.pub = create_publisher(..., "ros_discovery_info", ...);
common.sub = create_subscription(..., "ros_discovery_info", ...);
common.gid = ppant_gid;

// create_node 时
std::lock_guard<std::mutex> guard(common.node_update_mutex);
auto msg = common.graph_cache.add_node(common.gid, name, namespace_);
rmw_publish(common.pub, &msg, ...);

// DCPS 回调
graph_cache.add_writer(gid, dds_topic, dds_type, participant_gid, qos);

// graph API
return common.graph_cache.get_node_names(..., demangle_fn);

Fast-DDS RMW 对 GraphCache 的采用程度较低，但 QoS 检查 与 security 文件 仍复用本库。

14. 与 rmw 包的分工

主题	rmw 包	rmw_dds_common
`rmw_*` API 声明	✅	❌
`rmw_validate_*`、QoS 字符串	✅	❌
Graph 缓存与发现 msg	❌	✅
`rmw_qos_profile_check_compatible` 实现	声明	✅ 实现
DDS 调用	❌	❌

15. 调试建议

Graph 不完整：确认 ros_discovery_info pub/sub 存在；检查 update_participant_entities 是否被调用。
topic list 与 echo 不一致：可能是 demangle 回调未剥离 rt/；或 DDS 侧 add_writer 与 ROS 侧 associate_writer 未配对。
QoS 兼容误报：阅读 reason 字符串；对照 qos.cpp 规则。
Security 启动失败：用 get_security_files 检查 enclave 目录六文件是否齐全。
打印 cache 状态：operator<<(ostream, GraphCache) 可 dump 内部 map（调试构建）。

16. 推荐阅读顺序

docs/FEATURES.md — 功能清单
msg/*.msg — 发现协议数据结构
include/rmw_dds_common/context.hpp — RMW 嵌入字段
graph_cache.hpp API 分组 — 理解四路更新
graph_cache.cpp — add_node / associate_writer / get_names_and_types
qos.cpp — QoS 兼容规则
gid_utils.cpp — GID 互转
下游 rmw_cyclonedds_cpp/src/rmw_node.cpp — discovery 线程如何喂 GraphCache
设计文档 ros2/design#250

17. 小结

rmw_dds_common 是 ROS 2 DDS RMW 之间的“图与发现公共层”：

对上：为 rmw_get_node_names、ros2 topic list 等提供统一的缓存与查询；
对下：通过标准 msg 在 Participant 间同步 Node ↔ Endpoint 映射，并与 DDS builtin 发现互补；
横向：QoS、GID、Security、Time 工具避免在 Cyclone/Connext/FastRTPS 中重复实现。

理解本库是阅读 rmw_cyclonedds、rmw_connextdds graph 代码的前提；其本身 不包含 DDS 传输，仅维护 ROS 语义层面的分布式图。

文章互动

阅读 --

留言

0 条留言

正在加载留言…

fishBlog

专注嵌入式开发

rmw_dds_common 源码详细分析

rmw_dds_common 源码详细分析

1. 总体认识

1.1 核心职责

1.2 在 ROS 2 栈中的位置

2. 包结构与构建

2.1 依赖（`package.xml`）

2.2 构建产物

3. 发现消息（rosidl）

3.1 消息层次

3.2 传播路径（由各 RMW 实现 wiring）

4. GraphCache 数据模型

4.1 三张内部表

4.2 更新 API 分组

4.3 变更回调

5. GraphCache 工作流程

5.1 本地节点创建（典型时序）

5.2 本地 Publisher 创建

5.3 远端更新

5.4 introspection 查询

6. Context 结构

7. GID 工具（`gid_utils`）

8. QoS 兼容性（`qos.cpp`）

8.1 判定级别

8.2 主要 ERROR 规则（节选）

9. 安全工具（`security.cpp`）

10. 时间工具（`time_utils.cpp`）

11. 线程安全与锁

12. 测试

13. RMW 实现集成要点

14. 与 rmw 包的分工

15. 调试建议

16. 推荐阅读顺序

17. 小结

文章互动

留言

rmw_dds_common 源码详细分析

1. 总体认识

1.1 核心职责

1.2 在 ROS 2 栈中的位置

2. 包结构与构建

2.1 依赖（package.xml）

2.2 构建产物

3. 发现消息（rosidl）

3.1 消息层次

3.2 传播路径（由各 RMW 实现 wiring）

4. GraphCache 数据模型

4.1 三张内部表

4.2 更新 API 分组

4.3 变更回调

5. GraphCache 工作流程

5.1 本地节点创建（典型时序）

5.2 本地 Publisher 创建

5.3 远端更新

5.4 introspection 查询

6. Context 结构

7. GID 工具（gid_utils）

8. QoS 兼容性（qos.cpp）

8.1 判定级别

8.2 主要 ERROR 规则（节选）

9. 安全工具（security.cpp）

10. 时间工具（time_utils.cpp）

11. 线程安全与锁

12. 测试

13. RMW 实现集成要点

14. 与 rmw 包的分工

15. 调试建议

16. 推荐阅读顺序

17. 小结

文章互动

留言

2.1 依赖（`package.xml`）

7. GID 工具（`gid_utils`）

8. QoS 兼容性（`qos.cpp`）

9. 安全工具（`security.cpp`）

10. 时间工具（`time_utils.cpp`）