ROS2►iceoryx

06 C 绑定与 DDS 网关

06 C 绑定与 DDS 网关

本文重点：iceoryx_binding_c 的 C API 设计（storage 模式、C↔C++ 映射、payload/user-header API），以及 iceoryx_dds 的 iox-dds-gateway（GatewayGeneric 框架、DDS↔iceoryx 双向桥接、Cyclone DDS 数据写入器）。
源码锚点：iceoryx_binding_c/source/c_publisher.cpp、iceoryx_binding_c/include/iceoryx_binding_c/*.h、iceoryx_dds/include/iceoryx_dds/internal/gateway/*.inl、iceoryx_dds/source/gateway/main.cpp

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

1. 为什么需要 C 绑定

iceoryx 核心（iceoryx_posh）是 C++14 实现，大量使用模板（Publisher<T>）、cxx::expected、RAII。但下游一些用户是纯 C 代码——最典型的就是 CycloneDDS（C 语言实现的 DDS），其 ENABLE_SHM 零拷贝路径完全通过 iceoryx_binding_c 的 C API 调用 iceoryx（见 07-与CycloneDDS及ROS2集成.md）。

iceoryx_binding_c 提供的 C 对象一览：

C 句柄 头文件 对应 C++ 实体 用途
iox_pub_t publisher.h cpp2c_Publisher（包装 PublisherPortUser） 发布（loan/publish chunk）
iox_sub_t subscriber.h cpp2c_Subscriber（包装 SubscriberPortUser） 订阅（take/release chunk）
iox_ws_t wait_set.h cpp2c_WaitSet 阻塞等待多事件
iox_listener_t listener.h iox::popo::Listener 后台线程回调
iox_user_trigger_t user_trigger.h UserTrigger 用户自定义触发
iox_client_t / iox_server_t client.h / server.h UntypedClient / UntypedServer 请求-响应
iox_service_discovery_t service_discovery.h ServiceDiscovery 服务发现
iox_runtime_init() runtime.h PoshRuntime::initRuntime() 向 RouDi 注册进程

2. storage 模式：用户提供内存

C 没有构造函数，iceoryx 又刻意避免在 API 内部隐式堆分配，所以 C 绑定采用 “storage 结构体 + init 函数” 模式：调用者先在自己的栈/静态区准备一块 iox_xxx_storage_t，再交给 iox_xxx_init() 初始化并返回句柄。

1
2
3
4
5
typedef struct
{
    // only size for pointer is necessary
    uint64_t do_not_touch_me[1];
} iox_pub_storage_t;

注意注释 “only size for pointer is necessary”：在 v2.0 中这些 storage 结构体只需容纳一个指针——iox_pub_init 实际上仍会 new 一个 cpp2c_Publisher，并把指针存进 storage：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
iox_pub_t iox_pub_init(iox_pub_storage_t* self,
                       const char* const service,
                       const char* const instance,
                       const char* const event,
                       const iox_pub_options_t* const options)
{
    if (self == nullptr)
    {
        LogWarn() << "publisher initialization skipped - null pointer provided for iox_pub_storage_t";
        return nullptr;
    }
    // ... 校验 options 是否经过 iox_pub_options_init ...
    auto* me = new cpp2c_Publisher();
    self->do_not_touch_me[0] = reinterpret_cast<uint64_t>(me);

    me->m_portData = PoshRuntime::getInstance().getMiddlewarePublisher(
        ServiceDescription{
            IdString_t(TruncateToCapacity, service),
            IdString_t(TruncateToCapacity, instance),
            IdString_t(TruncateToCapacity, event),
        },
        publisherOptions);
    return me;
}

要点：

  • 服务寻址沿用 iceoryx 的 CaPro 三元组 {service, instance, event}（字符串），与 C++ 层 ServiceDescription 一一对应。
  • iox_pub_init 直接绑定到底层端口对象 PublisherPortData（getMiddlewarePublisher），后续所有操作都是围绕该共享内存端口的薄包装，没有额外拷贝层。
  • storage 模式让 CycloneDDS 这类用户可以写 iox_pub_init(&(iox_pub_storage_t){0}, ...)（复合字面量），不关心生命周期细节；iox_pub_deinit 负责 delete。

2.1 options 的防呆设计

C 结构体无法保证零初始化，为防止用户漏调 init，options 里埋了 magic number（initCheck）：

1
2
3
4
5
6
7
8
9
10
11
12
13
constexpr uint64_t PUBLISHER_OPTIONS_INIT_CHECK_CONSTANT = 123454321;

void iox_pub_options_init(iox_pub_options_t* options)
{
    // ...
    PublisherOptions publisherOptions;
    options->historyCapacity = publisherOptions.historyCapacity;
    options->nodeName = nullptr;
    options->offerOnCreate = publisherOptions.offerOnCreate;
    options->subscriberTooSlowPolicy = cpp2c::consumerTooSlowPolicy(publisherOptions.subscriberTooSlowPolicy);

    options->initCheck = PUBLISHER_OPTIONS_INIT_CHECK_CONSTANT;
}

若用户传入未 init 的 options，iox_pub_init 会走 errorHandler(Error::kBINDING_C__PUBLISHER_OPTIONS_NOT_INITIALIZED)（LogFatal 后终止）。

3. Publisher C API：loan / publish 与 user-header

发布侧核心 API（publisher.h）分三级，逐级增加控制粒度：

API payload 对齐 user-header 说明
iox_pub_loan_chunk 默认 8 字节 无 最常用
iox_pub_loan_aligned_chunk 自定义 无 大对齐类型（如 SIMD）
iox_pub_loan_aligned_chunk_with_user_header 自定义 自定义大小/对齐 CycloneDDS 用它挂 iceoryx_header_t

三者最终收敛到同一实现——直接调 C++ 端口层的 tryAllocateChunk：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
iox_AllocationResult iox_pub_loan_aligned_chunk_with_user_header(iox_pub_t const self,
                                                                 void** const userPayload,
                                                                 const uint32_t userPayloadSize,
                                                                 const uint32_t userPayloadAlignment,
                                                                 const uint32_t userHeaderSize,
                                                                 const uint32_t userHeaderAlignment)
{
    auto result = PublisherPortUser(self->m_portData)
                      .tryAllocateChunk(userPayloadSize, userPayloadAlignment, userHeaderSize, userHeaderAlignment)
                      .and_then([&userPayload](ChunkHeader* h) { *userPayload = h->userPayload(); });
    if (result.has_error())
    {
        return cpp2c::allocationResult(result.get_error());
    }

    return AllocationResult_SUCCESS;
}

发布与释放同样是对 ChunkHeader 的直译（用户拿到的是 payload 指针，内部通过 ChunkHeader::fromUserPayload 反查头部）：

1
2
3
4
5
6
7
8
9
void iox_pub_release_chunk(iox_pub_t const self, void* const userPayload)
{
    PublisherPortUser(self->m_portData).releaseChunk(ChunkHeader::fromUserPayload(userPayload));
}

void iox_pub_publish_chunk(iox_pub_t const self, void* const userPayload)
{
    PublisherPortUser(self->m_portData).sendChunk(ChunkHeader::fromUserPayload(userPayload));
}

其余管理 API：iox_pub_offer / iox_pub_stop_offer / iox_pub_is_offered / iox_pub_has_subscribers / iox_pub_get_service_description。

3.1 payload ↔ user-header ↔ chunk-header 转换（chunk.h）

一个 chunk 的内存布局是 ChunkHeader | user-header(可选) | user-payload（详见 posh 的 mepoo::ChunkHeader）。C 侧通过 chunk.h 的 6 个函数在三者之间导航：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/// @brief gets the user-payload from the chunk-header
void* iox_chunk_header_to_user_payload(iox_chunk_header_t* const chunkHeader);

const void* iox_chunk_header_to_user_payload_const(const iox_chunk_header_t* const chunkHeader);

/// @brief gets the user-header from the chunk-header
void* iox_chunk_header_to_user_header(iox_chunk_header_t* const chunkHeader);

const void* iox_chunk_header_to_user_header_const(const iox_chunk_header_t* const chunkHeader);

/// @brief gets the chunk-header from the user-payload
iox_chunk_header_t* iox_chunk_header_from_user_payload(void* const userPayload);

const iox_chunk_header_t* iox_chunk_header_from_user_payload_const(const void* const userPayload);

CycloneDDS 正是用 iox_chunk_header_from_user_payload + iox_chunk_header_to_user_header 把自己的 iceoryx_header_t（GUID、时间戳、keyhash 等 DDS 元数据）藏进 user-header（见 07 篇 §4）。

4. Subscriber C API

订阅侧与发布侧对称（subscriber.h）：

API 说明
iox_sub_init / iox_sub_deinit storage 模式创建/销毁
iox_sub_take_chunk(sub, &userPayload) 取一个 chunk（零拷贝，拿到共享内存指针）
iox_sub_release_chunk 归还 chunk（引用计数减一，勿忘！否则 chunk 泄漏）
iox_sub_release_queued_chunks 清空队列中所有未取的 chunk
iox_sub_has_chunks / iox_sub_has_lost_chunks 有无数据 / 是否因队列溢出丢过数据
iox_sub_subscribe / iox_sub_unsubscribe / iox_sub_get_subscription_state 订阅状态管理

iox_sub_options_t 中值得注意的字段：queueCapacity（接收队列深度）、historyRequest（连接时补发历史样本数）、queueFullPolicy（队列满时丢最旧 or 阻塞发布者，对应 C++ 的 QueueFullPolicy）。

5. WaitSet 与 Listener C API

两者是 iceoryx 的两种事件通知机制在 C 侧的映射：

  • WaitSet（iox_ws_t）：用户线程主动阻塞等待，语义类似 epoll。
  • Listener（iox_listener_t）：内部起后台线程，事件到达时调用用户回调，语义类似”信号槽”。

WaitSet 核心等待接口：

1
2
3
4
5
6
7
8
9
10
11
uint64_t iox_ws_timed_wait(iox_ws_t const self,
                           struct timespec timeout,
                           iox_notification_info_t* const notificationInfoArray,
                           const uint64_t notificationInfoArrayCapacity,
                           uint64_t* missedElements);

/// @brief waits until an event occurred
uint64_t iox_ws_wait(iox_ws_t const self,
                     iox_notification_info_t* const notificationInfoArray,
                     const uint64_t notificationInfoArrayCapacity,
                     uint64_t* missedElements);

可附着（attach）的对象包括 subscriber 的 state（如 SubscriberState_HAS_DATA，电平语义）和 event（如 SubscriberEvent_DATA_RECEIVED，边沿语义），以及 user-trigger、client、server、service-discovery。每个 attach 均有 _with_context_data 变体，把 void* 用户数据透传进回调：

1
2
3
4
5
6
ENUM iox_WaitSetResult iox_ws_attach_subscriber_state_with_context_data(iox_ws_t const self,
                                                                        iox_sub_t const subscriber,
                                                                        const ENUM iox_SubscriberState subscriberState,
                                                                        const uint64_t id,
                                                                        void (*callback)(iox_sub_t, void*),
                                                                        void* const contextData);

Listener 的 attach 接口形态一致（无 id、无超时，事件直接驱动回调线程）：

1
2
3
4
ENUM iox_ListenerResult iox_listener_attach_subscriber_event(iox_listener_t const self,
                                                             iox_sub_t const subscriber,
                                                             const ENUM iox_SubscriberEvent subscriberEvent,
                                                             void (*callback)(iox_sub_t));

CycloneDDS 的 shm_monitor.c 就是典型 Listener 用户：iox_listener_attach_subscriber_event_with_context_data(..., SubscriberEvent_DATA_RECEIVED, shm_subscriber_callback, &reader->m_iox_sub_context)，在 iceoryx 后台线程里把 chunk 搬进 DDS reader cache（见 07 篇 §5）。

5.1 C ↔ C++ 映射层

internal/ 下的翻译层保证两种语言的枚举/对象一致：

文件 方向 内容
c2cpp_enum_translation.hpp/.cpp C → C++ 如 iox_QueueFullPolicy → popo::QueueFullPolicy
cpp2c_enum_translation.hpp/.cpp C++ → C 如 AllocationError → iox_AllocationResult
cpp2c_publisher.hpp / cpp2c_subscriber.hpp — C 句柄背后的包装类（持有 m_portData 端口指针）
cpp2c_waitset.hpp — WaitSet 的 C 包装
c2cpp_binding.h — CLASS/ENUM 宏：同一头文件在 C 里是 struct/裸枚举，在 C++ 里是 class/强类型

6. iceoryx_dds：iox-dds-gateway

6.1 用途与定位

iceoryx 只做同一台主机内的共享内存通信。当数据需要跨主机时，iceoryx_dds 提供可执行程序 iox-dds-gateway：它作为一个普通 iceoryx 应用挂到 RouDi 上，把本机 iceoryx topic 的数据经 DDS（网络） 转发出去，并把远端 DDS 数据注入本机 iceoryx，从而组成”机内零拷贝 + 机间 DDS 网络”的混合拓扑。

1
2
主机 A                                      主机 B
app --共享内存--> iox-dds-gateway ==DDS/UDP网络==> iox-dds-gateway --共享内存--> app

注意：iceoryx_dds/ 目录下存在 COLCON_IGNORE 空标记文件（iceoryx_examples/ 同样有）。这意味着 ROS 2 Humble 用 colcon 构建工作区时该包被完全跳过、不参与编译。ROS 2 场景下跨主机通信由 CycloneDDS 自身的 UDP 路径完成（一份数据同时走 iceoryx 与网络，见 07 篇），并不需要 iox-dds-gateway。该网关面向的是”纯 iceoryx 应用 + 跨机扩展”的独立部署场景，需单独用 CMake（-DDDS_STACK=CYCLONE_DDS，见 iceoryx_dds/CMakeLists.txt）构建。

6.2 GatewayGeneric 框架

posh 提供了通用网关基类 gw::GatewayGeneric<channel_t>（iceoryx_posh/gateway/gateway_generic.hpp）：内部维护一组 Channel（iceoryx 终端 + 外部终端的配对），起两个线程分别周期执行：

  • discovery 线程（默认 1000ms）：消费 RouDi 广播的 CaPro 发现消息，调用子类 discover() 动态建/拆 Channel;
  • forwarding 线程（默认 50ms）：遍历所有 Channel 调用子类 forward() 搬运数据。

子类只需实现 loadConfiguration() / discover() / forward() 三个纯虚函数。iceoryx_dds 基于它派生出两个方向的网关：

1
2
3
4
5
6
7
8
9
10
11
12
/// @brief DDS Gateway implementation for the iceoryx to DDS direction.
template <typename channel_t = gw::Channel<popo::UntypedSubscriber, dds::data_writer_t>,
          typename gateway_t = gw::GatewayGeneric<channel_t>>
class Iceoryx2DDSGateway : public gateway_t
{
  public:
    /// @brief Creates a gateway with DDS set as interface
    Iceoryx2DDSGateway() noexcept;

    void loadConfiguration(const config::GatewayConfig& config) noexcept;
    void discover(const capro::CaproMessage& msg) noexcept;
    void forward(const channel_t& channel) noexcept;

Channel 的两端：iceoryx 侧是 UntypedSubscriber（iox→DDS 方向）或 UntypedPublisher（DDS→iox 方向）；外部侧是 dds::data_writer_t / data_reader_t（编译期通过 dds_types.hpp 绑定到 Cyclone 实现）。

6.3 iceoryx → DDS 方向

发现：收到 RouDi 的 OFFER CaPro 消息就自动为该服务建 Channel（STOP_OFFER 则拆除），内省服务被排除：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
switch (msg.m_type)
{
case capro::CaproMessageType::OFFER:
{
    if (!this->findChannel(msg.m_serviceDescription).has_value())
    {
        popo::SubscriberOptions options;
        options.queueCapacity = SUBSCRIBER_CACHE_SIZE;
        IOX_DISCARD_RESULT(setupChannel(msg.m_serviceDescription, options));
    }
    break;
}
case capro::CaproMessageType::STOP_OFFER:
{
    if (this->findChannel(msg.m_serviceDescription).has_value())
    {
        IOX_DISCARD_RESULT(this->discardChannel(msg.m_serviceDescription));
    }
    break;
}

转发：从 untyped subscriber take 出 chunk，把 ChunkHeader 的元信息填进 IoxChunkDatagramHeader，连同 user-header、payload 交给 DDS writer：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
template <typename channel_t, typename gateway_t>
inline void Iceoryx2DDSGateway<channel_t, gateway_t>::forward(const channel_t& channel) noexcept
{
    auto subscriber = channel.getIceoryxTerminal();
    while (subscriber->hasData())
    {
        subscriber->take().and_then([&](const void* userPayload) {
            auto dataWriter = channel.getExternalTerminal();
            auto chunkHeader = iox::mepoo::ChunkHeader::fromUserPayload(userPayload);
            iox::dds::IoxChunkDatagramHeader datagramHeader;
            datagramHeader.userHeaderId = chunkHeader->userHeaderId();
            datagramHeader.userHeaderSize = chunkHeader->userHeaderSize();
            datagramHeader.userPayloadSize = chunkHeader->userPayloadSize();
            datagramHeader.userPayloadAlignment = chunkHeader->userPayloadAlignment();
            dataWriter->write(datagramHeader,
                              static_cast<const uint8_t*>(chunkHeader->userHeader()),
                              static_cast<const uint8_t*>(chunkHeader->userPayload()));
            subscriber->release(userPayload);
        });
    }
}

IoxChunkDatagramHeader 是跨网络传输 chunk 的线格式头（版本号 + 端序 + user-header/payload 尺寸与对齐）：

1
2
3
4
5
6
7
8
9
10
11
static constexpr uint8_t DATAGRAM_VERSION{1U};

/// @note This must always be the first member and always 1 bytes in order to prevent issues with endianess when
/// deserialized or incorrectly detected versions due to different size
uint8_t datagramVersion{DATAGRAM_VERSION};
/// @note This must always be 1 byte in order to prevent issues with endianess when deserialized
Endianess endianness{Endianess::UNDEFINED};
uint16_t userHeaderId{0xFFFF};
uint32_t userHeaderSize{0U};
uint32_t userPayloadSize{0U};
uint32_t userPayloadAlignment{0U};

6.4 CycloneDataWriter：DDS 侧如何写

cyclone_data_writer.cpp 使用 CycloneDDS 的 C++ 绑定（cyclonedds-cxx）。topic 名由 CaPro 三元组拼接，类型是 IDL 定义的字节序列 Mempool::Chunk（iceoryx_dds/msg/Mempool.idl）：

1
2
3
4
5
6
7
8
void iox::dds::CycloneDataWriter::connect() noexcept
{
    m_publisher = ::dds::pub::Publisher(CycloneContext::getParticipant());
    auto topic = "/" + std::string(m_serviceId) + "/" + std::string(m_instanceId) + "/" + std::string(m_eventId);
    m_topic = ::dds::topic::Topic<Mempool::Chunk>(CycloneContext::getParticipant(), topic);
    m_writer = ::dds::pub::DataWriter<Mempool::Chunk>(m_publisher, m_topic);
    LogDebug() << "[CycloneDataWriter] Connected to topic: " << topic;
}

write() 把 序列化的 datagramHeader | user-header | payload 三段依次拷贝进 Mempool::Chunk 的字节 vector 再 m_writer.write(chunk)：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
datagramHeader.endianness = getEndianess();

auto serializedDatagramHeader = iox::dds::IoxChunkDatagramHeader::serialize(datagramHeader);
auto datagramSize =
    serializedDatagramHeader.size() + datagramHeader.userHeaderSize + datagramHeader.userPayloadSize;

auto chunk = Mempool::Chunk();
chunk.payload().reserve(datagramSize);

std::copy(serializedDatagramHeader.data(),
          serializedDatagramHeader.data() + serializedDatagramHeader.size(),
          std::back_inserter(chunk.payload()));
// ... 追加 userHeaderBytes、userPayloadBytes ...
m_writer.write(chunk);

显然这一步是有拷贝、有序列化的——网关做的是”共享内存 → 网络报文”的边界转换，零拷贝只存在于机内 iceoryx 段。

6.5 DDS → iceoryx 方向

DDS2IceoryxGateway::forward() 是镜像流程：先 peek DDS 侧的 datagram 头，按其中的 size/alignment 向 iceoryx loan 一个 chunk，再把网络数据 takeNext 直接解到 chunk 的 user-header/payload 里后 publish：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
while (reader->hasSamples())
{
    reader->peekNextIoxChunkDatagramHeader().and_then([&](auto datagramHeader) {
        constexpr uint32_t USER_HEADER_ALIGNMENT{1U};
        publisher
            ->loan(datagramHeader.userPayloadSize,
                   datagramHeader.userPayloadAlignment,
                   datagramHeader.userHeaderSize,
                   USER_HEADER_ALIGNMENT)
            .and_then([&](auto userPayload) {
                auto chunkHeader = iox::mepoo::ChunkHeader::fromUserPayload(userPayload);
                reader
                    ->takeNext(datagramHeader,
                               static_cast<uint8_t*>(chunkHeader->userHeader()),
                               static_cast<uint8_t*>(chunkHeader->userPayload()))
                    .and_then([&]() { publisher->publish(userPayload); })
                    // ... or_else: release + LogWarn ...

局限：DDS2IceoryxGateway::discover() 是空实现（源码注释 “requires dds discovery which is currently not implemented in the used dds stack”），因此 DDS→iox 方向只能靠 TOML 静态配置服务列表，无法自动发现远端 DDS topic。

6.6 网关主程序

iox-dds-gateway 可执行文件同时跑两个方向的网关，配置来自 TOML（默认 /etc/iceoryx/gateway_config.toml，解析失败则退回”全自动发现”默认配置）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
iox::runtime::PoshRuntime::initRuntime("iox-dds-gateway");

iox::config::GatewayConfig gatewayConfig;
iox::dds::Iceoryx2DDSGateway<> iox2ddsGateway;
iox::dds::DDS2IceoryxGateway<> dds2ioxGateway;

iox::config::TomlGatewayConfigParser::parse()
    .and_then([&](auto config) { gatewayConfig = config; })
    .or_else([&](auto err) {
        iox::dds::LogWarn() << "[Main] Failed to parse gateway config with error: "
                            << iox::config::TOML_GATEWAY_CONFIG_FILE_PARSE_ERROR_STRINGS[err];
        iox::dds::LogWarn() << "[Main] Using default configuration.";
        gatewayConfig.setDefaults();
    });

iox2ddsGateway.loadConfiguration(gatewayConfig);
dds2ioxGateway.loadConfiguration(gatewayConfig);

iox2ddsGateway.runMultithreaded();
dds2ioxGateway.runMultithreaded();

7. 小结

  • iceoryx_binding_c 是对 posh 端口层的零抽象成本直译：storage 模式规避隐式分配 API 的观感（内部仍有一次 new），loan/publish/take/release 与 C++ 完全同语义，是 CycloneDDS SHM 集成的地基。
  • iceoryx_dds 展示了 GatewayGeneric 的扩展方式：discovery + forwarding 双线程模板，把机内共享内存数据桥接到 DDS 网络；但在 ROS 2 Humble 构建中被 COLCON_IGNORE 排除，属独立部署组件。

下一篇

07-与CycloneDDS及ROS2集成.md：CycloneDDS 如何用这些 C API 实现同机零拷贝，以及 ROS 2 Humble 的完整启用链路。

文章互动

阅读 --

留言

0 条留言

正在加载留言…