iceoryx源码阅读(三)——共享内存通信(一)

iceoryx · 浏览次数 : 0

小编点评

**Code Analysis:** The provided code outlines a data structure called `ShmSafeUnmanagedChunk` for managing shared memory in a safe and efficient manner. **Key Concepts:** * **SharedMemoryChunk:** A data structure used to manage shared memory, which allows multiple processes to access the same memory region with proper synchronization. * **RelativePointerData:** A structure that represents a pointer in a shared memory segment, containing both the ID and offset of the pointer. * **Constructor (`ShmSafeUnmanagedChunk` constructor)`:** * Takes a `SharedChunk` object as input. * Extracts the ID and offset from the input `chunk`. * Creates a `RelativePointerData` object with these values. * Stores the `RelativePointerData` object in the `m_chunkManagement` member variable. **Constructor (`RelativePointerData` constructor)**: * Initializes the `m_idAndOffset` member variable with the ID and offset extracted from the input. * Checks if the ID and offset fall within valid ranges (48 bits for ID and 16 bits for offset). * If outside the valid range, sets the `m_idAndOffset` to `nullptr`. **`releaseToSharedChunk` method:** * Checks if the `m_chunkManagement` is logically null (indicating no shared memory management). * If management is valid, reset the `m_chunkManagement` and return a new `SharedChunk` object with an empty `ChunkManagement` pointer. **Usage:** The `ShmSafeUnmanagedChunk` constructor can be used to create an instance of the structure, passing a `SharedChunk` object as input. This instance can then be used to manage shared memory for multiple processes. **Additional Notes:** * `RelativePointerData` has two member variables: `m_idAndOffset` (integer representing ID and offset) and `id` (integer representing ID only). * The constructor checks for valid ID and offset values to ensure that they fall within the valid range of the shared memory management. * `releaseToSharedChunk` method allows the underlying `ChunkManagement` to be reset when no longer needed.

正文

0 导引

1 整体通信结构

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

image

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

2 RelativePointer

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

2.1 原理

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

image

不同进程映射的区域不同,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,存放共享内存的起始地址和结束地址。这里,我们贴了两个注册指针的函数——registerPtrWithIdregisterPtr——分别用于打开共享内存和创建共享内存时调用。

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: 构造函数,调用成员函数searchIdgetOffset计算该指针在注册表中的索引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,如下图所示:

image

据此,我们来看求取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_RANGEOFFSET_RANGE分别为后16为和48位为1的数字,取名为ID_MASKOFFSET_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 小结

本文介绍基于共享内存通信的主要数据结构,下文我们将介绍数据发送函数和接收函数的实现。

与iceoryx源码阅读(三)——共享内存通信(一)相似的内容: