ROS2►核心

rmw_cyclonedds 源码详细分析

rmw_cyclonedds 源码详细分析

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

rmw_cyclonedds_cpp 是 ROS 2 面向 Eclipse CycloneDDS 的 RMW 实现，将标准 rmw_* C API 映射到 Cyclone 的 DDS C API（dds/dds.h）及底层 DDSI serdata/sertype 插件接口。Humble 上常与 rmw_fastrtps_cpp 并列可选；启用共享内存时需 Cyclone 编译 SHM 支持并配置 CYCLONEDDS_URI。

外部依赖：DDS 本体在独立仓库 eclipse-cyclonedds/cyclonedds。未找到 CycloneDDS CMake 包时，本包 跳过编译。

1. 总体认识

1.1 核心职责

能力 说明
RMW 全量实现 几乎所有 rmw_* 符号在单文件 rmw_node.cpp（~5511 行）中 extern "C" 导出
自研 CDR 序列化 基于 rosidl introspection 遍历消息结构，非 Fast-CDR typesupport
DDSI 集成 自定义 sertype_rmw / serdata_rmw 接入 Cyclone 传输层
Graph 发现 嵌入 rmw_dds_common::Context，双通道（ParticipantEntitiesInfo + DCPS builtin）
Service/Client Basic RPC 映射：inline cdds_request_header（GUID + seq）
共享内存 可选 iceoryx SHM（DDS_HAS_SHM），支持 fixed-size 类型的 loan API
Domain 管理 按 domain id 延迟创建 Cyclone domain，支持 localhost-only

1.2 在 ROS 2 栈中的位置

上层rmw_cyclonedds 仓库ROS 共享DDS 栈rcl / rclcpp / rclpyrmw_cyclonedds_cpp\n~10K 行 C++rmwrmw_dds_commonrosidl_typesupport_introspectionCycloneDDS libddsciceoryx_binding_c\n可选 SHM
对比项 rmw_cyclonedds_cpp rmw_fastrtps_cpp rmw_connextdds
DDS CycloneDDS Fast-DDS RTI Connext
序列化 自研 introspection CDR Fast-CDR typesupport Fast-CDR typesupport
主源码组织 单巨型 .cpp + 序列化模块 多文件 OOP common + 薄 API 层
Graph rmw_dds_common 部分 common rmw_dds_common
Service RPC 自定义 basic header extended/basic 可配置 basic/extended
SHM 零拷贝 iceoryx（需配置） 有限 Pro 异步 publish

1.3 与 CycloneDDS 的边界

1
2
3
4
5
6
rcl → rmw_cyclonedds_cpp
        ├── rmw_node.cpp          # RMW 语义、实体生命周期、wait、graph
        ├── serdata.cpp / serdes  # DDSI 样本封装与 CDR 缓冲
        ├── TypeSupport2.cpp      # introspection → CDR 读写器
        └── dds_* API               # Participant/Writer/Reader/WaitSet
              └── Cyclone DDSI      # RTPS、发现、可选 SHM 传输

2. 包结构与构建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
rmw_cyclonedds/
├── README.md
├── shared_memory_support.md        # iceoryx / RouDi 使用说明
└── rmw_cyclonedds_cpp/
    ├── package.xml
    ├── CMakeLists.txt
    ├── cmake/get_rmw_cyclonedds_output_filter.cmake
    └── src/
        ├── rmw_node.cpp            # ★ 主实现 (~5511 行)
        ├── serdata.cpp / serdata.hpp
        ├── serdes.cpp / serdes.hpp
        ├── TypeSupport.cpp / TypeSupport2.cpp / *.hpp
        ├── Serialization.cpp       # CDR cursor 引擎
        ├── demangle.cpp
        ├── u16string.cpp             # wstring 序列化辅助
        ├── exception*.cpp
        ├── rmw_get_network_flow_endpoints.cpp
        └── rmw_cyclonedds_topic.idl  # 占位 IDL（dummy topic）
源文件 行数（约） 职责
rmw_node.cpp 5511 全部 RMW API、实体 struct、discovery 线程
serdata.cpp 740 serdata_rmw、SHM chunk 管理
TypeSupport_impl.hpp 537 sertype 操作函数
Serialization.cpp 621 CDR 对齐、序列化/反序列化 cursor
TypeSupport2.cpp 239 introspection 类型树
其余 <200 each 辅助

2.1 依赖（package.xml）

1
2
3
4
5
6
7
8
rmw_cyclonedds_cpp
├── cyclonedds              # libddsc（必需）
├── iceoryx_binding_c       # 仅当 Cyclone 启用 SHM 时链接
├── rmw / rmw_dds_common
├── rcutils / rcpputils
├── rosidl_runtime_c
├── rosidl_typesupport_introspection_c/cpp
└── tracetools              # 追踪点

Typesupport 注册：

1
2
3
register_rmw_implementation(
  "c:rosidl_typesupport_c:rosidl_typesupport_introspection_c"
  "cpp:rosidl_typesupport_cpp:rosidl_typesupport_introspection_cpp")

与 FastRTPS/Connext 不同，不注册 rosidl_typesupport_fastrtps_*。

2.2 条件编译：共享内存

1
2
3
4
5
get_target_property(_cyclonedds_has_shm CycloneDDS::ddsc SHM_SUPPORT_IS_AVAILABLE)
if(_cyclonedds_has_shm)
  find_package(iceoryx_binding_c REQUIRED)
  target_link_libraries(... iceoryx_binding_c)
endif()

启用后定义 DDS_HAS_SHM，开放 loan API 与 iceoryx chunk 路径。

3. 全局状态与 Domain 管理

3.1 全局单例 Cdds（gcdds()）

1
2
3
4
5
6
7
8
9
struct Cdds {
  std::mutex lock;
  std::mutex domains_lock;
  std::map<dds_domainid_t, CddsDomain> domains;

  /* 空 waitset 时 attach 永不触发的 guard，避免 Cyclone 无实体时立即返回 */
  dds_entity_t gc_for_empty_waitset;
  std::unordered_set<CddsWaitset *> waitsets;
};
  • CddsDomain：每个 ROS domain id 一份，含 localhost_only、refcount、domain_handle。
  • localhost-only：创建 domain 时在 CYCLONEDDS_URI 前注入 "localhost" 网络接口配置。
  • 同一 domain 内所有 node 的 localhost_only 必须一致，否则创建失败。

3.2 rmw_context_impl_t

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
struct rmw_context_impl_s {
  rmw_dds_common::Context common;
  dds_domainid_t domain_id;
  dds_entity_t ppant;           // DomainParticipant
  rmw_gid_t ppant_gid;

  dds_entity_t rd_participant;  // DCPS builtin readers
  dds_entity_t rd_subscription;
  dds_entity_t rd_publication;

  dds_entity_t dds_pub;         // 共用 DDS Publisher
  dds_entity_t dds_sub;         // 共用 DDS Subscriber

  size_t node_count;
  std::mutex initialization_mutex;
  bool is_shutdown;
  uint32_t client_service_id;
};

3.3 生命周期时序

与 Connext RMW 类似，Participant 在首个 node 创建时初始化：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
rmw_init()
  └── 仅分配 rmw_context_impl_t，拷贝 init_options（不创建 DDS 实体）

rmw_create_node()  [首个 node]
  └── rmw_context_impl_t::init()
        ├── check_create_domain()     # 必要时 dds_create_domain
        ├── dds_create_participant()
        ├── DCPSParticipant/Publication/Subscription readers
        ├── dds_create_publisher/subscriber
        ├── create_publisher/subscription("ros_discovery_info")
        ├── discovery_thread_start()
        └── graph_cache.add_node + publish

rmw_destroy_node()  [最后一个 node]
  └── rmw_context_impl_t::fini() → tear down participant / domain

rmw_shutdown 仅设 is_shutdown=true；rmw_context_fini 要求已 shutdown，释放 impl。

4. 实体内部结构

struct 关键字段 说明
CddsPublisher enth, sertype, pubiid, gid, is_loaning_available DataWriter + sertype 引用
CddsSubscription enth, rdcondh, gid, data_allocator DataReader + ReadCondition
CddsClient / CddsService CddsCS { pub, sub, id } 请求/响应 writer+reader 对
CddsGuardCondition gcondh DDS GuardCondition
CddsWaitset waitseth, attach 缓存向量 支持 reattach 优化
CddsEvent enth, event_type QoS 事件 status condition

rmw_*_t->data 指向上述结构；implementation_identifier = "rmw_cyclonedds_cpp"。

5. Topic 命名

前缀（namespace_prefix.hpp）：

前缀 用途
rt 普通 topic：/foo → rt/foo
rq + Request Service 请求侧
rr + Reply Service 响应侧

make_fqtopic(prefix, name, suffix, qos) 组装 DDS topic 名；avoid_ros_namespace_conventions 时跳过 rt 前缀。

内部 graph topic：ros_discovery_info（ParticipantEntitiesInfo，TRANSIENT_LOCAL）。

demangle.cpp 提供 topic/type 名还原，供 graph API 返回 ROS 语义字符串。

6. 类型系统与序列化

6.1 设计思路

Cyclone RMW 不使用 rosidl_typesupport_fastrtps_*，而是：

  1. 通过 introspection（C 或 C++）获取 MessageMembers / MessageMember 元数据；
  2. TypeSupport2.hpp 构建 StructValueType 类型树；
  3. Serialization.cpp 的 CDRCursor 按 OMG CDR 规则读写；
  4. serdata_rmw 继承 ddsi_serdata，持有 CDR 字节流或 iceoryx chunk 指针。

6.2 sertype_rmw / serdata_rmw

1
2
3
4
5
6
7
8
9
10
11
12
13
struct sertype_rmw : ddsi_sertype {
  CddsTypeSupport type_support;
  bool is_request_header;   // service 请求类型带 header
  std::unique_ptr<const BaseCDRWriter> cdr_writer;
  bool is_fixed;
  std::mutex serialize_lock;
};

class serdata_rmw : public ddsi_serdata {
  size_t m_size;
  std::unique_ptr<byte[]> m_data;  // 前 4 字节：CDR encapsulation
  // SHM: iox_chunk, iox_subscriber 等
};

Service 请求/响应包装：

1
2
typedef struct cdds_request_header { uint64_t guid; int64_t seq; } cdds_request_header_t;
typedef struct cdds_request_wrapper { cdds_request_header_t header; void * data; } cdds_request_wrapper_t;

这是 非标准 basic RPC 变体（header 布局与 Connext extended / Fast-DDS 不同），导致 默认情况下 service 难以跨 rmw 厂商互通（Connext 需 RMW_CONNEXT_CYCLONE_COMPATIBILITY_MODE）。

6.3 Typesupport 选择

get_typesupport() 优先 introspection C，其次 C++：

1
2
get_message_typesupport_handle(type_supports, rosidl_typesupport_introspection_c__identifier)
// fallback: rosidl_typesupport_introspection_cpp::typesupport_identifier

6.4 rmw_serialize / rmw_deserialize

在 rmw_node.cpp 中调用 serdes 与 TypeSupport 模块，格式字符串 "cdr"（rmw_get_serialization_format()）。

7. 发布与订阅数据路径

7.1 常规发布

1
2
3
4
rmw_publish()
  → CddsPublisher::write path
  → 序列化 ros_message → serdata_rmw
  → dds_writecdr(writer, serdata)  或 dds_write

7.2 常规 take

1
2
3
4
rmw_take() / rmw_take_with_info()
  → dds_take / dds_takecdr
  → 反序列化 serdata → ros_message
  → message_info_from_sample_info()

7.3 共享内存与 Loan（DDS_HAS_SHM）

当 Cyclone 启用 iceoryx 且消息为 fixed-size 时：

API 行为
rmw_borrow_loaned_message dds_data_allocator 从 writer loan chunk
rmw_publish_loaned_message dds_writecdr + IOX_CHUNK_CONTAINS_RAW_DATA
rmw_take_loaned_message 直接返回 iox chunk 指针（RAW）或反序列化（SERIALIZED）
rmw_return_loaned_message_from_* fini_and_free_sample 释放 chunk

未编译 SHM 时，上述 API 返回 RMW_RET_UNSUPPORTED。

配置见仓库 shared_memory_support.md：需 CYCLONEDDS_URI 启用 SharedMemory、运行 iox-roudi 等。

7.4 自包含类型检测

is_type_self_contained() 决定是否设置 can_loan_messages / is_loaning_available（无 string/动态数组等）。

8. Service / Client

8.1 创建逻辑（create_client_service）

  • Service：sub 收 rq/...Request，pub 发 rr/...Reply
  • Client：pub 发 rq/...Request，sub 收 rr/...Reply
  • 各端创建带 is_request_header=true 的 sertype_rmw
  • Writer/Reader QoS user_data 写入 serviceid= / clientid= 十六进制串，用于配对

8.2 请求-响应路径

1
2
3
4
5
6
7
8
9
10
rmw_send_request()
  → 填充 cdds_request_header { guid, seq }
  → dds_write(client.pub, &wrap)

rmw_take_request() / rmw_take_response()
  → dds_take → 解析 wrap.header
  → 重建 rmw_request_id_t（writer_guid + publication_handle + seq）

rmw_send_response()
  → 同 send_request，使用 service 侧 publisher

rmw_service_server_is_available 通过 matched endpoints + user_data 检查 client reader 是否就绪。

9. Graph 与 Discovery

9.1 初始化（rmw_context_impl_s::init 内）

与 Connext/FastRTPS 新实现一致：

  1. 创建 ros_discovery_info pub/sub（ParticipantEntitiesInfo）
  2. 创建 graph guard_condition
  3. 启动 discovery 线程

9.2 Discovery 线程（discovery_thread）

WaitSet 监听：

  • rd_participant → handle_DCPSParticipant
  • rd_publication → handle_DCPSPublication
  • rd_subscription → handle_DCPSSubscription
  • common.sub → handle_ParticipantEntitiesInfo
  • 退出 guard condition

更新 rmw_dds_common::GraphCache，必要时 rmw_publish 广播本地变更。

9.3 Node 创建时的 graph 更新

1
2
3
std::lock_guard<std::mutex> guard(common->node_update_mutex);
auto participant_msg = common->graph_cache.add_node(common->gid, name, namespace_);
rmw_publish(common->pub, &participant_msg, ...);

锁保证 update + publish 原子性，避免交错消息覆盖。

9.4 Graph API

rmw_get_node_names、rmw_get_topic_names_and_types、rmw_get_publishers_info_by_topic 等均在 rmw_node.cpp 后半部分，读取 graph_cache 并 demangle。

10. Wait Set 与 rmw_wait

10.1 空 WaitSet 问题

Cyclone 在 waitset 无任何 condition 时会立即返回。RMW 创建全局 dummy guard condition（永不 trigger）并 attach 到每个 waitset，使空 wait 能正确阻塞。

10.2 rmw_wait 流程

  1. 检查 ws->inuse（禁止同一 waitset 并发 wait）
  2. 若 subscription/guard/service/client/event 数组变化 → waitset_detach 后重新 attach
  3. Subscription/Service/Client 使用 ReadCondition（rdcondh）
  4. dds_waitset_wait + timeout
  5. 未就绪条目在 rcl 传入的数组中置 NULL

Event 通过 gather_event_entities 收集 status condition 并 attach。

11. QoS

11.1 ROS → DDS

create_readwrite_qos() / rmw_duration_to_dds() 映射：

  • history / depth
  • reliability（RELIABLE / BEST_EFFORT）
  • durability（VOLATILE / TRANSIENT_LOCAL）
  • deadline / lifespan / liveliness

11.2 DDS → ROS

dds_qos_to_rmw_qos()、get_readwrite_qos() 用于 rmw_publisher_get_actual_qos 等。

11.3 兼容性检查

1
2
3
rmw_qos_profile_check_compatible(...) {
  return rmw_dds_common::qos_profile_check_compatible(...);
}

11.4 Cyclone 配置

运行时行为大量受 CYCLONEDDS_URI 影响（发现 peer、追踪日志、SHM、网卡选择等），见 README 与 Cyclone 手册。

12. 事件与回调

12.1 支持的事件

mask_map 映射 rmw 事件到 DDS status：

  • RMW_EVENT_*_QOS_INCOMPATIBLE
  • RMW_EVENT_*_DEADLINE_MISSED
  • RMW_EVENT_LIVELINESS_*
  • RMW_EVENT_MESSAGE_LOST

12.2 两种通知机制

  1. Wait 路径：rmw_take_event 读取 status
  2. Callback 路径：rmw_*_set_on_new_message_callback、rmw_event_set_callback 使用 DDS listener + user_callback_data_t 计数

13. 安全（SROS2）

1
2
3
#if DDS_HAS_SECURITY && DDS_HAS_PROPERTY_LIST_QOS
#define RMW_SUPPORT_SECURITY 1
#endif

configure_qos_for_security() 在创建 participant 时设置 security 相关 QoS property；enforce 模式下失败则 init 中止。依赖 Cyclone 编译时启用 security。

Participant user_data 写入 enclave=...; 供 graph 使用。

14. 日志与追踪

  • rmw_set_log_severity → dds_set_log_mask(DDS_LC_*) 映射 rcutils 级别
  • tracetools：关键路径插入 LTTng 追踪点（与 ROS 2 全局 tracing 集成）
  • get_rmw_cyclonedds_output_filter.cmake：注册控制台输出过滤正则（网卡选择警告）

15. 未实现或受限功能

API / 能力 状态
rmw_get_serialized_message_size RMW_RET_UNSUPPORTED
rmw_publisher/subscription_get_network_flow_endpoints 未实现
rmw_subscription_set/get_content_filter 未实现
rmw_*_allocation（pub/sub 预分配） 未实现 / 报错
Loan API 仅 DDS_HAS_SHM + fixed-size 类型
Strict unique network flow endpoints 创建 subscription 时拒绝

16. 与 rmw_dds_common 的协作

组件 用法
rmw_dds_common::Context rmw_context_impl_t::common
GraphCache 节点/endpoint/topic 图
ParticipantEntitiesInfo discovery topic 消息
qos_profile_check_compatible QoS 兼容 API
security.hpp participant QoS 安全配置

Cyclone 实现是 rmw_dds_common 的主要消费者之一，graph 协议与 FastRTPS/Cyclone/Connext 新版对齐。

17. 互操作说明

场景 默认互操作
Pub/Sub（CDR + rt/ topic） 与 Fast-DDS、Connext 等 通常可互通
Service/Client 仅同 RMW 或专门兼容模式（header 布局自定义）
Connext ↔ Cyclone service Connext 侧 RMW_CONNEXT_CYCLONE_COMPATIBILITY_MODE=y
SHM 零拷贝 仅本机 + 同 RMW + fixed 类型 + QoS 满足 iceoryx 限制

18. 调试建议

  1. 确认 RMW：ros2 doctor --report 或检查 RMW_IMPLEMENTATION=rmw_cyclonedds_cpp。
  2. Discovery 问题：配置 CYCLONEDDS_URI 静态 peer；用 Cyclone ddsperf sanity 排除网络层。
  3. 大消息延迟：增大 Linux net.core.rmem_max（README 推荐）。
  4. Graph 空：检查 ros_discovery_info 与 DCPS reader 是否创建成功；看 discovery 线程是否运行。
  5. SHM：确认 RouDi 运行、QoS 符合 shared_memory_support.md 限制。
  6. Cyclone 跟踪：CYCLONEDDS_URI='<Tracing><Verbosity>trace</>...'。

19. 推荐阅读顺序

  1. rmw_node.cpp 前 450 行 — 全局 struct、identifier、日志
  2. rmw_context_impl_s::init（~1160 行） — Participant + discovery 初始化
  3. create_publisher / create_cdds_publisher — topic + sertype 创建
  4. serdata.hpp + serdata.cpp — DDSI 样本与 SHM
  5. TypeSupport2.hpp + Serialization.cpp — CDR 序列化核心
  6. discovery_thread + graph 回调 — 图更新
  7. rmw_wait（~4037 行） — waitset 语义
  8. Service 段（~4550 行） — cdds_request_wrapper
  9. Cyclone 上游 — dds_create_*、ddsi_serdata API
  10. shared_memory_support.md — 零拷贝配置

20. 小结

rmw_cyclonedds 是 单包、单主文件 型 RMW 实现，特点鲜明：

  • 对上：完整覆盖 Humble RMW 契约（loan/network flow 等部分可选）；
  • 对下：深度集成 Cyclone DDSI serdata 插件，序列化走 自研 introspection CDR；
  • 横向：通过 rmw_dds_common 统一 ROS graph；
  • 性能：可选 iceoryx 共享内存实现 fixed 类型零拷贝；
  • 注意：Service RPC 使用 自定义 inline header，跨厂商调用需额外兼容配置。

Humble 默认 RMW 仍为 Fast-DDS，选用 Cyclone 需显式 export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp 并在构建时确保 cyclonedds 包可用。

文章互动

阅读 --

留言

0 条留言

正在加载留言…