ROS2►iceoryx

03 popo：发布订阅与通信原语

03 popo：发布订阅与通信原语

源码锚点（iceoryx 2.0.6，根目录 /home/cp/work2/ros2Learn/ros2_humble/src/eclipse-iceoryx/iceoryx）：
用户 API：iceoryx_posh/include/iceoryx_posh/popo/（publisher.hpp、subscriber.hpp、sample.hpp、wait_set.hpp、listener.hpp、client.hpp、server.hpp…）
Port 层：.../internal/popo/ports/；building blocks：.../internal/popo/building_blocks/
无锁队列：iceoryx_hoofs/.../concurrent/sofi.hpp、fifo.hpp、resizeable_lockfree_queue.hpp

popo（Port Port / publish-subscribe）在 mepoo 之上实现了完整的通信语义：typed/untyped 发布订阅、多订阅者分发、事件通知（WaitSet/Listener）以及 2.0 新增的 request/response。

1. 分层总览

1
2
3
4
5
6
7
8
9
10
11
12
用户 API 层     Publisher<T,H> / Subscriber<T,H>        UntypedPublisher / UntypedSubscriber
                loan()/publish() → Sample<T> RAII        loan(size)/publish(payload*)
                     │                                        │
Base 层         BasePublisher / BaseSubscriber（持 PortUser、TriggerHandle，接 WaitSet/Listener）
                     │
Port 用户侧     PublisherPortUser ── ChunkSender ── ChunkDistributor      （发送路径）
                SubscriberPortUser ── ChunkReceiver ── ChunkQueuePopper   （接收路径）
                     │                    数据全部位于共享内存 ▼
共享内存数据    PublisherPortData{ChunkSenderData}   SubscriberPortData{ChunkReceiverData=ChunkQueueData+UsedChunkList}
                     ▲
Port RouDi 侧   PublisherPortRouDi / SubscriberPortRouDi —— RouDi 处理 CaPro
                消息（OFFER/SUB…），把订阅者队列指针挂进 publisher 的 ChunkDistributor

同一份 PortData 被两个进程以不同视图操作：应用进程用 *PortUser，RouDi 用 *PortRouDi——这是 iceoryx「端口数据放共享内存、逻辑放各自进程」的核心模式。

2. Typed / Untyped API 与 Sample RAII

2.1 Typed：loan / publish

1
2
3
4
5
6
template <typename T, typename H = mepoo::NoUserHeader>
class Publisher : public PublisherImpl<T, H>
{
  public:
    using PublisherImpl<T, H>::PublisherImpl;
};

PublisherImpl 提供三种发布方式（internal/popo/publisher_impl.hpp）：

  • loan(Args&&...)：按 sizeof(T)/alignof(T) 从端口 loan 一个 chunk 并就地构造 T，返回 cxx::expected<Sample<T,H>, AllocationError>；
  • publish(Sample<T,H>&&)：发布并交还所有权；
  • publishCopyOf(const T&) / publishResultOf(callable)：便捷封装。

Sample<T,H> 是 RAII 句柄（基于 SmartChunk，内部是带自定义 deleter 的 cxx::unique_ptr<T>）：析构未发布的 Sample 会自动 release chunk 回内存池；publish() 后释放所有权、不再触发 deleter：

1
2
3
4
5
6
7
8
9
10
11
/// @brief The Sample class is a mutable abstraction over types which are written to loaned shared memory.
/// These samples are publishable to the iceoryx system.
template <typename T, typename H = cxx::add_const_conditionally_t<mepoo::NoUserHeader, T>>
class Sample : public SmartChunk<PublisherInterface<T, H>, T, H>
{
    ...
    /// @brief Publish the sample via the publisher from which it was loaned and automatically
    /// release ownership to it.
    /// @details Only available for non-const type T.
    template <typename S = T, typename = ForPublisherOnly<S, T>>
    void publish() noexcept;

订阅侧对偶：Subscriber<T,H>::take() 返回 cxx::expected<Sample<const T, const H>, ChunkReceiveResult>——const T 保证订阅者不改共享数据，Sample 析构时自动 releaseChunk（引用计数 −1）。

2.2 Untyped

UntypedPublisher::loan(payloadSize, alignment...) 返回裸 void*（payload 指针），publish(void*) 内部用 ChunkHeader::fromUserPayload()（见 02 篇 back-offset）反查头部；UntypedSubscriber::take() 返回 const void*，需手动 release。适合网关等运行期才知道大小的场景。

user-header H（如内置的 RequestHeader、或用户自定义时间戳头）通过 ChunkSettings 参与 chunk 布局，用 Sample::getUserHeader() 访问。

3. Port 体系

3.1 PortData：放在共享内存里的状态

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
struct PublisherPortData : public BasePortData
{
    PublisherPortData(const capro::ServiceDescription& serviceDescription,
                      const RuntimeName_t& runtimeName,
                      mepoo::MemoryManager* const memoryManager,
                      const PublisherOptions& publisherOptions,
                      const mepoo::MemoryInfo& memoryInfo = mepoo::MemoryInfo()) noexcept;

    using ChunkQueueData_t = SubscriberPortData::ChunkQueueData_t;
    using ChunkDistributorData_t =
        ChunkDistributorData<DefaultChunkDistributorConfig, ThreadSafePolicy, ChunkQueuePusher<ChunkQueueData_t>>;
    using ChunkSenderData_t =
        ChunkSenderData<MAX_CHUNKS_ALLOCATED_PER_PUBLISHER_SIMULTANEOUSLY, ChunkDistributorData_t>;

    ChunkSenderData_t m_chunkSenderData;

    PublisherOptions m_options;

    std::atomic_bool m_offeringRequested{false};
    std::atomic_bool m_offered{false};
};

SubscriberPortData 对应持有 ChunkReceiverData_t（= ChunkQueueData + UsedChunkList）和 m_subscriptionState（SubscribeState 状态机：NOT_SUBSCRIBED → SUBSCRIBE_REQUESTED → SUBSCRIBED → …，见 iceoryx_posh_types.hpp）。这些 PortData 由 RouDi 在管理段分配，应用经 IPC 拿到 offset 后用 PublisherPortUser/SubscriberPortUser 包装。

3.2 User 侧与 RouDi 侧

PublisherPortUser 的 API 就是薄薄一层 ChunkSender 代理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/// @brief Allocate a chunk, the ownership of the SharedChunk remains in the PublisherPortUser for being able to
/// cleanup if the user process disappears
...
cxx::expected<mepoo::ChunkHeader*, AllocationError>
tryAllocateChunk(const uint32_t userPayloadSize,
                 const uint32_t userPayloadAlignment,
                 const uint32_t userHeaderSize = 0U,
                 const uint32_t userHeaderAlignment = 1U) noexcept;
...
void sendChunk(mepoo::ChunkHeader* const chunkHeader) noexcept;
...
void offer() noexcept;
...
bool hasSubscribers() const noexcept;

offer()/subscribe() 只是置位 m_offeringRequested/m_subscribeRequested 原子标志；RouDi 的发现循环通过 PublisherPortRouDi::tryGetCaProMessage() 轮询到变化，生成 CaPro（Canonical Protocol）消息做匹配，匹配成功后由 PublisherPortRouDi 调用 ChunkDistributor::tryAddQueue(订阅者的 ChunkQueueData*, historyRequest) 完成接线。订阅侧有两种 RouDi 策略类：SubscriberPortSingleProducer（1:n）与 SubscriberPortMultiProducer（n:m），由编译期 build::CommunicationPolicy 选择（默认 ManyToManyPolicy，见 cmake/iceoryx_posh_deployment.hpp.in）。

4. 数据路径 building blocks

4.1 ChunkSender：loan 记账 + 序号

ChunkSender（building_blocks/chunk_sender.hpp/.inl）扩展 ChunkDistributor：tryAllocate 从 MemoryManager 取 chunk（并做小优化：若上一个 chunk 仅自己持有且大小合适则复用 m_lastChunkUnmanaged），登记进 UsedChunkList（容量 = 每 publisher 同时 loan 上限，默认 8）；发送时摘除并盖序号：

1
2
3
4
5
6
7
8
inline bool ChunkSender<ChunkSenderDataType>::getChunkReadyForSend(const mepoo::ChunkHeader* const chunkHeader,
                                                                   mepoo::SharedChunk& chunk) noexcept
{
    if (getMembers()->m_chunksInUse.remove(chunkHeader, chunk))
    {
        chunk.getChunkHeader()->setSequenceNumber(getMembers()->m_sequenceNumber++);
        return true;
    }

UsedChunkList（internal/popo/used_chunk_list.hpp）是为「应用随时可能死掉」设计的记账结构：定长数组 + 64 位 ShmSafeUnmanagedChunk 元素，写入单周期完成、无 torn write，RouDi 清理时可安全遍历。

4.2 ChunkDistributor：多订阅者分发与 history

ChunkDistributor 持有订阅者队列列表和 history 环（ChunkDistributorData，MAX_QUEUES = MAX_SUBSCRIBERS_PER_PUBLISHER = 256，MAX_HISTORY_CAPACITY = 16）。deliverToAllStoredQueues 是发布的核心，实现了队列满两种策略：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
template <typename ChunkDistributorDataType>
inline uint64_t ChunkDistributor<ChunkDistributorDataType>::deliverToAllStoredQueues(mepoo::SharedChunk chunk) noexcept
{
    uint64_t numberOfQueuesTheChunkWasDeliveredTo{0U};
    typename ChunkDistributorDataType::QueueContainer_t remainingQueues;
    {
        typename MemberType_t::LockGuard_t lock(*getMembers());

        bool willWaitForConsumer = getMembers()->m_consumerTooSlowPolicy == ConsumerTooSlowPolicy::WAIT_FOR_CONSUMER;
        // send to all the queues
        for (auto& queue : getMembers()->m_queues)
        {
            bool isBlockingQueue = (willWaitForConsumer && queue->m_queueFullPolicy == QueueFullPolicy::BLOCK_PRODUCER);

            if (pushToQueue(queue.get(), chunk))
            {
                ++numberOfQueuesTheChunkWasDeliveredTo;
            }
            else
            {
                if (isBlockingQueue)
                {
                    remainingQueues.emplace_back(queue);
                }
                else
                {
                    ++numberOfQueuesTheChunkWasDeliveredTo;
                    ChunkQueuePusher_t(queue.get()).lostAChunk();
                }
            }
        }
    }

    // busy waiting until every queue is served
    while (!remainingQueues.empty())
    {
        std::this_thread::yield();
        ...
    }

    addToHistoryWithoutDelivery(chunk);

    return numberOfQueuesTheChunkWasDeliveredTo;
}
  • 推送本质是拷贝一个 64 位 ShmSafeUnmanagedChunk（引用计数 +1），不拷贝数据；
  • history 环满时先释放最旧（addToHistoryWithoutDelivery），新订阅者接入时 tryAddQueue 把最近 requestedHistory 条补发给它——即 ROS latched / DDS TRANSIENT_LOCAL 的等价物；
  • 加锁的是跨进程互斥量（ThreadSafePolicy，building_blocks/locking_policy.hpp），只保护队列列表/history 容器，真正的数据入队是无锁的。

队列满策略（popo/port_queue_policies.hpp）由双方协商：

订阅者 QueueFullPolicy 发布者 ConsumerTooSlowPolicy 行为
DISCARD_OLDEST_DATA（默认） 任意 SoFi 溢出挤掉最旧样本，lostAChunk() 置 m_queueHasLostChunks 供订阅者查询
BLOCK_PRODUCER WAIT_FOR_CONSUMER 发布者在上面的 while (!remainingQueues.empty()) 中 yield 忙等直至队列有空位（真正的背压，牺牲实时性）
BLOCK_PRODUCER DISCARD_OLDEST_DATA（默认） 不兼容，RouDi 匹配时拒绝连接（NACK）

4.3 ChunkQueue 与无锁队列（SoFi）

订阅者队列 ChunkQueueData 的存储是 cxx::VariantQueue：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
template <typename ChunkQueueDataProperties, typename LockingPolicy>
struct ChunkQueueData : public LockingPolicy
{
    ...
    cxx::UniqueId m_uniqueId{};

    static constexpr uint64_t MAX_CAPACITY = ChunkQueueDataProperties_t::MAX_QUEUE_CAPACITY;
    cxx::VariantQueue<mepoo::ShmSafeUnmanagedChunk, MAX_CAPACITY> m_queue;
    std::atomic_bool m_queueHasLostChunks{false};

    rp::RelativePointer<ConditionVariableData> m_conditionVariableDataPtr;
    cxx::optional<uint64_t> m_conditionVariableNotificationIndex;
    const QueueFullPolicy m_queueFullPolicy;
};

VariantQueue（iceoryx_hoofs/cxx/variant_queue.hpp）可在四种实现间选择：FiFo_SingleProducerSingleConsumer、SoFi_SingleProducerSingleConsumer、FiFo/SoFi_MultiProducerSingleConsumer（后两者映射到 ResizeableLockFreeQueue）。pub/sub 默认用 SoFi（Safely overflowing FiFo）——无锁、溢出时返回被挤出的最旧元素而不是失败：

1
2
3
4
5
6
7
8
/// @brief
/// Thread safe producer and consumer queue with a safe overflowing behavior.
/// SoFi is designed in a FIFO Manner but prevents data loss when pushing into
/// a full SoFi. When SoFi is full and a Sender tries to push, the data at the
/// current read position will be returned. SoFi is a Thread safe without using
/// locks. ...
template <class ValueType, uint64_t CapacityValue>
class SoFi

ChunkQueuePusher::push 展示了溢出样本如何归还引用计数、以及入队后如何跨进程唤醒等待者：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
template <typename ChunkQueueDataType>
inline bool ChunkQueuePusher<ChunkQueueDataType>::push(mepoo::SharedChunk chunk) noexcept
{
    auto pushRet = getMembers()->m_queue.push(chunk);
    bool hasQueueOverflow = false;

    // drop the chunk if one is returned by an overflow
    if (pushRet.has_value())
    {
        pushRet.value().releaseToSharedChunk();
        // tell the ChunkDistributor that we had an overflow and dropped a sample
        hasQueueOverflow = true;
    }

    {
        typename MemberType_t::LockGuard_t lock(*getMembers());
        if (getMembers()->m_conditionVariableDataPtr)
        {
            ConditionNotifier(*getMembers()->m_conditionVariableDataPtr.get(),
                              *getMembers()->m_conditionVariableNotificationIndex)
                .notify();
        }
    }

    return !hasQueueOverflow;
}

4.4 ChunkReceiver

ChunkReceiver = ChunkQueuePopper + UsedChunkList。tryGet() 出队后把 chunk 记入订阅者自己的 UsedChunkList（容量 = MAX_CHUNKS_HELD_PER_SUBSCRIBER_SIMULTANEOUSLY + 1 = 257，多出的 1 个允许用户在已满持有时先拿新再还旧，与 ara::com 对齐，见 chunk_receiver_data.hpp 注释）；release() 移除并减引用。队列容量与持有上限被刻意设为相等（MAX_SUBSCRIBER_QUEUE_CAPACITY = MAX_CHUNKS_HELD_PER_SUBSCRIBER_SIMULTANEOUSLY），保证轮询用户一次能吞下整条满队列。

5. WaitSet 与 Listener：跨进程事件机制

5.1 ConditionVariable：共享内存”条件变量”

iceoryx 的跨进程通知原语不是 pthread condvar，而是POSIX 信号量 + 通知位图：

1
2
3
4
5
6
7
8
9
10
struct ConditionVariableData
{
    ...
    posix::Semaphore m_semaphore =
        std::move(posix::Semaphore::create(posix::CreateUnnamedSharedMemorySemaphore, 0U)
        ...
    RuntimeName_t m_runtimeName;
    std::atomic_bool m_toBeDestroyed{false};
    std::atomic_bool m_activeNotifications[MAX_NUMBER_OF_NOTIFIERS];
};

ConditionVariableData 由 RouDi 在管理段分配（每个 WaitSet/Listener 一个，上限 MAX_NUMBER_OF_CONDITION_VARIABLES = 1024）。生产者侧 ConditionNotifier(condVar, index).notify()：先 m_activeNotifications[index] = true 再 sem_post；消费者侧 ConditionListener::wait()：sem_wait 醒来后扫描位图、收集并清零所有已置位的 index，返回排序后的通知索引向量（condition_listener.hpp/condition_notifier.hpp）。信号量创建在共享内存中（unnamed + pshared），因此天然跨进程。

订阅者队列通过 setConditionVariable(condVar, notificationIndex) 挂接（见 4.3 中 m_conditionVariableDataPtr），一个 condvar 用不同 index 区分最多 MAX_NUMBER_OF_NOTIFIERS = 256 个事件源。

5.2 WaitSet

1
2
3
4
5
6
7
8
/// @brief Logical disjunction of a certain number of Triggers
///
/// The WaitSet stores Triggers and allows the user to wait till one or more of those Triggers are triggered. It works
/// over process borders. With the creation of a WaitSet it requests a condition variable from RouDi and destroys it
/// with the destructor. Hence the lifetime of the condition variable is bound to the lifetime of the WaitSet.
/// @param[in] Capacity the amount of events/states which can be attached to the waitset
template <uint64_t Capacity = MAX_NUMBER_OF_ATTACHMENTS_PER_WAITSET>
class WaitSet

关键概念——事件（event）vs 状态（state）：

  • attachEvent(subscriber, SubscriberEvent::DATA_RECEIVED)：边沿触发。只在 notify() 发生时唤醒一次；若 wait 期间没有新 push，即使队列里还有旧数据也不会再报。
  • attachState(subscriber, SubscriberState::HAS_DATA)：电平触发。每次 wait 返回前用 WaitSetIsConditionSatisfiedCallback（如 hasNewChunks()）复查条件，只要队列非空就持续报告——不会因为”一次唤醒处理多条中只 take 了一条”而丢事件。

attach 时 WaitSet 生成 Trigger 并向源对象（subscriber 等）交付 TriggerHandle——包含 condvar 指针、唯一 trigger id 和 reset 回调；源对象触发即 TriggerHandle::trigger() → ConditionNotifier::notify()；TriggerHandle 析构时自动从 WaitSet 反注册（双向生命周期安全，popo/trigger_handle.hpp）。wait()/timedWait() 返回 NotificationInfoVector，可携带用户 id 与回调。

5.3 Listener

Listener（popo/listener.hpp）= 一个 condvar + 内部线程：threadLoop 里 ConditionListener::wait()，对每个激活 index 并发执行注册的回调（attachEvent(obj, event, createNotificationCallback(cb))）。与 WaitSet 的区别：Listener 是推模式、只支持 event（电平语义的 state 无法映射到”回调一次”模型）、完全线程安全；WaitSet 是拉模式，用户自己控制在哪个线程处理。两者共享 MAX_NUMBER_OF_NOTIFIERS = 256 个通知槽。

6. Request/Response（2.0 新增）

client/server 复用同一套 building blocks，每侧同时有发送与接收能力（internal/popo/ports/client_server_port_types.hpp）：

1
2
3
4
5
6
7
8
9
10
11
12
13
using ClientChunkQueueData_t = ChunkQueueData<ClientChunkQueueConfig, ThreadSafePolicy>;

using ServerChunkQueueData_t = ChunkQueueData<ServerChunkQueueConfig, ThreadSafePolicy>;

using ClientChunkDistributorData_t =
    ChunkDistributorData<ClientChunkDistributorConfig, ThreadSafePolicy, ChunkQueuePusher<ServerChunkQueueData_t>>;

using ServerChunkDistributorData_t =
    ChunkDistributorData<ServerChunkDistributorConfig, ThreadSafePolicy, ChunkQueuePusher<ClientChunkQueueData_t>>;

using ClientChunkReceiverData_t = ChunkReceiverData<MAX_RESPONSES_PROCESSED_SIMULTANEOUSLY, ClientChunkQueueData_t>;

using ServerChunkReceiverData_t = ChunkReceiverData<MAX_REQUESTS_PROCESSED_SIMULTANEOUSLY, ServerChunkQueueData_t>;
  • Client：ChunkSender（向 server 的请求队列推 request，distributor 只有 1 个队列）+ ChunkReceiver（响应队列，容量 16）。
  • Server：ChunkReceiver（请求队列，容量 1024，可服务 MAX_CLIENTS_PER_SERVER = 256 个 client）+ ChunkSender（响应经 distributor 精确回投到发起请求的那个 client 队列）。

路由信息放在内置 user-header RequestHeader/ResponseHeader（popo/rpc_header.hpp）里：

1
2
3
4
5
protected:
  uint8_t m_rpcHeaderVersion{RPC_HEADER_VERSION};
  uint32_t m_lastKnownClientQueueIndex{UNKNOWN_CLIENT_QUEUE_INDEX};
  cxx::UniqueId m_uniqueClientQueueId;
  int64_t m_sequenceId{0};

请求携带 client 响应队列的 UniqueId 与索引提示，server 发响应时调 ChunkSender::sendToQueue(chunkHeader, uniqueClientQueueId, lastKnownClientQueueIndex)（getQueueIndex 先试提示索引、失败再线性查）。sequenceId 由用户设置/校验以匹配乱序响应；ResponseHeader 另有 setServerError() 标志。用户 API：typed Client<Req,Res>/Server<Req,Res>（loan → send → take）与 untyped 版本，事件枚举 ClientEvent::RESPONSE_RECEIVED/ServerEvent::REQUEST_RECEIVED 可挂 WaitSet/Listener。

连接状态机为 ConnectionState（NOT_CONNECTED → CONNECT_REQUESTED → CONNECTED → …，iceoryx_posh_types.hpp）。

7. 与 DDS QoS 概念对照

DDS QoS / 概念 iceoryx 2.0 对应 备注
RELIABILITY RELIABLE QueueFullPolicy::BLOCK_PRODUCER + ConsumerTooSlowPolicy::WAIT_FOR_CONSUMER 进程内不丢包，队列满时阻塞发布者（忙等）
RELIABILITY BEST_EFFORT DISCARD_OLDEST_DATA（默认） 溢出挤掉最旧样本，hasLostChunksSinceLastCall() 可检测
HISTORY KEEP_LAST(n) 订阅队列容量 SubscriberOptions::queueCapacity（≤256） 接收侧 history
DURABILITY TRANSIENT_LOCAL PublisherOptions::historyCapacity（≤16）+ SubscriberOptions::historyRequest 迟到订阅者补发最近 n 条
DURABILITY VOLATILE historyCapacity = 0（默认）
DEADLINE / LIVELINESS 无直接对应 RouDi 有进程级 keep-alive 监控，非 per-topic；待验证细节
PARTITION / DOMAIN ServiceDescription{Service, Instance, Event} 三元组 精确匹配，无通配 QoS 协商
OWNERSHIP EXCLUSIVE 无；OneToManyPolicy 编译期限制单发布者 多 publisher 同 topic 在 ManyToManyPolicy 下自由并存
LATENCY_BUDGET 无（零拷贝路径延迟本身为 µs 级）
RPC（DDS-RPC / ROS service） Client/Server port + RequestHeader/ResponseHeader 2.0 新增
Listener / WaitSet（DDS 实体） popo::Listener / popo::WaitSet 语义高度对应：StatusCondition ≈ attachState，事件回调 ≈ Listener

注意：这些策略在 RouDi 做订阅匹配时校验兼容性（如 BLOCK_PRODUCER 订阅者遇到 DISCARD_OLDEST_DATA 发布者会被拒绝），行为上类似 DDS 的 QoS RxO（requested vs offered）检查。

下一篇：04-RouDi守护进程与服务发现.md

文章互动

阅读 --

留言

0 条留言

正在加载留言…