rmw_fastrtps 源码详细分析
工作区路径:/home/cp/work2/ros2Learn/ros2_humble/src/ros2/rmw_fastrtps
版本:6.2.10(Humble),子包 3 个,构建类型 ament_cmake,语言 C++,许可证 Apache 2.0,质量等级 QL2。
rmw_fastrtps 是 ROS 2 面向 eProsima Fast-DDS 的 RMW 实现族,将标准 rmw_* C API 映射到 Fast-DDS 的 DDS C++ API(fastdds/dds/*)及 RTPS 层。Humble 上它是 默认 RMW(get_default_rmw_implementation() 优先 rmw_fastrtps_cpp);也可选用 rmw_fastrtps_dynamic_cpp。
外部依赖:DDS 本体在独立仓库
eProsima/Fast-DDS(Humble 对应 2.6.x),序列化依赖Fast-CDR。
1. 总体认识
1.1 核心职责
| 能力 | 说明 |
|---|---|
| 双 RMW 注册 | rmw_fastrtps_cpp(静态 typesupport)与 rmw_fastrtps_dynamic_cpp(introspection)各自注册为独立 .so |
| 共享核心 | rmw_fastrtps_shared_cpp 实现全部 __rmw_* 逻辑,不单独注册 RMW |
| Fast-CDR 序列化 | 通过 TypeSupport 继承 TopicDataType,桥接 ROS 消息 ↔ RTPS payload |
| Graph 发现 | 嵌入 rmw_dds_common::Context;Fast-DDS 扩展 listener + ros_discovery_info topic |
| Service/Client | Fast-DDS extended RPC:SampleIdentity / related_sample_identity |
| 零拷贝 | SharedMem 传输 + Data Sharing + Loan API(plain 类型 + XML 启用 Data Sharing) |
| QoS 可配置 | ROS QoS + 环境变量 + Fast-DDS XML profile |
1.2 在 ROS 2 栈中的位置
| 对比项 | rmw_fastrtps_cpp | rmw_fastrtps_dynamic_cpp | rmw_cyclonedds_cpp | rmw_connextdds |
|---|---|---|---|---|
| DDS | Fast-DDS | Fast-DDS | CycloneDDS | RTI Connext |
| 序列化 | Fast-CDR + 编译期 callbacks | Fast-CDR + introspection 遍历 | 自研 introspection CDR | Fast-CDR typesupport |
| 源码组织 | shared + 薄 API + 实体创建 | 同左 + TypeSupportRegistry | 单巨型 rmw_node.cpp |
common + 薄 API |
| Graph | rmw_dds_common |
同左 | rmw_dds_common |
rmw_dds_common |
| Service RPC | extended(SampleIdentity) | 同左 | basic header | basic/extended 可配 |
| 默认 Humble RMW | 是 | 否 | 否 | 否 |
| SHM 零拷贝 | Data Sharing + SHM transport | 同左 | iceoryx(需 RouDi) | Pro 异步 publish |
1.3 与 Fast-DDS 的边界
1 | rcl → rmw_fastrtps_cpp / rmw_fastrtps_dynamic_cpp |
2. 三包结构与构建
1 | rmw_fastrtps/ |
| 包 | .cpp 行数(约) |
职责 |
|---|---|---|
rmw_fastrtps_shared_cpp |
6898 | 全部 __rmw_*、wait、graph listener、QoS、安全 |
rmw_fastrtps_cpp |
4328 | rmw_* 导出 + 静态 TypeSupport 实体创建 |
rmw_fastrtps_dynamic_cpp |
4588 | 同上 + TypeSupportRegistry |
| 合计 | ~17644 | — |
2.1 依赖(package.xml 摘要)
1 | rmw_fastrtps_cpp |
3. 双实现 + identifier 模式
3.1 两个 RMW identifier
| 包 | identifier 常量 | 注册方式 |
|---|---|---|
rmw_fastrtps_cpp |
"rmw_fastrtps_cpp"(eprosima_fastrtps_identifier) |
register_rmw_implementation() |
rmw_fastrtps_dynamic_cpp |
"rmw_fastrtps_dynamic_cpp" |
同上 |
每个 handle(rmw_node_t、rmw_publisher_t 等)的 implementation_identifier 必须与传入的 identifier 一致,否则返回 RMW_RET_INCORRECT_RMW_IMPLEMENTATION。
3.2 薄包装 → 共享实现
rmw_fastrtps_cpp 中典型模式:
1 | // rmw_fastrtps_cpp/src/rmw_publisher.cpp |
共享库中校验 identifier 并执行 Fast-DDS 调用:
1 | // rmw_fastrtps_shared_cpp/src/rmw_publish.cpp |
3.3 差异仅在 TypeSupport 与实体创建
| 步骤 | rmw_fastrtps_cpp |
rmw_fastrtps_dynamic_cpp |
|---|---|---|
| 获取 TypeSupport | TypeSupport::set_members(message_type_support_callbacks_t*) |
TypeSupportRegistry::get_message_type_support() |
| typesupport 来源 | rosidl_typesupport_fastrtps_* |
rosidl_typesupport_introspection_* |
| 生命周期 | 每 publisher 独立 TypeSupport 实例 | Registry 引用计数共享 |
| 其余路径 | 调用相同的 __rmw_publish / __rmw_take / __rmw_wait |
同左 |
4. Context 与 Participant 生命周期
4.1 rmw_context_impl_s
定义于 rmw_fastrtps_shared_cpp/include/rmw_fastrtps_shared_cpp/rmw_context_impl.hpp:
1 | struct rmw_context_impl_s { |
4.2 初始化时序
1 | rmw_init(options, context) |
设计要点:Participant 与 graph 基础设施 按 context 懒加载,首个 rmw_create_node 时才创建 Fast-DDS 实体;多 node 共享同一 DomainParticipant。
4.3 create_participant 关键行为
participant.cpp 中 create_participant():
| 配置项 | 行为 |
|---|---|
| enclave | 写入 DomainParticipantQos.user_data:enclave=<name>;,供 graph 过滤 |
| localhost_only | 禁用 builtin transport,仅 127.0.0.1 UDPv4 + SharedMem |
| RMW_FASTRTPS_USE_QOS_FROM_XML | 1 → leave_middleware_default_qos=true,history/publication 由 XML 决定 |
| RMW_FASTRTPS_PUBLICATION_MODE | SYNCHRONOUS / ASYNCHRONOUS / AUTO(XML 模式时忽略) |
| 默认 memory policy | PREALLOCATED_WITH_REALLOC(非 XML 模式) |
| Data Sharing | 默认 OFF;通过 XML 启用 |
| 安全 | security_root_path 存在时配置 PKI-DH + AES-GCM-GMAC + Access-Permissions |
Participant 内还预创建 Publisher 与 Subscriber 对象(entity_creation_mutex_ 保护后续 topic/writer/reader 创建)。
5. Graph 与 rmw_dds_common
5.1 双通道更新 GraphCache
通道 1 — 应用层 discovery topic
- Topic:
ros_discovery_info(avoid_ros_namespace_conventions=true) - 消息:
rmw_dds_common::msg::ParticipantEntitiesInfo - Publisher QoS:
TRANSIENT_LOCAL+KEEP_LAST depth=1 - Subscriber QoS:
KEEP_ALL listener_thread循环__rmw_wait+__rmw_take,忽略本 participant 的 GID
通道 2 — Fast-DDS 扩展 listener
ParticipantListener(custom_participant_info.hpp)重写:
on_participant_discovery→ 解析 user_data 中enclave,graph_cache.add/remove_participanton_subscriber_discovery/on_publisher_discovery→graph_cache.add/remove_entity(含 QoS)
节点增删时在 __rmw_create_node / __rmw_destroy_node 中加锁发布 ParticipantEntitiesInfo(node_update_mutex 保证 publish 原子性)。
5.2 Graph Guard Condition
graph_cache.set_on_change_callback 注册 lambda,在图变化时 __rmw_trigger_guard_condition(graph_guard_condition),供 rmw_node_get_graph_guard_condition 使用。
6. TypeSupport 与序列化
6.1 抽象基类
TypeSupport 继承 eprosima::fastdds::dds::TopicDataType:
1 | struct SerializedData { |
serialize()/deserialize()在TypeSupport_impl.cpp中实现,根据is_cdr_buffer分支- 普通 publish:
is_cdr_buffer=false,impl指向message_type_support_callbacks_t或 introspection members
6.2 静态 typesupport(rmw_fastrtps_cpp)
type_support_common.cpp:
set_members()读取ROSIDL_TYPESUPPORT_FASTRTPS_*宏判断 plain/boundedserializeROSmessage()调用编译期生成的callbacks->cdr_serializeis_plain()决定 Data Sharing / loan 能力
6.3 动态 typesupport(rmw_fastrtps_dynamic_cpp)
TypeSupportRegistry(单例 + 引用计数):
get_message_type_support()/get_request_type_support()/get_response_type_support()- 支持
rosidl_typesupport_introspection_c与_cpp - 进程退出时
~TypeSupportRegistry()清理残留
适用场景:运行时切换消息类型、工具链不生成 fastrtps typesupport 的包;性能通常低于静态路径。
7. Publisher / Subscription 数据路径
7.1 创建流程(以 publisher 为例)
rmw_fastrtps_cpp/src/publisher.cpp → create_publisher():
- 校验 topic 名、QoS、
require_unique_network_flow_endpoints map_ros_to_dds()生成 topic/type 名(rt/前缀、ros2msg/ros2srv等)- 构造包内
TypeSupport,注册到 participant rmw_qos_to_dds_attributes()+ XML profile 加载(按 topic 名匹配profile_name)- 创建
Topic+DataWriter+ 可选PubListener - 填充
CustomPublisherInfo,设置publisher_gid can_loan_messages = has_data_sharing && type_support->is_plain()
7.2 发布
| API | 路径 |
|---|---|
rmw_publish |
SerializedData{ros_message} → TypeSupport::serialize → DataWriter::write |
rmw_publish_serialized_message |
预序列化 CDR buffer → is_cdr_buffer=true |
rmw_publish_loaned_message |
直接 DataWriter::write(loaned_buffer)(无 SerializedData 包装) |
7.3 接收
rmw_take.cpp(~558 行):
DataReader::take→deserializeROSmessage或 loaned 路径- 支持
rmw_take_with_info(publication handle 等) ignore_local_publications在 init 时设为true(注释:fastrtps 尚未完整实现该选项语义)
7.4 实体私有结构
| 结构体 | 主要字段 |
|---|---|
CustomPublisherInfo |
DataWriter*, TypeSupport, type_support_impl_, PubListener |
CustomSubscriberInfo |
DataReader*, loan 相关状态 |
CustomServiceInfo |
request reader + response writer |
CustomClientInfo |
request writer + response reader + GUID 映射 |
均继承或组合 CustomEventInfo,支持 QoS/liveliness/deadline 事件。
8. Service / Client(Extended RPC)
Fast-DDS 使用 DDS 扩展 WriteParams 关联 request/response,与 Connext extended 模式互操作。
8.1 Client 发请求
rmw_request.cpp — __rmw_send_request:
1 | wparams.related_sample_identity().writer_guid() = info->reader_guid_; |
8.2 Service 收请求 / 发响应
__rmw_take_request:从 request reader take,读取related_sample_identity填入rmw_service_info_t__rmw_send_response:设置wparams.related_sample_identity()指向原 request
8.3 Client 收响应
__rmw_take_response:校验 related_sample_identity.writer_guid 匹配 client 的 reader/writer GUID,再反序列化。
互操作:与 Cyclone basic header 不兼容;与 Connext extended 可互通。跨 RMW service 调用需统一 RPC 映射。
9. Wait Set 与 rmw_wait
rmw_wait.cpp 使用 Fast-DDS WaitSet:
- 遍历 subscriptions / clients / services / events,attach 各
StatusCondition - 若已有 untaken sample(
get_first_untaken_info)→skip_wait=true - attach guard conditions(含 graph GC、listener thread GC)
wait_set->wait()或wait(timeout)- 未就绪条目在 rcl 数组中置 NULL
WaitSet 存储于 rmw_wait_set_t->data(eprosima::fastdds::dds::WaitSet*)。
10. QoS 与环境变量
10.1 默认 Fast-DDS 策略(非 XML 模式)
| 策略 | 默认值 |
|---|---|
| History memory | PREALLOCATED_WITH_REALLOC |
| Publication mode | SYNCHRONOUS |
| Data Sharing | OFF |
10.2 环境变量
| 变量 | 作用 |
|---|---|
RMW_IMPLEMENTATION |
rmw_fastrtps_cpp 或 rmw_fastrtps_dynamic_cpp |
RMW_FASTRTPS_PUBLICATION_MODE |
SYNCHRONOUS / ASYNCHRONOUS / AUTO |
RMW_FASTRTPS_USE_QOS_FROM_XML |
1 时 history/publication/datasharing 由 XML 覆盖 |
FASTRTPS_DEFAULT_PROFILES_FILE |
XML profile 路径 |
| (工作目录) | DEFAULT_FASTRTPS_PROFILES.xml |
10.3 XML Profile 匹配规则
- Publisher/Subscription:
profile_name= 最终 topic 名(含 namespace,FQN 以/开头时不拼 namespace) - Service:
profile_name="service"或 mangled 服务名 - Client:
profile_name="client"或 mangled 服务名 - 回退:
is_default_profile="true"
10.4 ROS ↔ DDS 映射
qos.cpp / rmw_qos.cpp:
rmw_qos_to_dds_attributes()创建实体时使用rtps_qos_to_rmw_qos()供ParticipantListener写入 graphrmw_qos_profile_check_compatible()委托rmw_dds_common::qos_profile_check_compatible()
11. Loaned Message 与零拷贝
11.1 启用条件
1 | // publisher.cpp / rmw_take.cpp |
需同时满足:
- XML 中 Data Sharing 为
AUTOMATIC或ON(且RMW_FASTRTPS_USE_QOS_FROM_XML=1) - 消息类型为 plain(POD,静态 typesupport 由
ROSIDL_TYPESUPPORT_FASTRTPS_PLAIN_TYPE判定)
11.2 API 路径
| API | 实现要点 |
|---|---|
rmw_borrow_loaned_message |
DataWriter::loan_sample() |
rmw_publish_loaned_message |
直接 write 借出的 buffer |
rmw_return_loaned_message_from_publisher |
return_loan() |
rmw_take_loaned_message |
reader loan + 反序列化跳过 |
11.3 传输层
默认 builtin transports:UDPv4(跨主机)+ SharedMem(同主机)。零拷贝 pipeline = Loan API + Data Sharing;仅 SHM transport 不等同于零拷贝。
Humble 上 plain 类型仍需 XML 启用 Data Sharing;Iron 及以后对 POD 要求放宽(见仓库 README)。
12. 事件、安全与追踪
12.1 RMW 事件
通过 CustomEventInfo + DataWriterListener / DataReaderListener 收集:
RMW_EVENT_*_QOS_INCOMPATIBLERMW_EVENT_*_DEADLINE_MISSEDRMW_EVENT_LIVELINESS_*RMW_EVENT_MESSAGE_LOST
rmw_event.cpp + event_helpers.hpp 实现 rmw_take_event 与 callback 注册。
12.2 安全(SROS2)
participant.cpp 在 HAVE_SECURITY 且提供 security_root_path 时:
- 使用
rmw_dds_common::get_security_files()解析证书路径 - 配置
PropertyPolicy(PKI-DH、AES-GCM-GMAC、Access-Permissions) rmw_security_logging.cpp处理安全日志属性
12.3 追踪
关键路径插入 tracetools 追踪点(如 TRACEPOINT(rmw_publish, ...))。
12.4 特性查询
rmw_features.cpp 当前仅声明支持:
RMW_FEATURE_MESSAGE_INFO_PUBLICATION_SEQUENCE_NUMBER
13. 未实现或受限功能
| API / 能力 | 状态 |
|---|---|
rmw_init_publisher/subscription_allocation |
RMW_RET_UNSUPPORTED |
RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_STRICTLY_REQUIRED |
创建 pub/sub 时拒绝 |
ignore_local_publications |
订阅选项强制为 true,功能未完整实现 |
| Loan API | 需 Data Sharing + plain 类型 |
| Keyed topics | discovery 路径显式 keyed=false |
| Graph 与 typesupport | discovery 使用静态 ParticipantEntitiesInfo typesupport |
14. 与 rmw_dds_common 的协作
| 组件 | 用法 |
|---|---|
rmw_dds_common::Context |
context->impl->common;含 pub/sub/gid/graph_cache/listener 线程 |
GraphCache |
节点、participant、endpoint、topic 图 |
ParticipantEntitiesInfo |
ros_discovery_info 消息类型 |
qos_profile_check_compatible |
QoS 兼容 API |
security.hpp |
安全文件路径解析 |
gid_utils |
GID 比较(listener 忽略本地消息) |
Fast-DDS 实现与 Cyclone/Connext 新版共用同一 graph 协议,保证 ros2 topic list / ros2 node info 等工具跨 RMW 一致。
15. 互操作说明
| 场景 | 默认互操作 |
|---|---|
Pub/Sub(CDR + rt/ topic) |
与 Cyclone、Connext 等 通常可互通 |
| Service/Client | 同 RMW 或 extended 兼容实现 |
| Fast-DDS ↔ Cyclone service | 不互通(RPC 映射不同) |
| Data Sharing 零拷贝 | 同进程/同机 + 同 RMW + plain + XML 配置 |
16. 调试建议
- 确认 RMW:
echo $RMW_IMPLEMENTATION或ros2 doctor --report;默认应为rmw_fastrtps_cpp。 - Discovery 问题:检查防火墙与 multicast;
localhost_only时仅 127.0.0.1 + SHM。 - Graph 空:确认
ros_discovery_infopub/sub 创建成功;listener_thread是否运行;enclave user_data 是否解析。 - QoS 不符预期:用
rmw_publisher_get_actual_qos;检查RMW_FASTRTPS_USE_QOS_FROM_XML与 XML profile 名是否匹配 topic。 - 零拷贝不生效:确认
can_loan_messages、Data Sharing XML、消息是否 plain。 - Fast-DDS 日志:通过 Fast-DDS 自身日志配置(Consumer 等)排查 RTPS 层。
17. 推荐阅读顺序
- 仓库
README.md— 环境变量、XML 示例、零拷贝 rmw_fastrtps_cpp/src/rmw_init.cpp+init_rmw_context_impl.cpp— context 懒加载与 discovery 拓扑custom_participant_info.hpp— ParticipantListener 与 graph 关系listener_thread.cpp—ros_discovery_info消费循环rmw_fastrtps_cpp/src/publisher.cpp— 实体创建、XML QoS、loan 标志TypeSupport.hpp+TypeSupport_impl.cpp— 序列化核心rmw_publish.cpp/rmw_take.cpp— 数据平面rmw_request.cpp/rmw_response.cpp— Service RPCrmw_wait.cpp— WaitSet 语义type_support_registry.cpp(dynamic 包)— 运行时 typesupport- Fast-DDS 上游文档 — Data Sharing、XML profiles、PublishMode
18. 小结
rmw_fastrtps 是 Humble 默认 RMW,采用 三包分层 设计:
- 对上:
rmw_fastrtps_cpp/rmw_fastrtps_dynamic_cpp导出完整rmw_*API,差异仅在 typesupport 策略; - 对中:
rmw_fastrtps_shared_cpp集中实现__rmw_*,避免双倍维护 wait/graph/service 逻辑; - 对下:映射到 Fast-DDS 2.6.x + Fast-CDR,利用 SharedMem、Data Sharing、extended RPC 等 eProsima 扩展;
- 横向:通过
rmw_dds_common统一 ROS graph,与 Cyclone/Connext 工具链对齐。
选用 rmw_fastrtps_dynamic_cpp 可换运行时 introspection,但生产环境通常保留默认 rmw_fastrtps_cpp 以获得编译期 typesupport 性能。高级调优依赖 RMW_FASTRTPS_* 环境变量 与 FASTRTPS_DEFAULT_PROFILES_FILE XML 组合,而非仅 ROS QoS API。
正在加载留言…