当前位置：首页 » 游戏攻略 » iceoryx源码阅读（三）——共享内存通信（一）

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

Gamer • 2024-05-14 08:50:12 • 游戏攻略

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<Info, CAPACITY> m_info;
    uint64_t m_maxRegistered{0U};
};

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

2.3 构造函数

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

职责：

RelativePointer
实例的构造。

入参：

ptr
：普通指针。

template <typename T>
inline RelativePointer<T>::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<T>::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<T>::offset_t RelativePointer<T>::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<T>::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<T>::get() const noexcept
{
    return static_cast<ptr_t>(computeRawPtr());
}

template <typename T>
inline T* RelativePointer<T>::computeRawPtr() const noexcept
{
    return getPtr(segment_id_t{m_id}, m_offset);
}

template <typename T>
inline T* RelativePointer<T>::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<mepoo::ShmSafeUnmanagedChunk, MAX_CAPACITY> 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<mepoo::ChunkManagement> ptr{chunk.release()};
        auto id = ptr.getId();
        auto offset = ptr.getOffset();
        m_chunkManagement =
            memory::RelativePointerData(static_cast<memory::RelativePointerData::identifier_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<uint64_t>(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<mepoo::ChunkManagement>(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 小结

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

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