iceoryx源码阅读（三）——共享内存通信（一）

目录

• 0 导引
• 1 整体通信结构
• 2 RelativePointer
  • 2.1 原理
  • 2.2 PointerRepository
  • 2.3 构造函数
  • 2.4 get函数
• 3 ShmSafeUnmanagedChunk
  • 3.1 队列数据
  • 3.2 RelativePointerData
  • 3.3 构造函数
  • 3.4 releaseToSharedChunk
• 4 小结

0 导引

• iceoryx源码阅读（一）——全局概览

• iceoryx源码阅读（二）——共享内存管理

• iceoryx源码阅读（三）——共享内存管理（一）

• iceoryx源码阅读（四）——共享内存通信（二）

• iceoryx源码阅读（五）——共享内存通信（三）

• iceoryx源码阅读（六）——共享内存创建

• iceoryx源码阅读（七）——服务发现机制

• iceoryx源码阅读（八）——IPC通信机制

1 整体通信结构

订阅-发布结构实现一对多的通信模式，消息发布者可以将消息推送到多个订阅者。基于共享内存的订阅-发布通信结构如下图所示：

每一对订阅者和发布者之间通过队列联系，队列元素为发送数据的描述。发送者往队列中推入描述，订阅者取出描述，据此从共享内存获取真正的消息数据。队列中为什么不直接存放消息数据本身呢？原因一则是消息的长度是不确定的，二则是一对多通信结果下，直接将消息存放在队列中更浪费空间。

2 RelativePointer

上节说到队列存放消息存放位置的描述，可以是地址吗？

2.1 原理

使用共享内存内存前，需要映射到进程的虚拟地址空间，如下图所示：

不同进程映射的区域不同，iceoryx使用数字唯一标识共享内存。实际上，iceoryx每个应用进程维护一张注册表，保存各个共享内存的起止地址，这里的数字就是共享内存在注册表中的索引。为了定位某个Chunk，还需要该Chunk相对共享内存首地址的偏移量。共享内存索引和偏移就定义了RelativePointer——用于定位共享内存的指定位置，相关代码如下所示：

template <typename T>
class RelativePointer final
{
  public:
    using ptr_t = T*;
    using offset_t = std::uintptr_t;
 
    explicit RelativePointer(ptr_t const ptr) noexcept;
    T* computeRawPtr() const noexcept;
 
  private:
    segment_id_underlying_t m_id{NULL_POINTER_ID};
    offset_t m_offset{NULL_POINTER_OFFSET};
};

上述代码中，除了共享内存索引和偏移外，还加了两个函数：

• 构造函数，通过普通指针构造RelativePointer对象。
• 根据RelativePointer获取其所代表的普通指针。

2.2 PointerRepository

上节我们引入了注册表的概念，了解了其作用，本节具体看看其实现。

constexpr uint64_t MAX_POINTER_REPO_CAPACITY{10000U};
template <typename id_t, typename ptr_t, uint64_t CAPACITY = MAX_POINTER_REPO_CAPACITY>
class PointerRepository final
{
  private:
    struct Info
    {
        ptr_t basePtr{nullptr};
        ptr_t endPtr{nullptr};
    };
 
  public:
    bool registerPtrWithId(const id_t id, const ptr_t ptr, const uint64_t size) noexcept;
    cxx::optional<id_t> registerPtr(const ptr_t ptr, const uint64_t size = 0U) noexcept;
 
  private:
    iox::cxx::vector m_info;
    uint64_t m_maxRegistered{0U};
};

m_info就是注册表，元素类型为Info，存放共享内存的起始地址和结束地址。这里，我们贴了两个注册指针的函数——registerPtrWithId和registerPtr——分别用于打开共享内存和创建共享内存时调用。

2.3 构造函数

构造函数根据普通指针构造相对指针实例，其代码实现如下：

职责：

RelativePointer实例的构造。

入参：

ptr：普通指针。

template <typename T>
inline RelativePointer::RelativePointer(ptr_t const ptr) noexcept
    : RelativePointer([this, ptr]() noexcept -> RelativePointer {
        const segment_id_t id{this->searchId(ptr)};
        const offset_t offset{this->getOffset(id, ptr)};
        return RelativePointer{offset, id};
    }())
{
}
 
template <typename T>
inline segment_id_underlying_t RelativePointer::searchId(ptr_t const ptr) noexcept
{
    if (ptr == nullptr)
    {
        return NULL_POINTER_ID;
    }
    return getRepository().searchId(ptr);
}
 
template <typename id_t, typename ptr_t, uint64_t CAPACITY>
inline id_t PointerRepository<id_t, ptr_t, CAPACITY>::searchId(const ptr_t ptr) const noexcept
{
    for (id_t id{1U}; id <= m_maxRegistered; ++id)
    {
        if ((ptr >= m_info[id].basePtr) && (ptr <= m_info[id].endPtr))
        {
            return id;
        }
    }
 
    return RAW_POINTER_BEHAVIOUR_ID;
}
 
template <typename T>
inline typename RelativePointer::offset_t RelativePointer::getOffset(const segment_id_t id,
                                                                           ptr_t const ptr) noexcept
{
    if (static_cast<segment_id_underlying_t>(id) == NULL_POINTER_ID)
    {
        return NULL_POINTER_OFFSET;
    }
    const auto* const basePtr = getBasePtr(id);
    return reinterpret_cast<offset_t>(ptr) - reinterpret_cast<offset_t>(basePtr);
}
 
template <typename T>
inline T* RelativePointer::getBasePtr(const segment_id_t id) noexcept
{
    return static_cast<ptr_t>(getRepository().getBasePtr(static_cast<segment_id_underlying_t>(id)));
}
 
template <typename id_t, typename ptr_t, uint64_t CAPACITY>
inline ptr_t PointerRepository<id_t, ptr_t, CAPACITY>::getBasePtr(const id_t id) const noexcept
{
    if ((id <= MAX_ID) && (id >= MIN_ID))
    {
        return m_info[id].basePtr;
    }
 
    return nullptr;
}

逐段代码分析：

• LINE 01 ~ LINE 09： 构造函数，调用成员函数searchId和getOffset计算该指针在注册表中的索引id和偏移，以此初始化两个成员。

• LINE 11 ～ LINE 33： 这部分就是遍历注册表中所有共享内存，找到包含给定地址的共享内存区域的，返回其id。

• LINE 35 ～ LINE 62： 从注册表中找出指定id共享内存首地址，入参指针减去首地址，计算得到偏移。

2.4 get函数

职责：

获取RelativePointer实例对应的普通指针。

返回：
普通指针。

template <typename T>
inline T* RelativePointer::get() const noexcept
{
    return static_cast<ptr_t>(computeRawPtr());
}
 
template <typename T>
inline T* RelativePointer::computeRawPtr() const noexcept
{
    return getPtr(segment_id_t{m_id}, m_offset);
}
 
template <typename T>
inline T* RelativePointer::getPtr(const segment_id_t id, const offset_t offset) noexcept
{
    if (offset == NULL_POINTER_OFFSET)
    {
        return nullptr;
    }
    const auto* const basePtr = getBasePtr(id);
    return reinterpret_cast<ptr_t>(offset + reinterpret_cast<offset_t>(basePtr));
}

整体代码分析：

上面代码逻辑和2.3节类似，通过id从注册表中获取共享内存首地址，加上偏移量得到普通指针。

3 ShmSafeUnmanagedChunk

上一篇文章中，我们介绍了SharedChunk，用于管理共享内存。本节将介绍ShmSafeUnmanagedChunk，用于基于共享内存的通信。可以认为是从两个角度描述Chunk。

3.1 队列数据

第1节中的队列中存放的描述数据结构就是ShmSafeUnmanagedChunk，具体代码（去除和本节无关的代码）如下：

struct ChunkQueueData : public LockingPolicy
{
    cxx::VariantQueue m_queue;
};

3.2 RelativePointerData

ShmSafeUnmanagedChunk只有唯一的成员变量m_chunkManagement，其类型为RelativePointerData：

class ShmSafeUnmanagedChunk
{
  private:
    memory::RelativePointerData m_chunkManagement;
};

RelativePointerData的成员就是一个整数，如下：

class RelativePointerData
{
  private:
    uint64_t m_idAndOffset{LOGICAL_NULLPTR};
};

但是第2节我们知道，描述消息数据在共享内存中的位置，我们需要注册表中的索引id和偏移offset，一个整数怎么够呢？实际上，这个整数按位分成两部分，前48位表示offset，后16位表示id，如下图所示：

据此，我们来看求取id和offset的实现：

using identifier_t = uint16_t;
static constexpr uint64_t ID_BIT_SIZE{16U};
static constexpr identifier_t ID_RANGE{std::numeric_limits<identifier_t>::max()};
static constexpr offset_t OFFSET_RANGE{(1ULL << 48U) - 1U};
 
RelativePointerData::identifier_t RelativePointerData::id() const noexcept
{
    return static_cast<identifier_t>(m_idAndOffset & ID_RANGE);
}
 
RelativePointerData::offset_t RelativePointerData::offset() const noexcept
{
    return (m_idAndOffset >> ID_BIT_SIZE) & OFFSET_RANGE;
}

都是一些位运算，其中ID_RANGE和OFFSET_RANGE分别为后16为和48位为1的数字，取名为ID_MASK和OFFSET_MASK（掩码）更合适。

3.3 构造函数

发送数据的核心就是将SharedChunk转化为ShmSafeUnmanagedChunk，推入队列容器中。这就是ShmSafeUnmanagedChunk的构造函数的职责。

职责：

使用SharedChunk实例构造ShmSafeUnmanagedChunk实例。

入参：

ShmSafeUnmanagedChunk::ShmSafeUnmanagedChunk(mepoo::SharedChunk chunk) noexcept
{
    if (chunk)
    {
        memory::RelativePointer ptr{chunk.release()};
        auto id = ptr.getId();
        auto offset = ptr.getOffset();
        m_chunkManagement =
            memory::RelativePointerData(static_castidentifier_t>(id), offset);
    }
}

整体代码分析：

上述代码就是使用第2节中介绍的构造函数，根据普通指针构造RelativePointer，然后得到id和offset，以此构造RelativePointerData：

static constexpr identifier_t MAX_VALID_ID{ID_RANGE - 1U};
static constexpr offset_t MAX_VALID_OFFSET{OFFSET_RANGE - 1U};
 
constexpr RelativePointerData::RelativePointerData(identifier_t id, offset_t offset) noexcept
    : m_idAndOffset(static_cast(id) | (offset << ID_BIT_SIZE))
{
    if ((id > MAX_VALID_ID) || (offset > MAX_VALID_OFFSET))
    {
        m_idAndOffset = LOGICAL_NULLPTR;
    }
}

结合3.2节对RelativePointerData的介绍，上述构造函数是显然的。

3.4 releaseToSharedChunk

接收端需要将ShmSafeUnmanagedChunk转为SharedChunk，这就是releaseToSharedChunk的职责。

职责：

通过ShmSafeUnmanagedChunk构造SharedChunk实例。

返回：

SharedChunk实例。

SharedChunk ShmSafeUnmanagedChunk::releaseToSharedChunk() noexcept
{
    if (m_chunkManagement.isLogicalNullptr())
    {
        return SharedChunk();
    }
    auto chunkMgmt = memory::RelativePointer(m_chunkManagement.offset(),
                                                                     memory::segment_id_t{m_chunkManagement.id()});
    m_chunkManagement.reset();
    return SharedChunk(chunkMgmt.get());
}

根据id和offset构造RelativePointer实例，然后通过2.4节介绍的get方法获得指向ChunkManagement指针，据此构造SharedChunk实例（SharedChunk唯一的成员数据就是ChunkManagement指针，见：SharedChunk的数据成员）。

4 小结

本文介绍基于共享内存通信的主要数据结构，具体如下图所示：

下文我们将介绍数据发送函数和接收函数的实现。