iceoryx源码阅读（四）——共享内存通信（二）
iceoryx源码阅读（五）——共享内存通信（三）
iceoryx源码阅读（六）——共享内存创建
iceoryx源码阅读（七）——服务发现机制
订阅-发布结构实现一对多的通信模式，消息发布者可以将消息推送到多个订阅者。基于共享内存的订阅-发布通信结构如下图所示：
每一对订阅者和发布者之间通过队列联系，队列元素为发送数据的描述。发送者往队列中推入描述，订阅者取出描述，据此从共享内存获取真正的消息数据。队列中为什么不直接存放消息数据本身呢？原因一则是消息的长度是不确定的，二则是一对多通信结果下，直接将消息存放在队列中更浪费空间。
上节说到队列存放消息存放位置的描述，可以是地址吗？
使用共享内存内存前，需要映射到进程的虚拟地址空间，如下图所示：
不同进程映射的区域不同，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获取其所代表的普通指针。上节我们引入了注册表的概念，了解了其作用，本节具体看看其实现。
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——分别用于打开共享内存和创建共享内存时调用。
构造函数根据普通指针构造相对指针实例，其代码实现如下：
职责：
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共享内存首地址，入参指针减去首地址，计算得到偏移。
职责：
获取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从注册表中获取共享内存首地址，加上偏移量得到普通指针。
上一篇文章中，我们介绍了SharedChunk，用于管理共享内存。本节将介绍ShmSafeUnmanagedChunk，用于基于共享内存的通信。可以认为是从两个角度描述Chunk。
第1节中的队列中存放的描述数据结构就是ShmSafeUnmanagedChunk，具体代码（去除和本节无关的代码）如下：
struct ChunkQueueData : public LockingPolicy
{
cxx::VariantQueue<mepoo::ShmSafeUnmanagedChunk, MAX_CAPACITY> m_queue;
};
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（掩码）更合适。
发送数据的核心就是将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的介绍，上述构造函数是显然的。
接收端需要将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的数据成员）。
本文介绍基于共享内存通信的主要数据结构，下文我们将介绍数据发送函数和接收函数的实现。