ROS2►iceoryx

02 mepoo：共享内存段与固定大小内存池

02 mepoo：共享内存段与固定大小内存池

源码锚点（iceoryx 2.0.6，根目录 /home/cp/work2/ros2Learn/ros2_humble/src/eclipse-iceoryx/iceoryx）：
公共头：iceoryx_posh/include/iceoryx_posh/mepoo/（chunk_header.hpp、chunk_settings.hpp、mepoo_config.hpp、segment_config.hpp）
内部头：iceoryx_posh/include/iceoryx_posh/internal/mepoo/（mem_pool.hpp、memory_manager.hpp、mepoo_segment.hpp/.inl、segment_manager.hpp/.inl、chunk_management.hpp、shared_chunk.hpp、typed_mem_pool.hpp）
实现：iceoryx_posh/source/mepoo/；无锁 free-list：iceoryx_hoofs/.../concurrent/loffli.hpp

mepoo（Memory Pool）是 iceoryx 零拷贝通信的地基：RouDi 启动时创建共享内存段并在段内切好一组固定大小的 chunk 池；应用进程 attach 后，publisher 直接在共享内存里 loan chunk、写数据、投递 offset——全程没有 memcpy，也没有运行时 malloc。

1. 总体结构

1
2
3
4
5
6
7
8
9
10
11
12
RouDi 进程
 ├─ 管理段（iceoryx_mgmt）
 │    ├─ SegmentManager           ← 所有段的元信息，应用通过它查询自己可映射的段
 │    ├─ 各 MemPool 的 LoFFLi 空闲索引表（管理内存）
 │    ├─ ChunkManagement 池（引用计数对象池）
 │    └─ port 数据 / 条件变量等（见 03 篇）
 │
 └─ 用户数据段（每个 [reader组, writer组] 一个，shm 名 = writer 组名）
      ├─ MemPool #0: 128B  × 10000   ┐
      ├─ MemPool #1: 1KB   × 5000    │ 按 chunk 大小递增排列
      ├─ ...                         │ 的 bucket（桶）
      └─ MemPool #6: 4MB   × 10      ┘

两类内存的分离是刻意设计：管理数据（free-list、引用计数）只放在管理段，用户数据段里只有 chunk 本身。这样即使应用进程只有数据段的只读权限，也不影响引用计数的读写（管理段对所有应用可读写）。

2. MePooSegment 与 SegmentManager：段与用户组权限

2.1 段的创建（RouDi 侧）

MePooSegment 封装一个 POSIX 共享内存对象 + 一个 MemoryManager。构造时按 writer 组名创建 shm 并用 ACL 精细设置权限：

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
template <typename SharedMemoryObjectType, typename MemoryManagerType>
inline MePooSegment<SharedMemoryObjectType, MemoryManagerType>::MePooSegment(
    const MePooConfig& mempoolConfig,
    posix::Allocator& managementAllocator,
    const posix::PosixGroup& readerGroup,
    const posix::PosixGroup& writerGroup,
    const iox::mepoo::MemoryInfo& memoryInfo) noexcept
    : m_sharedMemoryObject(std::move(createSharedMemoryObject(mempoolConfig, writerGroup)))
    , m_readerGroup(readerGroup)
    , m_writerGroup(writerGroup)
    , m_memoryInfo(memoryInfo)
{
    using namespace posix;
    AccessController accessController;
    if (!(readerGroup == writerGroup))
    {
        accessController.addPermissionEntry(
            AccessController::Category::SPECIFIC_GROUP, AccessController::Permission::READ, readerGroup.getName());
    }
    accessController.addPermissionEntry(
        AccessController::Category::SPECIFIC_GROUP, AccessController::Permission::READWRITE, writerGroup.getName());
    ...
    m_memoryManager.configureMemoryManager(mempoolConfig, managementAllocator, *m_sharedMemoryObject.getAllocator());
    m_sharedMemoryObject.finalizeAllocation();
}

要点：

  • 读写权限按 POSIX 用户组划分：writer 组 READWRITE、reader 组 READ、others NONE，通过文件 ACL（AccessController::writePermissionsToFile）落到 shm 文件描述符上。
  • shm 名字就是 writer 组名（SharedMemoryObjectType::create(writerGroup.getName(), ...)），大小 = MemoryManager::requiredChunkMemorySize(config)。
  • 创建后调用 rp::BaseRelativePointer::registerPtr() 把「段基址 → segmentId」注册进 relative-pointer 机制，之后所有跨进程指针都以 (segmentId, offset) 形式存储。

2.2 SegmentManager：按用户查段

SegmentManager 持有所有 MePooSegment，本身也放在管理段里。应用注册时它按调用者的用户组算出该映射哪些段、以什么权限映射：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
template <typename SegmentType>
inline typename SegmentManager<SegmentType>::SegmentMappingContainer
SegmentManager<SegmentType>::getSegmentMappings(const posix::PosixUser& user) noexcept
{
    // get all the groups the user is in
    auto groupContainer = user.getGroups();
    ...
    for (const auto& groupID : groupContainer)
    {
        for (const auto& segment : m_segmentContainer)
        {
            if (segment.getWriterGroup() == groupID)
            {
                // a user is allowed to be only in one writer group, as we currently only support one memory manager per
                // process
                if (!foundInWriterGroup)
                {
                    mappingContainer.emplace_back(segment.getWriterGroup().getName(), ...);
                    foundInWriterGroup = true;
                }
                else
                {
                    errorHandler(Error::kMEPOO__USER_WITH_MORE_THAN_ONE_WRITE_SEGMENT);
                    ...

约束：一个用户最多属于一个 writer 组（一个进程只支持一个可写 MemoryManager）；其余匹配 reader 组的段以只读方式映射。getSegmentInformationWithWriteAccessForUser() 则返回可写段的 MemoryManager 引用与 segmentId，供该进程的 publisher 分配 chunk 使用。

3. MemPool：固定大小块 + 无锁 free-list

3.1 为什么固定大小

  • O(1) 且无碎片：分配 = 从 free-list 弹出一个索引，addr = base + index * chunkSize；释放 = 压回索引。没有 first-fit 扫描、没有 external fragmentation，实时性可证。
  • 跨进程安全：free-list 是 32 位索引数组（LoFFLi），不含绝对指针，任何进程映射到不同基址都能用。
  • 崩溃可清理：chunk 的归属由索引可逆推（offset / chunkSize），RouDi 能在应用死亡后可靠回收。
  • 代价是内部碎片（申请 200B 会拿到 1KB bucket 的 chunk），因此 bucket 配置要贴合实际消息大小分布（见第 9 节）。

3.2 结构与分配路径

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
class MemPool
{
  public:
    using freeList_t = concurrent::LoFFLi;
    static constexpr uint64_t CHUNK_MEMORY_ALIGNMENT = 8U; // default alignment for 64 bit

    MemPool(const cxx::greater_or_equal<uint32_t, CHUNK_MEMORY_ALIGNMENT> chunkSize,
            const cxx::greater_or_equal<uint32_t, 1> numberOfChunks,
            posix::Allocator& managementAllocator,
            posix::Allocator& chunkMemoryAllocator) noexcept;
    ...
    void* getChunk() noexcept;
    ...
    void freeChunk(const void* chunk) noexcept;

  private:
    ...
    rp::RelativePointer<uint8_t> m_rawMemory;

    uint32_t m_chunkSize{0U};
    /// needs to be 32 bit since loffli supports only 32 bit numbers
    /// (cas is only 64 bit and we need the other 32 bit for the aba counter)
    uint32_t m_numberOfChunks{0U};

    std::atomic<uint32_t> m_usedChunks{0U};
    std::atomic<uint32_t> m_minFree{0U};

    freeList_t m_freeIndices;
};

注意两个 allocator：chunk 本体从 chunkMemoryAllocator（用户数据段）划出，LoFFLi 索引表从 managementAllocator（管理段）划出。

getChunk / freeChunk（source/mepoo/mem_pool.cpp）：

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
void* MemPool::getChunk() noexcept
{
    uint32_t l_index{0U};
    if (!m_freeIndices.pop(l_index))
    {
        std::cerr << "Mempool [m_chunkSize = " << m_chunkSize << ...
        return nullptr;
    }
    ...
    m_usedChunks.fetch_add(1U, std::memory_order_relaxed);
    adjustMinFree();

    return m_rawMemory + l_index * m_chunkSize;
}

void MemPool::freeChunk(const void* chunk) noexcept
{
    cxx::Expects(m_rawMemory <= chunk
                 && chunk <= m_rawMemory + (static_cast<uint64_t>(m_chunkSize) * (m_numberOfChunks - 1U)));

    auto offset = static_cast<const uint8_t*>(chunk) - m_rawMemory;
    cxx::Expects(offset % m_chunkSize == 0);

    uint32_t index = static_cast<uint32_t>(offset / m_chunkSize);

    if (!m_freeIndices.push(index))
    {
        errorHandler(Error::kPOSH__MEMPOOL_POSSIBLE_DOUBLE_FREE);
    }

    m_usedChunks.fetch_sub(1U, std::memory_order_relaxed);
}

3.3 LoFFLi：Lock-Free Free-List

concurrent::LoFFLi（hoofs）是单链式无锁空闲索引栈，head 是 64 位原子 Node{indexToNextFreeIndex, abaCounter}——32 位索引 + 32 位 ABA 计数器打包进一次 CAS，这就是 chunk 数被限制在 32 位的原因：

1
2
3
4
5
6
7
8
struct alignas(8) Node
{
    Index_t indexToNextFreeIndex;
    uint32_t abaCounter;
};

static_assert(sizeof(Node) <= 8U,
              "The size of 'Node' must not exceed 8 bytes in order to be lock-free on 64 bit systems!");

push 还能检测重复归还（返回 false → kPOSH__MEMPOOL_POSSIBLE_DOUBLE_FREE），是 double-free 的最后防线。

4. ChunkHeader：真实内存布局

每个 chunk 的头部是一个 ChunkHeader（32 字节，8 字节对齐），后随可选的 user-header 与按需对齐的 user-payload：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
private:
  // the order of these members must be changed carefully and if this happens, the m_chunkHeaderVersion
  // needs to be adapted in order to be able to detect incompatibilities between publisher/subscriber
  // or record&replay, m_chunkSize and m_chunkHeaderVersion should therefore neither changed the type,
  // nor the position

  // size of the whole chunk, including the header
  uint32_t m_chunkSize{0U};
  uint8_t m_chunkHeaderVersion{CHUNK_HEADER_VERSION};
  // reserved for future functionality and used to indicate the padding bytes; currently not used and set to `0`
  uint8_t m_reserved{0};
  // currently just a placeholder
  uint16_t m_userHeaderId{NO_USER_HEADER};
  popo::UniquePortId m_originId{popo::InvalidPortId};
  uint64_t m_sequenceNumber{0U};
  uint32_t m_userHeaderSize{0U};
  uint32_t m_userPayloadSize{0U};
  uint32_t m_userPayloadAlignment{1U};
  UserPayloadOffset_t m_userPayloadOffset{sizeof(ChunkHeader)};

字段说明：

字段 含义
m_chunkSize 整个 chunk 大小（含头），即所属 bucket 的 chunkSize
m_chunkHeaderVersion 布局版本号（record&replay 兼容性检测），当前为 1
m_userHeaderId user-header 类型 id；无 user-header 时为 NO_USER_HEADER(0x0000)，2.0 中有 user-header 时统一填 UNKNOWN_USER_HEADER(0xFFFF)（占位，尚未开放自定义 id）
m_originId 发送方 publisher 的 UniquePortId，由 ChunkSender（friend）通过私有 setOriginId 写入
m_sequenceNumber 发送序号，ChunkSender::send 时递增写入
m_userPayloadOffset payload 相对 ChunkHeader 起始地址的偏移

4.1 对齐计算与 back-offset

payload 对齐由构造函数按三种情形计算（source/mepoo/chunk_header.cpp）：

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
if (userHeaderSize == 0U)
{
    if (userPayloadAlignment <= alignof(mepoo::ChunkHeader))
    {
        // the most simple case with no user-header and the user-payload adjacent to the ChunkHeader
        m_userPayloadOffset = sizeof(ChunkHeader);
    }
    else
    {
        // the second most simple case with no user-header but the user-payload alignment
        // exceeds the ChunkHeader alignment and is therefore not necessarily adjacent
        uint64_t addressOfChunkHeader = reinterpret_cast<uint64_t>(this);
        uint64_t headerEndAddress = addressOfChunkHeader + sizeof(ChunkHeader);
        uint64_t alignedUserPayloadAddress =
            iox::cxx::align(headerEndAddress, static_cast<uint64_t>(userPayloadAlignment));
        uint64_t offsetToUserPayload = alignedUserPayloadAddress - addressOfChunkHeader;
        ...
        m_userPayloadOffset = static_cast<UserPayloadOffset_t>(offsetToUserPayload);

        // this is safe since the alignment of the user-payload is larger than the one from the ChunkHeader
        ...
        auto addressOfBackOffset = alignedUserPayloadAddress - sizeof(UserPayloadOffset_t);
        auto backOffset = reinterpret_cast<UserPayloadOffset_t*>(addressOfBackOffset);
        *backOffset = m_userPayloadOffset;
    }
}

布局图（最复杂情形，带 user-header 与自定义对齐）：

1
2
3
4
5
chunk 起始（8 字节对齐）
┌───────────────────┬──────────────┬─── padding ───┬────────────┬──────────────────┐
│ ChunkHeader (32B) │ user-header  │  (对齐填充)    │ backOffset │   user-payload    │
└───────────────────┴──────────────┴───────────────┴─────4B─────┴──────────────────┘
                     ↑ 总是紧邻 ChunkHeader                       ↑ 按 userPayloadAlignment 对齐

back-offset 是紧贴 payload 前面的 4 字节，存 payload→ChunkHeader 的偏移。这样 ChunkHeader::fromUserPayload() 只需读 payload 前 4 字节就能 O(1) 反查头部——这正是 Untyped API 里用户只持有 payload 指针也能 release/publish 的原因：

1
2
3
4
5
6
7
8
9
10
11
12
ChunkHeader* ChunkHeader::fromUserPayload(void* const userPayload) noexcept
{
    if (userPayload == nullptr)
    {
        return nullptr;
    }
    uint64_t userPayloadAddress = reinterpret_cast<uint64_t>(userPayload);
    // the back-offset is always stored in front of the user-payload, no matter if a user-header is used or not or if
    // the user-payload has a custom alignment
    auto backOffset = reinterpret_cast<UserPayloadOffset_t*>(userPayloadAddress - sizeof(UserPayloadOffset_t));
    return reinterpret_cast<ChunkHeader*>(userPayloadAddress - *backOffset);
}

（无 user-header 且默认对齐时，m_userPayloadOffset 字段本身恰好位于 payload 前 4 字节，兼作 back-offset，不占额外空间。）

所需 chunk 大小由 ChunkSettings::create(userPayloadSize, userPayloadAlignment, userHeaderSize, userHeaderAlignment) 预先算出（mepoo/chunk_settings.hpp），并校验对齐是 2 的幂、user-header 对齐不超过 ChunkHeader 对齐等。

5. ChunkManagement 与 SharedChunk：跨进程引用计数

5.1 ChunkManagement

引用计数不放在 chunk 里，而是放在管理段的专用池中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
struct ChunkManagement
{
    using base_t = ChunkHeader;
    using referenceCounterBase_t = uint64_t;
    using referenceCounter_t = std::atomic<referenceCounterBase_t>;

    ChunkManagement(const cxx::not_null<base_t*> chunkHeader,
                    const cxx::not_null<MemPool*> mempool,
                    const cxx::not_null<MemPool*> chunkManagementPool) noexcept;

    iox::rp::RelativePointer<base_t> m_chunkHeader;
    referenceCounter_t m_referenceCounter{1U};
    /// @todo optimization: check if this can be replaced by an offset relative to the this pointer
    iox::rp::RelativePointer<MemPool> m_mempool;
    iox::rp::RelativePointer<MemPool> m_chunkManagementPool;
};

跨进程安全的三个要素：

  1. std::atomic<uint64_t> 计数器位于所有参与进程都以读写方式映射的管理段中，x86/ARM 上无锁原子（fetch_add/fetch_sub）跨进程有效——原子性由 CPU 缓存一致性保证，与进程无关，只要求同一物理内存。
  2. 所有指针都是 RelativePointer（segmentId + offset），各进程映射基址不同也能解引用。
  3. ChunkManagement 自身从 m_chunkManagementPool 分配，总数 = 所有 mempool 的 chunk 总数（generateChunkManagementPool，见下节），保证永不耗尽。

数据段可能对订阅方只读，但计数器在管理段——这就是”读者也能增减引用计数”的机制。

5.2 SharedChunk

SharedChunk 是持有 ChunkManagement* 的进程内智能句柄（类似 shared_ptr，但本身非线程安全，跨线程需各持副本）：

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

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

计数归零时把 chunk 还给数据池、把 ChunkManagement 还给管理池，两次 freeChunk 都是无锁的 LoFFLi push。

派生形态 ShmSafeUnmanagedChunk（shm_safe_unmanaged_chunk.hpp）把 ChunkManagement 的 RelativePointer 压缩进 64 位 RelativePointerData，可单周期写入、不会 torn write——队列（ChunkQueueData）、history、UsedChunkList 中存的都是它，即使应用写到一半崩溃，RouDi 也能读到一致值并完成回收。它「持有引用但不自动管理」：releaseToSharedChunk() 不增计数移交所有权，cloneToSharedChunk() 增计数复制。

6. MemoryManager：找最小合适 bucket

MemoryManager 管理一个段内的所有 MemPool。配置阶段要求 bucket 按 chunkSize 严格递增加入，最后生成 ChunkManagement 池并封锁再添加：

1
2
3
4
5
6
7
8
9
10
11
void MemoryManager::configureMemoryManager(const MePooConfig& mePooConfig,
                                           posix::Allocator& managementAllocator,
                                           posix::Allocator& chunkMemoryAllocator) noexcept
{
    for (auto entry : mePooConfig.m_mempoolConfig)
    {
        addMemPool(managementAllocator, chunkMemoryAllocator, entry.m_size, entry.m_chunkCount);
    }

    generateChunkManagementPool(managementAllocator);
}

分配策略是线性扫描找第一个（即最小的）能装下 requiredChunkSize 的 bucket——因为有序，第一个命中即 best-fit；bucket 数很少（≤32），线性扫描比树查找更缓存友好：

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

三种错误：NO_MEMPOOLS_AVAILABLE（没配任何池）、NO_MEMPOOL_FOR_REQUESTED_CHUNK_SIZE（消息比最大 bucket 还大，FATAL）、MEMPOOL_OUT_OF_CHUNKS（命中的 bucket 耗尽，MODERATE——注意不会降级到更大的 bucket，直接报错）。

内存需求计算（用于确定 shm 大小）：

  • requiredChunkMemorySize = Σ chunkCount × (size + sizeof(ChunkHeader))，按 8 字节对齐；
  • requiredManagementMemorySize = Σ LoFFLi 索引表 + 总 chunk 数 × sizeof(ChunkManagement) + ChunkManagement 池自己的 LoFFLi。

7. TypedMemPool

TypedMemPool<T>（internal/mepoo/typed_mem_pool.hpp/.inl）是给单一类型 T 用的独立小内存池：自带一个数据 MemPool（chunkSize 按 sizeof(T)+alignof(T)+ChunkHeader 算出）和一个配对的 ChunkManagement 池，createObject(args...) 在 chunk 里 placement-new 构造 T 并返回引用计数的 SharedPointer<T>：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
template <typename T>
template <typename... Targs>
inline cxx::expected<SharedPointer<T>, TypedMemPoolError> TypedMemPool<T>::createObject(Targs&&... args) noexcept
{
    auto chunkManagement = acquireChunkManagementPointer();
    if (chunkManagement.has_error())
    {
        return cxx::error<TypedMemPoolError>(chunkManagement.get_error());
    }

    auto newObject = SharedPointer<T>::create(SharedChunk(*chunkManagement), std::forward<Targs>(args)...);
    ...
    return cxx::success<SharedPointer<T>>(newObject.value());
}

它不属于用户数据通路（pub/sub 用的是 MemoryManager），主要供 RouDi 在管理内存里放置需要共享的、带生命周期管理的对象。

8. 应用侧 attach 流程（PoshRuntime）

RouDi 与应用通过 IPC 通道（Unix Domain Socket / mq）握手，应用侧的 SharedMemoryUser 完成映射：

1
2
3
4
5
6
7
8
9
10
11
应用: PoshRuntime::initRuntime("app")
  → IpcRuntimeInterface: 发送 REG(runtimeName, pid, uid, ...) 给 RouDi
  → RouDi: 记录进程，回复 REG_ACK(shmTopicSize, segmentManagerOffset,
                                  timestamp, mgmtSegmentId)
  → 应用: SharedMemoryUser 构造
       ① shm_open("iceoryx_mgmt", OPEN_EXISTING, READ_WRITE) 映射管理段
          registerPtr(mgmtSegmentId, 基址)                     ← relative pointer 登记
       ② 由 (segmentId, offset) 解出 SegmentManager*
       ③ segmentManager->getSegmentMappings(当前进程用户)
       ④ 逐段 shm_open(组名, OPEN_EXISTING, 可写?RW:RO) 映射数据段
          registerPtr(每段的 segmentId, 基址)

REG_ACK 的解析在 ipc_runtime_interface.cpp（waitForRegAck 读出 shm 大小、SegmentManager 偏移与 segmentId），映射在：

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
SharedMemoryUser::SharedMemoryUser(const size_t topicSize,
                                   const uint64_t segmentId,
                                   const rp::BaseRelativePointer::offset_t segmentManagerAddressOffset) noexcept
{
    // create and map the already existing shared memory region
    posix::SharedMemoryObject::create(roudi::SHM_NAME,
                                      topicSize,
                                      posix::AccessMode::READ_WRITE,
                                      posix::OpenMode::OPEN_EXISTING,
                                      posix::SharedMemoryObject::NO_ADDRESS_HINT)
        .and_then([this, segmentId, segmentManagerAddressOffset](auto& sharedMemoryObject) {
            rp::BaseRelativePointer::registerPtr(
                segmentId, sharedMemoryObject.getBaseAddress(), sharedMemoryObject.getSizeInBytes());
            ...
            this->openDataSegments(segmentId, segmentManagerAddressOffset);
            ...
        })
        .or_else([](auto&) { errorHandler(Error::kPOSH__SHM_APP_MAPP_ERR); });
}

void SharedMemoryUser::openDataSegments(const uint64_t segmentId,
                                        const rp::BaseRelativePointer::offset_t segmentManagerAddressOffset) noexcept
{
    auto ptr = rp::BaseRelativePointer::getPtr(segmentId, segmentManagerAddressOffset);
    auto segmentManager = reinterpret_cast<mepoo::SegmentManager<>*>(ptr);

    auto segmentMapping = segmentManager->getSegmentMappings(posix::PosixUser::getUserOfCurrentProcess());

各进程映射基址可以不同（NO_ADDRESS_HINT，由 OS 决定）——一切跨进程数据结构都只存 (segmentId, offset)，这是 iceoryx 不要求相同虚拟地址的关键。

9. chunk 生命周期状态图

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
                 ┌────────────────────────────────────────────────┐
                 │              MemPool free-list (LoFFLi)         │
                 └────────────────────────────────────────────────┘
                     │ getChunk()  (pop index)          ▲
                     ▼                                  │ refCount 1→0:
           [ALLOCATED / LOANED]                         │ freeChunk()
publisher loan(): placement-new ChunkHeader,            │
ChunkManagement(refCount=1)，登记进                      │
publisher 的 UsedChunkList                               │
                     │ send()                            │
                     ▼                                   │
              [DELIVERED]                                │
ChunkDistributor 推入每个订阅者队列（每队列 +1 ref），    │
写入 history（+1 ref），发送方 UsedChunkList 移除（-1）  │
     │                         │                         │
     ▼                         ▼                         │
[IN QUEUE]               [IN HISTORY]────释放/被挤出─────┤
订阅者 take():                                           │
出队进入其 UsedChunkList                                  │
     │                                                   │
     ▼                                                   │
  [HELD BY SUBSCRIBER] ──release()（Sample 析构）────────┘

队列溢出(DISCARD_OLDEST_DATA)：被挤出的旧 chunk 直接 -1 ref
应用崩溃：RouDi 遍历该进程 port 的 UsedChunkList/队列/history 强制归还

引用计数的每个持有点：发送方 UsedChunkList、每个订阅者队列槽、distributor history、订阅者 UsedChunkList、以及进程内的每个 SharedChunk/Sample 副本。

10. 配置与容量参数

10.1 默认 MePooConfig

未提供配置时使用 MePooConfig::setDefaults()：

1
2
3
4
5
6
7
8
9
10
11
12
MePooConfig& MePooConfig::setDefaults() noexcept
{
    m_mempoolConfig.push_back({128, 10000});
    m_mempoolConfig.push_back({1024, 5000});
    m_mempoolConfig.push_back({1024 * 16, 1000});
    m_mempoolConfig.push_back({1024 * 128, 200});
    m_mempoolConfig.push_back({1024 * 512, 50});
    m_mempoolConfig.push_back({1024 * 1024, 30});
    m_mempoolConfig.push_back({1024 * 1024 * 4, 10});

    return *this;
}
bucket（payload 字节） chunk 数 数据内存约
128 10000 1.5 MB
1 KB 5000 5.2 MB
16 KB 1000 16 MB
128 KB 200 25.6 MB
512 KB 50 25.6 MB
1 MB 30 30 MB
4 MB 10 40 MB

（实际每 chunk 另加 32B ChunkHeader；总计约 144 MB 数据段。）MePooConfig::optimize() 会排序并合并相同 size 的条目。

10.2 RouDi TOML 配置

RouDi 可用 -c 指定 TOML（示例 iceoryx_posh/etc/iceoryx/roudi_config_example.toml）：

1
2
3
4
5
6
7
8
9
10
11
12
13
[general]
version = 1

[[segment]]              # 可带 reader = "组名" / writer = "组名"，缺省为启动 RouDi 的用户组

[[segment.mempool]]
size = 128
count = 10000

[[segment.mempool]]
size = 1024
count = 5000
# ... 与默认配置相同的 7 档

每个 [[segment]] 对应一个 SegmentConfig::SegmentEntry（reader 组、writer 组、一份 MePooConfig），见 mepoo/segment_config.hpp。

10.3 编译期容量常量

来自 iceoryx_posh_types.hpp 与 cmake/IceoryxPoshDeployment.cmake（可用 -DIOX_MAX_* 覆盖）：

常量 默认值 说明
MAX_NUMBER_OF_MEMPOOLS 32 每段最多 bucket 数
MAX_SHM_SEGMENTS 100 最多共享内存段数
IOX_MAX_PUBLISHERS 512 全系统 publisher port 上限
IOX_MAX_SUBSCRIBERS 1024 全系统 subscriber port 上限
IOX_MAX_CHUNKS_ALLOCATED_PER_PUBLISHER_SIMULTANEOUSLY 8 单 publisher 同时 loan 的 chunk 数
IOX_MAX_PUBLISHER_HISTORY 16 history 容量上限
IOX_MAX_CHUNKS_HELD_PER_SUBSCRIBER_SIMULTANEOUSLY 256 单 subscriber 同时持有的 chunk 数（=队列容量上限）
MAX_PROCESS_NUMBER 300 可注册进程数
CHUNK_DEFAULT_USER_PAYLOAD_ALIGNMENT 8 默认 payload 对齐

下一篇：03-popo发布订阅与通信原语.md

文章互动

阅读 --

留言

0 条留言

正在加载留言…