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

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

news/2024/9/23 15:21:39

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<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的数据成员）。