ROS2►iceoryx

08 示例与性能工具

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. 示例总览

示例 演示点 关键 API
icehello 最小 pub/sub Publisher<T>::loan/publish、Subscriber<T>::take
icedelivery typed/untyped 四种发布姿势 publishCopyOf、publishResultOf、untyped loan(size)
iceperf 各 IPC 技术延迟基准 ping-pong 轮转、多 payload 扫描
waitset 阻塞式多事件等待 WaitSet::attachState/attachEvent/wait
callbacks (listener) 事件驱动回调 Listener::attachEvent + UserTrigger
request_response 请求-响应模式 Client<Req,Res>、Server<Req,Res>、sequenceId
user_header 自定义 user-header（时间戳） Publisher<Data, Header>、getUserHeader()
singleprocess RouDi 与应用同进程 PoshRuntimeSingleProcess、RouDi 内嵌
icediscovery 服务发现 ServiceDiscovery::findService、通配符

其余目录（*_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]
姿势 API 场景
#1 loan() 后就地填写 常规零拷贝
#2 loan(args...) 就地构造 有非默认构造参数
#3 publishCopyOf(obj) 小对象，接受一次拷贝
#4 publishResultOf(callable, args...) 由回调直接写入 loan 出的内存

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};
技术 实现文件 说明
POSIX MQ mq.cpp 内核消息队列（macOS 不支持）
Unix Domain Socket uds.cpp 内核 socket
iceoryx C++ API iceoryx.cpp untyped loan/publish/take
iceoryx C API iceoryx_c.cpp binding_c 同路径

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"

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

选项 显示内容 排障用途
--mempool 每个共享内存段、每档 chunk 尺寸的 total/used/min free mempool 是否耗尽、chunk 尺寸档位是否合理（printMemPoolInfo，introspection_app.cpp:259）
--process 已注册进程列表（PID、名字） 应用是否成功挂上 RouDi
--port 所有 publisher/subscriber 端口、CaPro 三元组、连接关系与 runtime 归属 topic 名对不上、pub/sub 没连上（如 CycloneDDS 场景下能看到 DDS_CYCLONE 服务名的端口）

用法示例：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

症状 可能原因 检查/处置
应用启动即卡住，反复打印等待 RouDi，最终 terminate iox-roudi 没起 / 起晚了 先 pgrep iox-roudi；RouDi 必须先于所有应用（含开启 SHM 的 ROS 2 节点）启动
应用被 SIGKILL 后重启报”process already registered” RouDi 里残留旧进程注册 RouDi 监控模式（-m on）可自动清理；否则重启 RouDi
loan 返回 RUNNING_OUT_OF_CHUNKS / CycloneDDS 报 OUT_OF_RESOURCES mempool 耗尽或没有足够大的 chunk 档位 iox-introspection-client --mempool 看 min free；调大 RouDi TOML 中对应尺寸档位的 chunk 数量/大小（记得算上 ChunkHeader + user-header 开销）
日志 TOO_MANY_CHUNKS_HELD_IN_PARALLEL 订阅端 take 后不 release，持有超上限 检查是否遗漏 release/dds_return_loan/Sample 生命周期；C API 尤其容易漏 iox_sub_release_chunk
chunk 泄漏（mempool used 只增不减） publisher loan 后既不 publish 也不 release 内省 mempool 视图定位段；审查所有早退/异常路径
订阅端丢样本（iox_sub_has_lost_chunks 为真） 队列溢出（发布快于消费） 增大 queueCapacity（上限 256）、降低发布频率、或把 queueFullPolicy 设为阻塞发布者（配合发布端 subscriberTooSlowPolicy）
段权限错误（无法 map 共享内存段） 应用用户不在段配置的用户组里 参考 iceoryx_examples/ice_access_control；检查 RouDi 段配置 TOML 的 reader/writer 用户组与进程属组
pub/sub 建了但对不上 CaPro 三元组不一致（含大小写） iox-introspection-client --port 对照两侧服务描述

13. 相关文档

下一篇

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

文章互动

阅读 --

留言

0 条留言

正在加载留言…