ROS2►iceoryx

05 零拷贝数据路径详解

05 零拷贝数据路径详解

发布端：popo/building_blocks/chunk_sender.inl、chunk_distributor.inl；订阅端：chunk_receiver.inl、chunk_queue_popper.inl
内存：mepoo/memory_manager.cpp、mem_pool.cpp、shared_chunk.cpp
源码根：ros2_humble/src/eclipse-iceoryx/iceoryx（版本 2.0.6）

1. 传统中间件 vs iceoryx：拷贝次数对比

以一次进程间 pub/sub 为例，传统 socket 类中间件（UDP loopback / UDS）的路径：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
传统 (以 UDS/UDP loopback 为例, 至少 2~4 次拷贝 + 2 次系统调用):

  发布进程                    内核                     订阅进程
  ┌────────┐  ①序列化拷贝   ┌────────┐   ③拷贝到     ┌────────┐
  │ 用户数据 │──→ 发送缓冲 ──→│ 内核缓冲 │──→ 接收缓冲 ──→│ 反序列化 │
  └────────┘   ②write()    └────────┘   read()       └────────┘

iceoryx (0 次拷贝, 数据本体不移动):

  发布进程                共享内存                  订阅进程
  loan() ──────────→ ┌──────────────┐
  直接原地构造 T      │  chunk (T)    │ ←────────── take()
  publish() 只推送 ──→│              │   拿到的是同一块内存的
  8 字节相对指针      └──────────────┘   const T* 只读引用
       └→ 订阅者无锁队列 ← ChunkDistributor

iceoryx 里”发送”一个 4 MB 的样本和发送 1 KB 的样本代价相同：真正跨进程传递的只是一个 8 字节的 ShmSafeUnmanagedChunk（相对指针），数据本体自始至终躺在共享内存 mempool 的同一个 chunk 里。这正是 iceperf 基准中 iceoryx 延迟与 payload 大小无关的原因（见第 11 节）。

参与者与所属文件（全部在 iceoryx_posh 中）：

构件 文件 职责
Publisher<T> / Sample<T> include/iceoryx_posh/internal/popo/publisher_impl.inl 类型安全 API
PublisherPortUser source/popo/ports/publisher_port_user.cpp 端口用户视图
ChunkSender include/.../building_blocks/chunk_sender.inl 分配/发送 chunk
ChunkDistributor include/.../building_blocks/chunk_distributor.inl 推送到 N 个订阅队列 + history
MemoryManager / MemPool source/mepoo/memory_manager.cpp、mem_pool.cpp 共享内存池
ChunkQueuePusher/Popper include/.../building_blocks/chunk_queue_*.inl 无锁队列两端
ChunkReceiver include/.../building_blocks/chunk_receiver.inl 订阅端取 chunk
SharedChunk source/mepoo/shared_chunk.cpp 引用计数句柄

2. 发布端：loan() 的完整调用链

1
2
3
4
5
6
7
8
9
10
11
Publisher<T>::loan()                          publisher_impl.inl:39
  → PublisherImpl::loanSample()               publisher_impl.inl:71
      → PublisherPortUser::tryAllocateChunk() publisher_port_user.cpp:42
          → ChunkSender::tryAllocate()        chunk_sender.inl:100
              ├─ (快路径) 复用 m_lastChunkUnmanaged
              └─ MemoryManager::getChunk()    memory_manager.cpp:152
                    → MemPool::getChunk()     mem_pool.cpp:78   (LoFFLi 无锁空闲链表 pop)
                    → placement-new ChunkHeader + ChunkManagement
                    → 返回 SharedChunk (引用计数=1)
  → 在 chunk 的 userPayload 上 placement-new T(...)
  → 包装成 Sample<T,H> 返回

2.1 类型层：在共享内存上原地构造

1
2
3
4
5
6
7
template <typename T, typename H, typename BasePublisherType>
template <typename... Args>
inline cxx::expected<Sample<T, H>, AllocationError>
PublisherImpl<T, H, BasePublisherType>::loan(Args&&... args) noexcept
{
    return std::move(loanSample().and_then([&](auto& sample) { new (sample.get()) T(std::forward<Args>(args)...); }));
}

loanSample() 把 sizeof(T)/alignof(T) 交给端口层：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
template <typename T, typename H, typename BasePublisherType>
inline cxx::expected<Sample<T, H>, AllocationError> PublisherImpl<T, H, BasePublisherType>::loanSample() noexcept
{
    static constexpr uint32_t USER_HEADER_SIZE{std::is_same<H, mepoo::NoUserHeader>::value ? 0U : sizeof(H)};

    auto result = port().tryAllocateChunk(sizeof(T), alignof(T), USER_HEADER_SIZE, alignof(H));
    if (result.has_error())
    {
        return cxx::error<AllocationError>(result.get_error());
    }
    else
    {
        return cxx::success<Sample<T, H>>(convertChunkHeaderToSample(result.value()));
    }
}

2.2 ChunkSender::tryAllocate()：快路径与 mempool 分配

ChunkSender 有个重要优化：m_lastChunkUnmanaged 缓存上一次发送的 chunk，若它已无其他持有者（所有订阅者都消费完、引用计数==1）且大小够用，就原地复用——发布循环稳定后基本不再触碰 mempool：

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
auto& lastChunkUnmanaged = getMembers()->m_lastChunkUnmanaged;
mepoo::ChunkHeader* lastChunkChunkHeader =
    lastChunkUnmanaged.isNotLogicalNullptrAndHasNoOtherOwners() ? lastChunkUnmanaged.getChunkHeader() : nullptr;

if (lastChunkChunkHeader && (lastChunkChunkHeader->chunkSize() >= requiredChunkSize))
{
    auto sharedChunk = lastChunkUnmanaged.cloneToSharedChunk();
    if (getMembers()->m_chunksInUse.insert(sharedChunk))
    {
        auto chunkSize = lastChunkChunkHeader->chunkSize();
        lastChunkChunkHeader->~ChunkHeader();
        new (lastChunkChunkHeader) mepoo::ChunkHeader(chunkSize, chunkSettings);
        lastChunkChunkHeader->setOriginId(originId);
        return cxx::success<mepoo::ChunkHeader*>(lastChunkChunkHeader);
    }
    ...
}
else
{
    // BEGIN of critical section, chunk will be lost if the process terminates in this section
    // get a new chunk
    auto getChunkResult = getMembers()->m_memoryMgr->getChunk(chunkSettings);

    if (!getChunkResult.has_error())
    {
        auto& chunk = getChunkResult.value();

        // if the application allocated too much chunks, return no more chunks
        if (getMembers()->m_chunksInUse.insert(chunk))
        {
            // END of critical section
            chunk.getChunkHeader()->setOriginId(originId);
            return cxx::success<mepoo::ChunkHeader*>(chunk.getChunkHeader());
        }

m_chunksInUse（UsedChunkList）登记该进程当前借出的 chunk，超过上限报 TOO_MANY_CHUNKS_ALLOCATED_IN_PARALLEL；它也是 RouDi 崩溃清理的依据（第 9.3 节）。

2.3 MemoryManager::getChunk()：按大小挑 mempool

MemoryManager 持有若干按 chunk 大小升序排列的 MemPool（数据段中）以及一个 ChunkManagement 管理池（管理段中）。分配时线性找第一个足够大的池：

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
cxx::expected<SharedChunk, MemoryManager::Error> MemoryManager::getChunk(const ChunkSettings& chunkSettings) noexcept
{
    void* chunk{nullptr};
    MemPool* memPoolPointer{nullptr};
    const auto requiredChunkSize = chunkSettings.requiredChunkSize();
    ...
    for (auto& memPool : m_memPoolVector)
    {
        uint32_t chunkSizeOfMemPool = memPool.getChunkSize();
        if (chunkSizeOfMemPool >= requiredChunkSize)
        {
            chunk = memPool.getChunk();
            memPoolPointer = &memPool;
            aquiredChunkSize = chunkSizeOfMemPool;
            break;
        }
    }
    ...
    else
    {
        auto chunkHeader = new (chunk) ChunkHeader(aquiredChunkSize, chunkSettings);
        auto chunkManagement = new (m_chunkManagementPool.front().getChunk())
            ChunkManagement(chunkHeader, memPoolPointer, &m_chunkManagementPool.front());
        return cxx::success<SharedChunk>(SharedChunk(chunkManagement));
    }
}

MemPool::getChunk() 本身是无锁的——空闲索引由 LoFFLi（Lock-Free Free List）维护：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
void* MemPool::getChunk() noexcept
{
    uint32_t l_index{0U};
    if (!m_freeIndices.pop(l_index))
    {
        std::cerr << "Mempool [m_chunkSize = " << m_chunkSize << ", numberOfChunks = " << m_numberOfChunks
                  << ", used_chunks = " << m_usedChunks << " ] has no more space left" << std::endl;
        return nullptr;
    }
    ...
    m_usedChunks.fetch_add(1U, std::memory_order_relaxed);
    adjustMinFree();

    return m_rawMemory + l_index * m_chunkSize;
}

每个 chunk 的布局：ChunkHeader（含 chunkSize、序列号、originId、userPayload 偏移等）+ 可选 user-header + 用户 payload。配套的 ChunkManagement（引用计数 + 指回 header/mempool 的相对指针）单独放在管理段的管理池里。

3. 发布端：publish() 如何把相对指针推入订阅队列

1
2
3
4
5
6
7
8
9
Sample<T>::publish() → PublisherImpl::publish()        publisher_impl.inl:87
  → PublisherPortUser::sendChunk()                     publisher_port_user.cpp:56
      → ChunkSender::send()                            chunk_sender.inl:184
          → getChunkReadyForSend(): 从 m_chunksInUse 移出、打序列号
          → ChunkDistributor::deliverToAllStoredQueues() chunk_distributor.inl:141
              → 对每个订阅队列 ChunkQueuePusher::push() chunk_queue_pusher.inl:47
                  → VariantQueue<ShmSafeUnmanagedChunk>::push()  (无锁 FIFO/SOFI)
                  → ConditionNotifier::notify()        (可选, 唤醒 WaitSet/Listener)
              → addToHistoryWithoutDelivery()          (latched/history 支持)

PublisherImpl::publish() 只做指针换算，不碰数据：

1
2
3
4
5
6
7
template <typename T, typename H, typename BasePublisherType>
inline void PublisherImpl<T, H, BasePublisherType>::publish(Sample<T, H>&& sample) noexcept
{
    auto userPayload = sample.release(); // release the Samples ownership of the chunk before publishing
    auto chunkHeader = mepoo::ChunkHeader::fromUserPayload(userPayload);
    port().sendChunk(chunkHeader);
}

核心的多播分发在 ChunkDistributor::deliverToAllStoredQueues()。注意队列列表 m_queues 就是 RouDi 在 CaPro SUB 握手时挂进来的订阅者接收队列（见 04 篇 5.3 节）：

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
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())

进入队列的元素是 ShmSafeUnmanagedChunk——一个被压缩到 8 字节 的 (segmentId, offset) 相对指针，指向 ChunkManagement。8 字节保证了写入的原子性（防 torn write），这是崩溃安全的关键：

1
2
3
4
5
6
7
8
9
// Torn writes are problematic since RouDi needs to cleanup all chunks when an application crashes. If the size is
// larger than 8 bytes on a 64 bit system, torn writes happens and the data is only partially written when the
// application crashes at the wrong time. RouDi would then read corrupt data and try to access invalid memory.
static_assert(sizeof(ShmSafeUnmanagedChunk) <= 8U,
              "The ShmSafeUnmanagedChunk size must not exceed 64 bit to prevent torn writes!");
...
static_assert(std::is_trivially_copyable<ShmSafeUnmanagedChunk>::value,
              "The ShmSafeUnmanagedChunk must be trivially copyable to prevent Frankenstein objects when the copy ctor "
              "works on half dead objects!");

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;
}

队列类型是 cxx::VariantQueue（chunk_queue_data.hpp 第 46 行），订阅端口默认使用 SOFI（Safely Overflowing FIFO，SoFi_SingleProducerSingleConsumer/多生产者用 lock-free queue，iceoryx_hoofs/cxx/variant_queue.hpp 第 39~45 行）：满时 push 会”挤出”最老的元素并返回给推送方释放——天然实现 KEEP_LAST 语义。

4. 订阅端：take() 调用链与相对指针→绝对地址

1
2
3
4
5
6
7
8
9
Subscriber<T>::take()                          subscriber_impl.inl:34
  → BaseSubscriber::takeChunk()                base_subscriber.inl:89
      → SubscriberPortUser::tryGetChunk()      subscriber_port_user.cpp:67
          → ChunkReceiver::tryGet()            chunk_receiver.inl:74
              → ChunkQueuePopper::tryPop()     chunk_queue_popper.inl:47
                  → m_queue.pop() 得到 ShmSafeUnmanagedChunk
                  → releaseToSharedChunk(): 相对指针 → 本进程绝对地址
              → m_chunksInUse.insert(chunk)    (登记, 引用计数保持)
  → userPayload() 包成 Sample<const T> (只读)

tryPop() 中完成相对指针到绝对地址的转换（rp::RelativePointer 查本进程注册的 segment 基址表）并做 header 版本校验：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
template <typename ChunkQueueDataType>
inline cxx::optional<mepoo::SharedChunk> ChunkQueuePopper<ChunkQueueDataType>::tryPop() noexcept
{
    auto retVal = getMembers()->m_queue.pop();

    // check if queue had an element that was poped and return if so
    if (retVal.has_value())
    {
        auto chunk = retVal.value().releaseToSharedChunk();

        auto receivedChunkHeaderVersion = chunk.getChunkHeader()->chunkHeaderVersion();
        if (receivedChunkHeaderVersion != mepoo::ChunkHeader::CHUNK_HEADER_VERSION)
        {
            ...
            return cxx::nullopt_t();
        }
        return cxx::make_optional<mepoo::SharedChunk>(chunk);
    }

ShmSafeUnmanagedChunk::releaseToSharedChunk() 的实现（offset+id 还原为 ChunkManagement*）：

1
2
3
4
5
6
7
8
9
10
SharedChunk ShmSafeUnmanagedChunk::releaseToSharedChunk() noexcept
{
    if (m_chunkManagement.isLogicalNullptr())
    {
        return SharedChunk();
    }
    auto chunkMgmt = rp::RelativePointer<mepoo::ChunkManagement>(m_chunkManagement.offset(), m_chunkManagement.id());
    m_chunkManagement.reset();
    return SharedChunk(chunkMgmt);
}

上层 ChunkReceiver::tryGet() 把 chunk 登记进订阅者自己的 m_chunksInUse（持有上限 MAX_CHUNKS_HELD_PER_SUBSCRIBER_SIMULTANEOUSLY），最后 SubscriberImpl::take() 包装成只读 Sample<const T>：

1
2
3
4
5
6
7
8
9
10
11
12
13
template <typename T, typename H, typename BaseSubscriberType>
inline cxx::expected<Sample<const T, const H>, ChunkReceiveResult>
SubscriberImpl<T, H, BaseSubscriberType>::take() noexcept
{
    auto result = BaseSubscriberType::takeChunk();
    if (result.has_error())
    {
        return cxx::error<ChunkReceiveResult>(result.get_error());
    }
    auto userPayloadPtr = static_cast<const T*>(result.value()->userPayload());
    auto samplePtr = cxx::unique_ptr<const T>(userPayloadPtr, m_sampleDeleter);
    return cxx::success<Sample<const T, const H>>(std::move(samplePtr));
}

5. 引用计数与释放回 MemPool

SharedChunk 是 chunk 的智能句柄，引用计数存在共享内存的 ChunkManagement 里（所有进程共享同一个计数）。持有者包括：发布者的 m_chunksInUse/m_lastChunkUnmanaged/history、每个订阅队列里的元素、每个订阅者 take() 后的 Sample。任何进程中最后一个 SharedChunk 析构时归还两块内存：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
SharedChunk::~SharedChunk() noexcept
{
    decrementReferenceCounter();
}
...
void SharedChunk::decrementReferenceCounter() noexcept
{
    if ((m_chunkManagement != nullptr)
        && (m_chunkManagement->m_referenceCounter.fetch_sub(1U, std::memory_order_relaxed) == 1U))
    {
        freeChunk();
    }
}

void SharedChunk::freeChunk() noexcept
{
    m_chunkManagement->m_mempool->freeChunk(m_chunkManagement->m_chunkHeader);
    m_chunkManagement->m_chunkManagementPool->freeChunk(m_chunkManagement);
    m_chunkManagement = nullptr;
}

MemPool::freeChunk() 把索引 push 回 LoFFLi 空闲链表，并带双重释放检测（mem_pool.cpp 第 96~112 行，kPOSH__MEMPOOL_POSSIBLE_DOUBLE_FREE）。订阅者侧 Sample 析构 → SampleDeleter → SubscriberPortUser::releaseChunk() → ChunkReceiver::release() 从 m_chunksInUse 移除，触发上述计数递减。

6. 跨进程同步：无锁队列 + ConditionVariable

数据路径上没有互斥锁跨进程共享（ChunkDistributor 的锁默认是同进程/RouDi 间的 ThreadSafePolicy 互斥量，push/pop 队列本身无锁）。订阅者等待新数据有两种方式：

  1. 轮询：循环 take() 直到 NO_CHUNK_AVAILABLE。
  2. 事件驱动：WaitSet/Listener 把 ConditionVariableData（在共享内存中，含一个进程间 POSIX 信号量 + 通知位数组）挂到订阅端口；发布者 push 后 ConditionNotifier::notify()：
1
2
3
4
5
6
7
8
9
10
void ConditionNotifier::notify() noexcept
{
    if (m_notificationIndex < MAX_NUMBER_OF_NOTIFIERS)
    {
        getMembers()->m_activeNotifications[m_notificationIndex].store(true, std::memory_order_release);
    }
    getMembers()->m_semaphore.post().or_else([](auto) {
        errorHandler(Error::kPOPO__CONDITION_NOTIFIER_SEMAPHORE_CORRUPT_IN_NOTIFY, nullptr, ErrorLevel::FATAL);
    });
}

订阅进程在 ConditionListener::wait() 中 sem_wait 阻塞，醒来后扫描通知位数组确定是哪个 trigger（condition_listener.cpp 第 91~117 行 waitImpl）。这是数据路径上唯一的系统调用点，且只在订阅者选择阻塞等待时发生。

7. 多订阅者：同一 chunk 只读共享

N 个订阅者的场景下，deliverToAllStoredQueues() 把同一个 chunk 的相对指针 push 进 N 个队列，每次 push 都通过 SharedChunk 拷贝构造把引用计数 +1。没有任何数据复制：

  • 所有订阅者 take() 得到的 Sample<const T> 指向同一物理内存，类型系统强制只读（const T*）。
  • 订阅者数据段通常以 READ_ONLY mmap（shared_memory_user.cpp 第 62 行按段权限决定），越权写会 SIGSEGV。
  • 各订阅者独立释放，最后一个释放者把 chunk 归还 mempool（第 5 节）。

这也带来约束：订阅者拿到样本后不能修改；需要修改就得自己 loan 新 chunk 拷贝（此时才有一次拷贝）。

8. history / latched topic 支持

对应 ROS/DDS 的 transient-local / latched 语义：

  • 发布侧 PublisherOptions::historyCapacity（上限 MAX_PUBLISHER_HISTORY = 编译期 CMake 变量 IOX_MAX_PUBLISHER_HISTORY，默认 16，见 iceoryx_posh/cmake/IceoryxPoshDeployment.cmake 第 5556 行）。每次发送后 chunk 也存入 m_history 环形缓存：deliverToAllStoredQueues() 末尾调用 addToHistoryWithoutDelivery(chunk)（chunk_distributor.inl 第 212 行），容量满则释放最老的（第 296312 行）。
  • 订阅侧 SubscriberOptions::historyRequest。RouDi 转发 SUB 消息时携带该值，发布者 tryAddQueue() 在挂接队列的同时把 history 里最后 requestedHistory 个 chunk 立即补发给新订阅者：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
const auto currChunkHistorySize = getMembers()->m_history.size();

if (requestedHistory > getMembers()->m_historyCapacity)
{
    LogWarn() << "Chunk history request exceeds history capacity! Request is " << requestedHistory
              << ". Capacity is " << getMembers()->m_historyCapacity << ".";
}

// if the current history is large enough we send the requested number of chunks, else we send the
// total history
const auto startIndex =
    (requestedHistory <= currChunkHistorySize) ? currChunkHistorySize - requestedHistory : 0u;
for (auto i = startIndex; i < currChunkHistorySize; ++i)
{
    pushToQueue(queueToAdd, getMembers()->m_history[i].cloneToSharedChunk());
}

另外未 offer 时 sendChunk() 不投递、只入 history（publisher_port_user.cpp 第 56~72 行），供 AUTOSAR field 之类”先设值后 offer”的用法。

9. 异常场景

9.1 订阅队列满 / 订阅者太慢

由发布者的 subscriberTooSlowPolicy 与订阅者的 queueFullPolicy 组合决定（不兼容组合在 RouDi 匹配时就被拒绝，见 04 篇 5.3 节）：

组合 行为
DISCARD_OLDEST_DATA + DISCARD_OLDEST_DATA（默认） SOFI 队列挤掉最老样本，lostAChunk() 置丢失标记，订阅者可用 hasMissedData() 查询（chunk_queue_popper.inl 第 76 行）
WAIT_FOR_CONSUMER + BLOCK_PRODUCER 发布者在 deliverToAllStoredQueues() 中对满队列 busy-wait（std::this_thread::yield() 循环，chunk_distributor.inl 第 173~210 行），实现可靠背压；应用退出时靠 PREPARE_APP_TERMINATION 让 RouDi 强制 STOP_OFFER 解除阻塞

9.2 mempool 耗尽 / 样本过大

MemoryManager::getChunk() 区分三种错误（memory_manager.cpp 第 172~201 行）：无 mempool（FATAL）、无足够大的 mempool（NO_MEMPOOL_FOR_REQUESTED_CHUNK_SIZE，样本大小超过配置上限）、池空（MEMPOOL_OUT_OF_CHUNKS）。上层映射为 AllocationError::RUNNING_OUT_OF_CHUNKS 等返回给 loan() 调用者。chunk 耗尽常见原因：订阅者长期持有 Sample 不释放、history 容量过大、queueCapacity × 订阅者数超过池容量。

9.3 chunk 泄漏检测与崩溃清理

  • 每个端口的 m_chunksInUse（UsedChunkList）记录借出未还的 chunk。应用正常销毁端口或崩溃后 RouDi 清理端口时调用 releaseAllChunks() → ChunkSender::releaseAll()：
1
2
3
4
5
6
7
template <typename ChunkSenderDataType>
inline void ChunkSender<ChunkSenderDataType>::releaseAll() noexcept
{
    getMembers()->m_chunksInUse.cleanup();
    this->cleanup();
    getMembers()->m_lastChunkUnmanaged.releaseToSharedChunk();
}
  • 应用把 Sample 的 payload 指针传错/重复释放会触发 kPOPO__CHUNK_SENDER_INVALID_CHUNK_TO_FREE_FROM_USER / kPOSH__MEMPOOL_POSSIBLE_DOUBLE_FREE。
  • 源码注释明确标注了两处”临界区”：tryAllocate/send 中若进程恰好在拿到 chunk 与登记之间死亡，该 chunk 会泄漏（chunk_sender.inl 第 142、188 行注释）；以及若应用死在 ChunkDistributor 持锁期间，RouDi 清理会死锁并 FATAL（chunk_distributor.inl 第 351~358 行 cleanup() 注释）。
  • 内省 mempool topic 展示每个池的 usedChunks 与历史最小空闲 minFree（mem_pool.cpp 的 MemPoolInfo），是定位泄漏的第一工具。

10. 端到端时序图

订阅进程ConditionVariable(sem)订阅队列(无锁,共享内存)MemPool(共享内存)发布进程订阅进程ConditionVariable(sem)订阅队列(无锁,共享内存)MemPool(共享内存)发布进程placement-new T 原地构造(0拷贝)读 const T* (同一块物理内存)loan(): LoFFLi pop 空闲index → chunkpublish(): push 8字节相对指针 (refcount++)notify(): sem_postsem_wait 返回take(): pop → 相对指针转绝对地址Sample析构: refcount-- == 0 → freeChunk

11. 延迟特性与 iceperf 基准

iceoryx_examples/iceperf 是官方 ping-pong 延迟基准：leader/follower 两个进程对 1 KB~4096 KB 的 payload 各做 N 次 round trip，对比 POSIX MQ、UDS、iceoryx C++/C API（iceperf_leader.cpp 的 doMeasurement()）。README 中官方参考结果（Ubuntu 18.04, Xeon E3-1505M v5, 10 万次往返，单程平均延迟 µs）：

Payload MQ UDS iceoryx
1 kB 3.1 4.3 0.73
64 kB 23 27 0.6
512 kB 160 210 0.61
4096 kB 1200 1700 0.61

关键观察：iceoryx 延迟 ≈ 0.6 µs 且与 payload 大小无关（只传指针），而 MQ/UDS 随大小线性增长（每字节都要过内核拷贝两次）。运行方式：

1
2
3
iox-roudi &
./iceperf-bench-follower &
./iceperf-bench-leader -n 100000

延迟构成（iceoryx 路径）：LoFFLi pop/push 若干原子操作 + 队列 push/pop + （事件模式下）一次 sem_post/sem_wait。轮询模式可再省掉信号量，进入亚微秒级。

12. 使用约束

零拷贝的前提是”数据放哪儿都能被别的进程直接解释”，因此：

  1. POD / 无指针数据：类型必须可以直接放在共享内存——不能含裸指针、引用、虚函数表、std::string/std::vector 等堆分配成员（各进程堆地址不同、且堆不共享）。iceoryx 提供 iox::cxx::string/vector（定长、可放共享内存）替代。严格说要求可 relocatable，编译期没有强制的 is_trivially_copyable 校验（待验证：typed API 不做此 static_assert，需开发者自行保证）。
  2. 固定大小上限：chunk 来自预配置的 mempool，样本（含 ChunkHeader/user-header）不能超过最大 mempool 的 chunk 大小；动态大小需求要按最坏情况配置池（TOML，见 04 篇 8.1 节），或用 untyped API 自带 size 参数 loan。
  3. 同机限制：共享内存不跨主机。跨机需要网关（iceoryx-dds、cyclonedds/FastDDS 的 iceoryx 集成）把 chunk 内容再序列化上网络——那一跳自然不再是零拷贝。
  4. 其他实践约束：订阅者拿到的是 const 视图不可写；loan 出的 chunk 应尽快 publish 或释放（占用 m_chunksInUse 名额）；单个 publisher 并行借出 chunk 数、单个 subscriber 并行持有 chunk 数均有编译期上限（iceoryx_posh_types.hpp 中 MAX_CHUNKS_ALLOCATED_PER_PUBLISHER_SIMULTANEOUSLY / MAX_CHUNKS_HELD_PER_SUBSCRIBER_SIMULTANEOUSLY）。

下一篇

文章互动

阅读 --

留言

0 条留言

正在加载留言…