08 示例与性能工具

本文重点：走读 iceoryx_examples 中最有代表性的示例（每个示例的演示点 + 关键 API），iceperf 的测试方法与结果解读，tools/ 排障工具（iox-introspection-client），以及常用排障 checklist。
源码锚点：iceoryx_examples/*、tools/introspection/

源码根目录：/home/cp/work2/ros2Learn/ros2_humble/src/eclipse-iceoryx/iceoryx（v2.0.6）

注意：iceoryx_examples/ 目录带有 COLCON_IGNORE，ROS 2 Humble 工作区构建时不编译这些示例。想跑示例需单独构建 iceoryx（tools/iceoryx_build_test.sh 或 CMake -DEXAMPLES=ON）。所有多进程示例都要求先启动 iox-roudi。

1. 示例总览

其余目录（*_in_c 为对应 C API 版本；icecrystal 演示内省、iceoptions 演示 QoS 选项、complexdata 演示 iceoryx 容器类型、ice_access_control 演示段权限、icedocker 演示容器部署、iceensemble 多发布者）从略。

2. icehello：最小可用样例

发布侧三步：init runtime → 建 publisher → loan/publish。CaPro 三元组 {"Radar", "FrontLeft", "Object"} 即”topic”：

1
2
3
4
5
6
7
8
9
10
11
12

auto loanResult = publisher.loan();
//! [loan]
//! [publish]
if (!loanResult.has_error())
{
    auto& sample = loanResult.value();
    // Sample can be held until ready to publish
    sample->x = ct;
    sample->y = ct;
    sample->z = ct;
    sample.publish();
}

订阅侧轮询 take()，返回 expected<Sample, ChunkReceiveResult>，NO_CHUNK_AVAILABLE 是正常空态而非错误：

1
2
3
4
5
6
7

//! [receive]
auto takeResult = subscriber.take();
if (!takeResult.has_error())
{
    std::cout << APP_NAME << " got value: " << takeResult.value()->x << std::endl;
}
//! [receive]

要点：sample 是 RAII 智能指针（iox::popo::Sample），未 publish 即析构会自动归还 chunk，不泄漏。运行：iox-roudi + 两个进程各跑一个可执行文件。

3. icedelivery：四种发布姿势与 untyped API

iox_publisher.cpp 依次演示 typed API 的四种用法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

//! [API Usage #1]
//  * Retrieve a typed sample from shared memory.
//  * Sample can be held until ready to publish.
//  * Data is default constructed during loan
publisher.loan()
    .and_then([&](auto& sample) {
        sample->x = sampleValue1;
        sample->y = sampleValue1;
        sample->z = sampleValue1;
        sample.publish();
    })
    .or_else([](auto& error) {
        // Do something with error
        std::cerr << "Unable to loan sample, error: " << error << std::endl;
    });
//! [API Usage #1]

iox_publisher_untyped.cpp / iox_subscriber_untyped.cpp 则是 UntypedPublisher::loan(payloadSize) 返回 void*、UntypedSubscriber::take() 后手动 release——正是 06/07 篇中 CycloneDDS 与 dds-gateway 使用的形态。

4. waitset：阻塞等待

ice_waitset_basic.cpp 演示”attach 订阅者状态 + 阻塞 wait”的标准循环，并示范了信号安全退出（signal handler 里 markForDestruction 唤醒阻塞的 wait）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

while (keepRunning)
{
    // We block and wait for samples to arrive.
    auto notificationVector = waitset->wait();

    for (auto& notification : notificationVector)
    {
        // ...
        if (notification->doesOriginateFrom(&subscriber))
        {
            // Consume a sample
            subscriber.take()
                .and_then([](auto& sample) { std::cout << " got value: " << sample->counter << std::endl; })
                .or_else([](auto& reason) {
                    std::cout << "got no data, return code: " << static_cast<uint64_t>(reason) << std::endl;
                });
            // We could consume all samples but do not need to.
            // If there is more than one sample we will wake up again since the state of the subscriber is still
            // iox::popo::SubscriberState::HAS_DATA in this case.
        }
    }
}

注意 attach 的是 state（SubscriberState::HAS_DATA，电平触发：只要还有数据 wait 立即返回）而非 event（DATA_RECEIVED，边沿触发：只在新数据到达时醒）。同目录其他文件依次演示：gateway（一个回调处理 N 个订阅者）、grouping（用 id 分组）、individual（每个附着对象单独处理）、timer_driven_execution（用 user-trigger 实现定时器）、trigger（自定义类实现可附着的 trigger 接口）。

5. callbacks：Listener 事件驱动

WaitSet 需要用户线程自己转循环；Listener 则起后台线程，事件到达即调用回调（这也是 CycloneDDS shm_monitor 的用法）。示例用两个订阅者 + 一个 4 秒心跳 UserTrigger：

1
2
3
4
5
6
7
8
9
10
11
12

listener
    .attachEvent(subscriberLeft,
                 iox::popo::SubscriberEvent::DATA_RECEIVED,
                 iox::popo::createNotificationCallback(onSampleReceivedCallback))
    .or_else([](auto) {
        std::cerr << "unable to attach subscriberLeft" << std::endl;
        std::exit(EXIT_FAILURE);
    });
listener
    .attachEvent(subscriberRight,
                 iox::popo::SubscriberEvent::DATA_RECEIVED,
                 iox::popo::createNotificationCallback(onSampleReceivedCallback))

回调签名是普通函数指针 void(Subscriber<T>*)（源码注释强调：Listener 不持有回调所有权、且不支持捕获 lambda）；ice_callbacks_listener_as_class_member.cpp 进一步演示用 _with_context_data 变体把 this 传进静态成员回调。

6. request_response：客户端/服务端

v2.0 新增的请求-响应模式。client loan 请求、设置 sequenceId、send()，随后轮询 take() 响应并校验序号：

1
2
3
4
5
6
7
8
9
10
11
12
13

//! [send request]
client.loan()
    .and_then([&](auto& request) {
        request.getRequestHeader().setSequenceId(requestSequenceId);
        expectedResponseSequenceId = requestSequenceId;
        requestSequenceId += 1;
        request->augend = fibonacciLast;
        request->addend = fibonacciCurrent;
        std::cout << APP_NAME << " Send Request: " << fibonacciLast << " + " << fibonacciCurrent << std::endl;
        request.send().or_else(
            [&](auto& error) { std::cout << "Could not send Request! Error: " << error << std::endl; });
    })
    .or_else([](auto& error) { std::cout << "Could not allocate Request! Error: " << error << std::endl; });

server 侧（server_cxx_basic.cpp）对称：server.take() 拿请求 → server.loan(request) 借响应（自动关联该请求）→ 填结果 → send()。client_cxx_waitset.cpp、server_cxx_listener.cpp 分别演示与 WaitSet/Listener 组合。底层机制是一对共享内存队列 + RequestHeader/ResponseHeader（iceoryx_posh/popo/rpc_header.hpp）。

7. user_header：自定义头（时间戳）

发布者模板第二参数指定 user-header 类型，loan 后通过 getUserHeader() 写元数据——与 payload 分离、不侵入消息类型：

1
2
3
4
5
6
7
8
9
10
11

//! [loan sample]
publisher.loan(Data{fibonacciCurrent})
    .and_then([&](auto& sample) {
        //! [loan was successful]
        sample.getUserHeader().publisherTimestamp = timestamp;
        sample.publish();

        std::cout << APP_NAME << " sent data: " << fibonacciCurrent << " with timestamp " << timestamp << "ms"
                  << std::endl;
        //! [loan was successful]
    })

同目录提供 untyped C++ 与 C 版本；C 版本 publisher_c_api.c 用的正是 iox_pub_loan_aligned_chunk_with_user_header + iox_chunk_header_to_user_header——与 CycloneDDS 挂 iceoryx_header_t 的做法（07 篇 §5.2/5.3）完全同构，是理解 SHM 集成的最佳热身示例。

8. singleprocess：内嵌 RouDi

演示不起独立 iox-roudi，把 RouDi 组件直接实例化在自己进程里，publisher/subscriber 以线程形式通信（也是集成测试的常用手法）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

//! [roudi config]
iox::RouDiConfig_t defaultRouDiConfig = iox::RouDiConfig_t().setDefaults();
iox::roudi::IceOryxRouDiComponents roudiComponents(defaultRouDiConfig);
//! [roudi config]

//! [roudi]
constexpr bool TERMINATE_APP_IN_ROUDI_DTOR_FLAG = false;
iox::roudi::RouDi roudi(
    roudiComponents.rouDiMemoryManager,
    roudiComponents.portManager,
    iox::roudi::RouDi::RoudiStartupParameters{iox::roudi::MonitoringMode::OFF, TERMINATE_APP_IN_ROUDI_DTOR_FLAG});
//! [roudi]

// create a single process runtime for inter thread communication
//! [runtime]
iox::runtime::PoshRuntimeSingleProcess runtime("singleProcessDemo");

关键差异：用 PoshRuntimeSingleProcess 替代 PoshRuntime::initRuntime()（后者走 IPC 通道向外部 RouDi 注册）。限制：单进程模式下没有跨进程通信，仅线程间。

9. icediscovery：服务发现

iox_find_service.cpp 演示 ServiceDiscovery 的同步查询，支持通配符（iox::capro::Wildcard）：

1
2
3
4
5
6
7

//! [search for unique service]
serviceDiscovery.findService(iox::capro::IdString_t{"Radar"},
                             iox::capro::IdString_t{"FrontLeft"},
                             iox::capro::IdString_t{"Image"},
                             printSearchResult,
                             iox::popo::MessagingPattern::PUB_SUB);
//! [search for unique service]

• iox_offer_service.cpp：建 publisher 即自动 offer 服务；
• iox_wait_for_service.cpp：把 ServiceDiscovery attach 到 WaitSet（ServiceDiscoveryEvent::SERVICE_REGISTRY_CHANGED），阻塞等待特定服务上线；
• iox_discovery_monitor.cpp：attach 到 Listener，服务注册表变化时回调——实现”发现即回调”的监控器。

10. iceperf：延迟基准

10.1 测试方法

leader/follower 两个进程做 ping-pong 往返：leader 发一个 payload，follower 原样回发，重复 N 次（默认 10000，可用 -n 调整）。单向延迟 = 总耗时 / (N × 2)：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

iox::units::Duration IcePerfBase::latencyPerfTestLeader(const uint64_t numRoundTrips) noexcept
{
    auto start = std::chrono::steady_clock::now();

    // run the performance test
    for (auto i = 0U; i < numRoundTrips; ++i)
    {
        auto perfTopic = receivePerfTopic();
        sendPerfTopic(perfTopic.payloadSize, RunFlag::RUN);
    }

    auto finish = std::chrono::steady_clock::now();

    constexpr uint64_t TRANSMISSIONS_PER_ROUNDTRIP{2U};
    auto duration = std::chrono::duration_cast<std::chrono::nanoseconds>(finish - start);
    auto latencyInNanoSeconds =
        (static_cast<uint64_t>(duration.count()) / (numRoundTrips * TRANSMISSIONS_PER_ROUNDTRIP));
    return iox::units::Duration::fromNanoseconds(latencyInNanoSeconds);
}

payload 从 1KB 扫到 4MB，横向对比四种 IPC 技术（同一套 IcePerfBase 抽象的四个实现）：

1
2

std::vector<std::tuple<uint32_t, iox::units::Duration>> latencyMeasurements;
const std::vector<uint32_t> payloadSizesInKB{1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096};

iceoryx 侧的收发即 untyped 零拷贝（无 memcpy payload，仅写头部字段）：

1
2
3
4
5
6
7
8
9
10
11

void Iceoryx::sendPerfTopic(const uint32_t payloadSizeInBytes, const RunFlag runFlag) noexcept
{
    m_publisher.loan(payloadSizeInBytes).and_then([&](auto& userPayload) {
        auto sendSample = static_cast<PerfTopic*>(userPayload);
        sendSample->payloadSize = payloadSizeInBytes;
        sendSample->runFlag = runFlag;
        sendSample->subPackets = 1;

        m_publisher.publish(userPayload);
    });
}

10.2 运行与解读

1
2
3

iox-roudi -c iceoryx_examples/iceperf/roudi_config.toml   # 需要大 mempool（最大 payload 4MB）
./iceperf-bench-leader        # 终端 2
./iceperf-bench-follower     # 终端 3

结果为 Markdown 表格（| Payload Size [kB] | Average Latency [µs] |）。含义：MQ/UDS 的延迟随 payload 线性增长（两次内核拷贝），iceoryx 的延迟基本与 payload 无关（常数级，只传指针）——这正是零拷贝的核心卖点，payload 越大优势越显著。leader 还会先通过 iceoryx topic {"IcePerf","Settings","Generic"} 把测试参数发给 follower（iceperf_leader.cpp:104-113），因此即便只测 MQ/UDS 也需要 RouDi 在跑。

11. tools：排障工具

11.1 iox-introspection-client

tools/introspection/ 构建出 iox-introspection-client，本质是一个订阅 RouDi 内省 topic（Introspection 服务，见 iceoryx_posh/roudi/introspection_types.hpp）的 ncurses 客户端：

1
2
3
4
5
6
7
8
9
10
11

"  introspection [OPTIONS] [SUBSCRIPTION]\n"
"  introspection --help\n"
"  introspection --version\n"
...
"  -t, --time <ms>   Update period (in milliseconds) for the display of introspection data\n"
...
"  Select which introspection data you would like to receive.\n"
"  --all             Subscribe to all available introspection data.\n"
"  --mempool         Subscribe to mempool introspection data.\n"
"  --port            Subscribe to port introspection data.\n"
"  --process         Subscribe to process introspection data.\n"

三类视图对应三类排障问题：

用法示例：iox-introspection-client --all -t 1000。

11.2 其他工具

• tools/iceoryx_build_test.sh：一键构建 + 测试脚本（build-all 包含示例与内省）；
• iceoryx_posh/roudi 提供的 iox-roudi 本身支持 -c 指定 TOML 配置（段/mempool 布局）、-l 日志级别、-m 监控模式，是排障的第一现场。

12. 排障 checklist

13. 相关文档

• C API 与网关：06-C绑定与DDS网关.md
• CycloneDDS/ROS 2 集成：07-与CycloneDDS及ROS2集成.md
• 官方示例索引：iceoryx_examples/README.md

下一篇

本篇是系列最后一篇，返回 README.md。

Eclipse iceoryx 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/eclipse-iceoryx/iceoryx
版本：2.0.6（iceoryx_posh/package.xml），语言 C++17（含 C binding），许可证 Apache 2.0。

iceoryx 是面向**进程间通信（IPC）**的中间件，核心卖点是 真零拷贝（true zero-copy）：Publisher 与 Subscriber 通过 POSIX 共享内存直接传递数据指针，payload 不经 socket 拷贝。在 ROS 2 Humble 中，它主要作为 CycloneDDS 共享内存传输层的后端（ENABLE_SHM + iceoryx_binding_c），也可通过 rmw_iceoryx 作为独立 RMW 使用。

1. 总体架构

与 CycloneDDS 的“协议栈 + API”不同，iceoryx 是纯 IPC 层，不负责 DDS/ROS 的发现与 QoS，只负责内存池 + 端口 + 数据分发。

源码规模：iceoryx_posh 约 91 个 .cpp，iceoryx_hoofs 约 36 个 .cpp，合计 1300+ 文件。

2. 核心概念

2.1 RouDi（Routing and Discovery）

RouDi 是 iceoryx 的中心守护进程，必须在所有应用之前启动（ROS 2 中通常由 launch 或 iox-roudi 启动）。

职责（见 internal/roudi/roudi.hpp）：

• 创建并管理 共享内存段（management + payload）
• 处理 Runtime 的 IPC 注册（进程注册、版本校验）
• 通过 PortManager 分配 Publisher/Subscriber/Client/Server 端口
• 维护 ServiceRegistry（服务发现）
• 进程异常退出时 清理 chunk 与端口
• 提供 Introspection（mempool/port/process 监控）

应用侧通过 PoshRuntime::initRuntime(name) 与 RouDi 建立 Unix Domain Socket 通道，请求创建端口。

2.2 ServiceDescription（三元组寻址）

iceoryx 用 Service / Instance / Event 三元组标识通信端点（类似 AUTOSAR ara::com）：

1
2

/// @brief class for the identification of a communication event including information on the service, the service
/// instance and the event id.

CycloneDDS SHM 集成时，会将 DDS topic 名映射为这三元组。接口类型枚举还包括 DDS、ROS1 等，表明其设计面向多协议网关。

2.3 Chunk — 零拷贝的数据单元

Chunk 是共享内存中的传输胶囊，布局见 doc/design/chunk_header.md：

1
2
3

+===================+===============+====================+============+
|   ChunkHeader     |  User-Header  |    User-Payload    |  Padding   |
+===================+===============+====================+============+

ChunkHeader（mepoo/chunk_header.hpp）包含：

• chunkSize、chunkHeaderVersion
• originId（Publisher 唯一 ID）、sequenceNumber
• userHeaderSize / userPayloadSize / userPayloadAlignment
• 通过 back-offset 从 user-payload 反查 ChunkHeader

设计约束：

• 共享内存映射到各进程不同虚拟地址 → 禁止裸指针，必须用相对/可重定位指针
• 支持 record & replay 的版本号与 origin 追踪

3. iceoryx_hoofs — 基础库

hoofs = Healthy Overly Optimistic Foundation Stuff（项目自嘲式命名），提供无 STL 依赖（或最小依赖）的基础能力。

3.1 目录结构

3.2 相对指针（共享内存关键）

doc/design/relocatable_pointer.md 说明：

• 各进程将同一段 SHM 映射到不同基址
• relocatable_ptr：指针与 pointee 在同一段内，存偏移量
• RelativePointer：跨段引用，通过全局 segment id 解析

RouDi 的 Port 数据结构、Chunk 队列、Subscriber 列表等都建立在相对指针之上，这是 iceoryx 能在多进程间安全传递“指针”的基础。

3.3 无锁队列

doc/design/lockfree_queue.md + concurrent/lockfree_queue.hpp：Subscriber 侧 chunk 队列采用无锁设计，降低 pub/sub 热路径上的锁竞争。

4. iceoryx_posh — 核心中间件

posh = Posix Shared Memory。源码按功能分目录：

1
2
3
4
5
6
7
8

iceoryx_posh/source/
├── capro/      # Capabilities & Protocol — ServiceDescription、发现消息
├── mepoo/      # Memory Pool — ChunkHeader、MemoryManager、SharedChunk
├── popo/       # Posix Objects — Publisher/Subscriber/Client/Server/WaitSet
├── roudi/      # RouDi 守护进程实现
├── runtime/    # PoshRuntime — 应用与 RouDi 的 IPC 接口
├── gateway/    # 网关相关
└── version/    # 版本与兼容性检查

4.1 MePoo（Memory Pool）

默认 mempool 通过 MePooConfig::setDefaults() 设置；RouDi 配置文件（TOML）可覆盖。chunk 不足时 loan() 返回 AllocationError::RUNNING_OUT_OF_CHUNKS。

4.2 popo — 面向用户的 API

popo = Posix Objects（仿 COM 命名）。

Publisher 典型流程（publisher_impl.hpp）：

1
2
3
4
5

// 1. 从 mempool loan 一块共享内存
auto sample = publisher.loan();
// 2. 写入 sample->payload
// 3. 发布（ChunkDistributor 推送到所有 Subscriber 队列）
publisher.publish(std::move(sample));

publishCopyOf() 是带拷贝的便捷路径，不是零拷贝。

4.3 Building Blocks — 内部构建块

Pub/Sub 内部分层（自底向上）：

1
2
3
4
5
6
7
8
9

MemoryManager
    ↓
ChunkSender  ←→  ChunkReceiver
    ↓                  ↓
ChunkDistributor   ChunkQueuePopper
    ↓                  ↓
PublisherPort      SubscriberPort
    ↓                  ↓
Publisher          Subscriber

Publisher 与 Subscriber 匹配时，RouDi 的 PortManager 将 Subscriber 的 ChunkQueue 注册到 Publisher 的 ChunkDistributor（port_manager.hpp 的 acquirePublisherPortData / acquireSubscriberPortData）。

4.4 Port 架构（User / RouDi 分离）

每个 Port 有三层：

这种分离保证：RouDi 可直接操作 SHM 中的 port 元数据，而应用通过 User 层访问，进程崩溃时 RouDi 仍能 cleanup。

4.5 PoshRuntime

1
2
3
4
5
6
7
8
9

/// @brief The runtime that is needed for each application to communicate with the RouDi daemon
class PoshRuntime
{
  public:
    static PoshRuntime& initRuntime(const RuntimeName_t& name) noexcept;
    virtual PublisherPortUserType::MemberType_t*
    getMiddlewarePublisher(const capro::ServiceDescription& service, ...) noexcept = 0;
    virtual SubscriberPortUserType::MemberType_t*
    getMiddlewareSubscriber(const capro::ServiceDescription& service, ...) noexcept = 0;

• 单例，initRuntime() 注册进程名（须唯一）
• 通过 IPC 消息向 RouDi 请求创建 Publisher/Subscriber/Client/Server
• 也支持 SingleProcess 模式（测试用，无需 RouDi）

4.6 Request/Response（ROS 2 Service 基础）

doc/design/request_response_communication.md：

• Client：ChunkSender 发 request + ChunkReceiver 收 response
• Server：ChunkReceiver 收 request + ChunkSender 发 response
• Request/Response Header 含 sequence ID，支持异步 RPC
• 同一 ServiceDescription 只允许一个 Server

这与 ROS 2 rclcpp::Client / rclcpp::Service 的语义对齐。

5. iceoryx_binding_c — C 绑定

路径：iceoryx_binding_c/include/iceoryx_binding_c/

C binding 是 CycloneDDS SHM 集成的直接依赖——CycloneDDS 为 C 项目，不能直接用 C++ Publisher<T>。

6. 与 CycloneDDS / ROS 2 的集成

6.1 CycloneDDS SHM 路径

CycloneDDS 编译选项 ENABLE_SHM=AUTO 会查找 iceoryx_binding_c，启用 DDS_HAS_SHM。

关键桥接代码：cyclonedds/src/core/ddsi/src/ddsi_shm_transport.c

1
2
3
4
5
6
7
8
9
10
11

void *shm_create_chunk(iox_pub_t iox_pub, size_t size) {
  // ...
  iox_pub_loan_aligned_chunk_with_user_header(
      iox_pub, &iox_chunk, (uint32_t)size,
      IOX_C_CHUNK_DEFAULT_USER_PAYLOAD_ALIGNMENT,
      sizeof(iceoryx_header_t), 8);
  // ...
  ice_hdr->data_size = (uint32_t)size;
  ice_hdr->shm_data_state = IOX_CHUNK_UNINITIALIZED;
  return iox_chunk;
}

数据流（同机 ROS 2 + CycloneDDS + SHM）：

1
2
3
4
5
6
7
8
9
10

rclcpp::Publisher::publish(msg)
  → rmw_cyclonedds_cpp::dds_write
    → dds_write / dds_writecdr
      → [SHM 启用且订阅者在同机]
          → iox_pub_loan_chunk (共享内存)
          → 序列化 payload 到 chunk（仍有一次写入 SHM）
          → iox_pub_publish_chunk
          → Subscriber 侧 iox_sub_take_chunk
          → dds_read 直接读 SHM 指针（零拷贝取数据）
      → [否则] UDP/TCP RTPS 网络路径

注意：

• 跨进程时 payload 仍要写入共享内存一次（不是完全无 touch），但 subscriber 侧不再拷贝到用户 buffer（loan 模式下）
• 网络 RTPS 只传递 chunk 指针/通知（iceoryx 内部机制），不传 payload 本体
• 需先启动 RouDi，且 Publisher/Subscriber 在同一 iceoryx domain

6.2 依赖关系（package.xml）

1
2
3
4

<!-- cyclonedds/package.xml -->
<depend>iceoryx_binding_c</depend>
<depend>iceoryx_posh</depend>
<depend>iceoryx_hoofs</depend>

6.3 iceoryx_dds 网关（可选）

iceoryx_dds/ 提供 iceoryx ↔ Cyclone DDS 双向网关：

• gateway/iox_to_dds.hpp — iceoryx → DDS
• gateway/dds_to_iox.hpp — DDS → iceoryx
• 用于跨网络或异构系统桥接，Humble 默认 ROS 工作流不必须。

7. 配置与部署

7.1 RouDi 配置

• TOML 配置文件（roudi_config.toml）指定 mempool 大小、segment 数量
• 环境变量：IOX_ROUDI_CONFIG_FILE
• 默认 mempool 档位覆盖常见消息大小（128B ~ 4MB 等）

7.2 运行顺序

1
2
3
4
5
6
7

# 1. 启动 RouDi（必须最先）
iox-roudi

# 2. 启动 ROS 2 节点（CycloneDDS SHM 模式）
export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp
# cyclonedds.xml 中启用 SharedMemory
ros2 run ...

7.3 平台支持

8. 事件驱动 API

CycloneDDS SHM 路径中，subscriber 收到 iceoryx notification 后才会去 take chunk。

9. 错误处理与健壮性

• ConsumerTooSlowPolicy：Subscriber 队列满时，Publisher 可 BLOCK 或 DISCARD_OLDEST_DATA
• 进程崩溃：RouDi 检测并 cleanup 泄漏的 chunk（UsedChunkList）
• 版本兼容：Runtime 注册时校验 VersionInfo（MAJOR/MINOR/PATCH）
• expected<T, E>：Hoofs 提供的 Rust 风格错误返回，贯穿 loan/publish API

10. 与 CycloneDDS 的职责对比

11. 推荐阅读顺序

目标：理解 ROS 2 大消息零拷贝

1. doc/design/chunk_header.md — Chunk 内存布局
2. doc/design/relocatable_pointer.md — 为何不能用裸指针
3. iceoryx_posh/include/iceoryx_posh/popo/publisher.hpp → publisher_impl.hpp
4. internal/popo/building_blocks/chunk_sender.hpp → chunk_distributor.hpp
5. internal/roudi/port_manager.hpp — 端口如何匹配
6. iceoryx_binding_c/include/iceoryx_binding_c/publisher.h — C API
7. cyclonedds/src/core/ddsi/src/ddsi_shm_transport.c — 与 DDS 的接缝
8. 示例：iceoryx_examples/icedelivery/、iceoryx_examples/singleprocess/

目标：调试 SHM 问题

1. 确认 RouDi 运行：iox-introspection-client 或 iceoryx_introspection
2. 检查 mempool 耗尽、chunk 泄漏
3. 查看 CycloneDDS SharedMemory 配置段

12. 设计特点小结

如果你希望，我可以：

1. 把本文写入 ros2doc/cyclonedds/ 同级目录 ros2doc/iceoryx/Eclipse iceoryx 源码详细分析.md
2. 继续深入 CycloneDDS ↔ iceoryx 的 topic 映射规则 或 RouDi 启动与 mempool 配置 的源码级追踪

Eclipse iceoryx 源码分析文档

源码：ros2_humble/src/eclipse-iceoryx/iceoryx · v2.0.6
进程间零拷贝共享内存中间件 · ROS 2 Humble 中作为 CycloneDDS 的 SHM 传输后端

推荐阅读顺序

第一步：整体框架（必读）

第二步：集成与实践

总览

模块 → 源码路径

核心概念速记

快速上手（ROS 2 零拷贝）

1
2
3
4
5
6
7
8
9

# 1. 启动 RouDi
iox-roudi

# 2. 启用 CycloneDDS 共享内存
export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp
export CYCLONEDDS_URI='<CycloneDDS><Domain><SharedMemory><Enable>true</Enable></SharedMemory></Domain></CycloneDDS>'

# 3. 观察内存池占用
iox-introspection-client --mempool

细节与限制（QoS 约束、POD 类型要求、mempool 配置）见 07-与CycloneDDS及ROS2集成.md；
排障从 08-示例与性能工具.md 的 checklist 开始。

对照阅读：../cyclonedds/README.md（网络路径） vs 本系列（同机共享内存路径）。

osrf_pycommon 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/osrf/osrf_pycommon
版本：2.1.7（package.xml），构建类型 ament_python，Python ≥3.5，许可证 Apache 2.0。

osrf_pycommon 是 OSRF（Open Source Robotics Foundation）维护的 Python 通用工具库，体量很小（约 47 个文件、4 个子模块），不依赖 ROS 运行时。在 ROS 2 Humble 工作区中，它主要被 launch / launch_ros / launch_testing 依赖，承担子进程异步执行、终端颜色处理和 CLI 扩展模式等基础能力。

1. 总体认识

1.1 设计原则（来自 docs/index.rst）

• 只使用标准库或极少量外部依赖（importlib-metadata）
• 尽量支持 Linux / macOS / Windows，不支持处优雅降级
• 纯 Python 3，无 C 扩展

1.2 模块结构

1
2
3
4
5
6
7
8
9
10

osrf_pycommon/
├── osrf_pycommon/
│   ├── process_utils/     # 子进程执行（同步/异步）★ ROS 2 最常用
│   ├── terminal_color/    # ANSI 颜色与转义序列
│   ├── terminal_utils.py  # 终端尺寸、is_tty
│   └── cli_utils/         # CLI verb 模式、参数解析辅助
├── tests/                 # unittest 单元测试
├── docs/                  # Sphinx 文档
├── setup.py
└── package.xml

2. process_utils — 核心模块

路径：osrf_pycommon/process_utils/
这是 ROS 2 生态中使用最广泛的部分。

2.1 公开 API

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

from .async_execute_process import async_execute_process
from .async_execute_process import asyncio
from .async_execute_process import AsyncSubprocessProtocol
from .async_execute_process import get_loop

from .impl import execute_process
from .impl import execute_process_split
from .impl import which

__all__ = [
    'async_execute_process',
    'asyncio',
    'AsyncSubprocessProtocol',
    'get_loop',
    'execute_process',
    'execute_process_split',
    'which',
]

2.2 同步执行路径

1
2
3

execute_process / execute_process_split  (impl.py)
    ├── emulate_tty=False → execute_process_nopty.py
    └── emulate_tty=True  → execute_process_pty.py (Unix only)

nopty 实现（execute_process_nopty.py）：

• 基于 subprocess.Popen + select.select（Unix）或 readline（Windows）
• 按行缓冲输出，保留换行符
• 可用 stderr_to_stdout 合并或分离 stderr

pty 实现（execute_process_pty.py）：

• 用 pty.openpty() 让子进程认为在 TTY 上运行
• 子进程会输出彩色日志、启用行缓冲（如 Python -u 行为）
• 注意 pty 数量有限，大量并行可能 OSError: out of pty devices
• Windows 无 pty，自动回退 nopty

2.3 异步执行路径

1
2
3

async_execute_process  (async_execute_process_asyncio/impl.py)
    ├── emulate_tty=False → loop.subprocess_exec / subprocess_shell
    └── emulate_tty=True  → pty + connect_read_pipe

AsyncSubprocessProtocol（async_execute_process.py）：

• 继承 asyncio.SubprocessProtocol
• 可覆写 on_stdout_received / on_stderr_received / on_process_exited
• protocol.complete 是 Future，完成时结果为 return code

get_loop（get_loop_impl.py）：

• 线程局部单例事件循环
• Windows 强制使用 ProactorEventLoop（子进程管道需要）
• 处理 Python 3.10 的 DeprecationWarning

2.4 which 回退实现

Python 3.3+ 优先用 shutil.which；旧版本使用 _which_backport，支持：

• Windows PATHEXT（.exe 等）
• 相对路径 ./script
• 目录排除（避免把目录当可执行文件）

3. 在 ROS 2 launch 中的关键作用

3.1 ExecuteLocal — 启动节点进程

launch/actions/execute_local.py 是最大消费者：

1
2

from osrf_pycommon.process_utils import async_execute_process
from osrf_pycommon.process_utils import AsyncSubprocessProtocol

ExecuteLocal 用 async_execute_process 启动 ros2 run、节点可执行文件等，并通过自定义 Protocol 将 stdout/stderr 转为 launch 事件（ProcessStdout / ProcessStderr）。

3.2 LaunchService 事件循环

1
2
3
4
5
6
7

loop = osrf_pycommon.process_utils.get_loop()
run_async_task = loop.create_task(self.run_async(
    shutdown_when_idle=shutdown_when_idle
))
while True:
    try:
        return loop.run_until_complete(run_async_task)

整个 ros2 launch 的异步调度建立在 get_loop() 返回的事件循环上。

3.3 查找可执行文件

• launch/substitutions/find_executable.py → which
• launch_ros/substitutions/executable_in_package.py → which

用于解析 launch 文件中 $(find-pkg-prefix) 等替换后的可执行路径。

3.4 launch_testing 颜色剥离

• launch_testing/tools/output.py
• launch_testing/asserts/assert_output.py

使用 remove_ansi_escape_sequences 在断言输出时去掉 ANSI 转义，避免颜色码导致测试失败。

4. terminal_color — 终端颜色

路径：osrf_pycommon/terminal_color/

4.1 子模块

4.2 两种着色方式

1. 直接 ANSI 码

1
2

from osrf_pycommon.terminal_color import ansi
print(ansi('red') + "error" + ansi('reset'))

2. @{} 标记替换

1
2

from osrf_pycommon.terminal_color import format_color
print(format_color("This is @{bf}blue@{reset}."))

支持 @{redf} / @{rf} / @{r} 等多种简写；背景色必须带 b 后缀（如 @{rb}）。

4.3 平台行为

• Linux/macOS：正常输出 ANSI 转义
• Windows：ansi() 返回空字符串；需用 print_color() 或 print_ansi_color_win32() 才能显示颜色
• 可调用 disable_ansi_color_substitution_globally() 全局关闭颜色

4.4 ANSI 处理工具

1
2
3
4
5
6

def remove_ansi_escape_sequences(string):
    """
    Removes any ansi escape sequences found in the given string and returns it.
    """
    global _ansi_re
    return _ansi_re.sub('', string)

正则 \033\[\d{1,2}[m] 匹配常见 SGR 序列；另有 split_by_ansi_escape_sequence 用于分段处理。

5. terminal_utils — 终端工具

单文件 terminal_utils.py，仅 3 个公开符号：

用途相对独立，ROS 2 核心路径较少直接引用。

6. cli_utils — CLI 扩展模式

路径：osrf_pycommon/cli_utils/

6.1 verb_pattern — 插件式子命令

这是 ROS 2 早期 ros2 xxx verb 风格 CLI 的基础模式（ros2 topic echo 等），基于 setuptools entry_points 动态加载 verb：

verb 模块约定结构（见 docs/cli_utils.rst）：

1
2
3
4
5
6

verb = 'myverb'
description = '...'
def prepare_arguments(parser): ...
def main(args): ...
# 可选
def argument_preprocessor(args): ...

注意：当前 Humble 工作区中，ros2cli 不再直接依赖 osrf_pycommon，而是使用自有的 ros2cli 框架；cli_utils 仍可作为构建类似 CLI 的参考库，也被 colcon 等工具的历史版本使用过。

6.2 common — 参数解析辅助

典型场景：构建工具需要把「传给 make 的并行参数」与「传给目标的参数」分开。

7. 依赖与构建

7.1 package.xml

1

<exec_depend>python3-importlib-metadata</exec_depend>

verb_pattern.list_verbs 用 importlib.metadata.entry_points() 发现插件；Python 3.8+ 内置，旧版通过 importlib-metadata 包提供。

7.2 setup.py 要点

• ament_python 包，注册 resource/osrf_pycommon 索引
• 测试 extra：flake8、pytest
• zip_safe=True

8. 测试结构

1
2
3
4
5
6
7

tests/
├── test_code_format.py          # flake8
└── unit/
    ├── test_process_utils/      # 同步/异步/pty 子进程
    ├── test_terminal_color/
    ├── test_terminal_utils.py
    └── test_cli_utils/

进程测试包含 stdout_stderr_ordering 等 fixture，验证 nopty/pty 模式下输出顺序行为。

9. 在 ROS 2 栈中的位置

1
2
3
4
5
6
7
8
9
10

用户: ros2 launch my_pkg launch.py
         │
         ▼
    launch.LaunchService.run()
         │ get_loop()
         ▼
    ExecuteLocal.execute()
         │ async_execute_process(AsyncSubprocessProtocol, cmd)
         ▼
    子进程 (talker, listener, ...)

与 CycloneDDS、iceoryx 等不同，osrf_pycommon 不参与通信，只服务于 Python 工具链的运行时基础设施。

10. 设计特点小结

11. 推荐阅读顺序

1. launch 集成：launch/actions/execute_local.py — 看 Protocol 如何包装子进程 I/O
2. 异步基础：process_utils/async_execute_process_asyncio/impl.py + get_loop_impl.py
3. 同步/pty：process_utils/impl.py → execute_process_nopty.py / execute_process_pty.py
4. 测试断言：launch_testing/asserts/assert_output.py — 颜色剥离用法
5. CLI 模式：docs/cli_utils.rst + cli_utils/verb_pattern.py
6. 文档：docs/process_utils.rst、docs/terminal_color.rst

如果你希望，我可以把本文写入 ros2doc/osrf_pycommon/，或继续分析 launch 中 ExecuteLocal 如何基于 AsyncSubprocessProtocol 转发 stdout 的完整调用链。

ament_cmake_ros 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/ros2/ament_cmake_ros
子包数量：2

1. 定位

ament_cmake_ros 目录含 2 个 ROS 2 包，工作区路径见下。

2. 子包列表

3. 在 ROS 2 Humble 栈中的关系

见 ros2总览.md 分层图。

4. 推荐阅读顺序

1. 阅读各子包 package.xml 2. 入口源码 3. 下游依赖方

5. 小结

ament_cmake_ros 为含 2 个子包的源码树，是 ROS 2 Humble 发行版的一部分。

common_interfaces 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/ros2/common_interfaces
子包数量：13

1. 定位

common_interfaces 目录含 13 个 ROS 2 包，工作区路径见下。

2. 子包列表

3. 在 ROS 2 Humble 栈中的关系

标准消息定义

4. 推荐阅读顺序

1. 阅读各子包 package.xml 2. 入口源码 3. 下游依赖方

5. 小结

common_interfaces 为含 13 个子包的源码树，是 ROS 2 Humble 发行版的一部分。

console_bridge_vendor 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/ros2/console_bridge_vendor
子包数量：1

1. 定位

Vendor / 第三方依赖打包包：通过 ExternalProject 或系统包 shim 为 ROS 2 构建提供固定版本的 console_bridge，本身几乎无 ROS 业务逻辑。

2. 子包列表

3. 核心组件

典型结构：CMakeLists.txt 下载/查找库 + package.xml 导出 find_package 依赖。消费方在 package.xml 中 <depend> 本 vendor 包即可。

4. 在 ROS 2 Humble 栈中的关系

见 ros2总览.md 分层图。

5. 推荐阅读顺序

1. 阅读各子包 package.xml 2. 入口源码 3. 下游依赖方

6. 小结

console_bridge_vendor 为单包仓库，提供 console_bridge_vendor 功能。

demos 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/ros2/demos
子包数量：21

1. 定位

demos 目录含 21 个 ROS 2 包，工作区路径见下。

2. 子包列表

3. 在 ROS 2 Humble 栈中的关系

官方 demo 节点

4. 推荐阅读顺序

1. 阅读各子包 package.xml 2. 入口源码 3. 下游依赖方

5. 小结

demos 为含 21 个子包的源码树，是 ROS 2 Humble 发行版的一部分。

eigen3_cmake_module 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/ros2/eigen3_cmake_module
子包数量：1

1. 定位

Vendor / 第三方依赖打包包：通过 ExternalProject 或系统包 shim 为 ROS 2 构建提供固定版本的 eigen3_cmake_module，本身几乎无 ROS 业务逻辑。

2. 子包列表

3. 核心组件

典型结构：CMakeLists.txt 下载/查找库 + package.xml 导出 find_package 依赖。消费方在 package.xml 中 <depend> 本 vendor 包即可。

4. 在 ROS 2 Humble 栈中的关系

见 ros2总览.md 分层图。

5. 推荐阅读顺序

1. 阅读各子包 package.xml 2. 入口源码 3. 下游依赖方

6. 小结

eigen3_cmake_module 为单包仓库，提供 eigen3_cmake_module 功能。

example_interfaces 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/ros2/example_interfaces
子包数量：1

1. 定位

example_interfaces 目录含 1 个 ROS 2 包，工作区路径见下。

2. 子包列表

3. 在 ROS 2 Humble 栈中的关系

见 ros2总览.md 分层图。

4. 推荐阅读顺序

1. 阅读各子包 package.xml 2. 入口源码 3. 下游依赖方

5. 小结

example_interfaces 为单包仓库，提供 example_interfaces 功能。