【iceoryx】 Freelist_loffli freelist

在iceoryx中，publisher 获取一块chunk，填充数据，发布；供subscriber 读取。这一系列操作要用到内存管理中的freeList。

FreeList的init()

void LoFFLi::init(cxx::not_null<Index_t*> freeIndicesMemory, const uint32_t capacity) noexcept
{
    cxx::Expects(capacity > 0 && "A capacity of 0 is not supported!");
    constexpr uint32_t INTERNALLY_RESERVED_INDICES{1U};
    cxx::Expects(capacity < (std::numeric_limits<Index_t>::max() - INTERNALLY_RESERVED_INDICES)
                 && "Requested capacityexceeds limits!");
    cxx::Expects(m_head.is_lock_free() && "std::atomic<LoFFLi::Node> must be lock-free!");

    m_nextFreeIndex = freeIndicesMemory;
    m_size = capacity;
    m_invalidIndex = m_size + 1;

    if (m_nextFreeIndex != nullptr)
    {
        for (uint32_t i = 0; i < m_size + 1; i++)
        {
            // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) upper limit of index is set by m_size
            m_nextFreeIndex[i] = i + 1;
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

init看起来相对简单，拿 [[segment.mempool]] ， size = 128 ， count = 10000 举例，所做的操作如下：
1、m_invalidIndex = 10001；
2、m_size = 10000；
3、m_nextFreeIndex[10000] 依此填充 1，2，3，4…10001

publisher getChunk()

追踪代码可知，最终调用MemoryManager::getChunk(const ChunkSettings& chunkSettings)
核心代码：

   for (auto& memPool : m_memPoolVector)
    {
        uint32_t chunkSizeOfMemPool = memPool.getChunkSize();
        if (chunkSizeOfMemPool >= requiredChunkSize)
        {
            chunk = memPool.getChunk();
            memPoolPointer = &memPool;
            aquiredChunkSize = chunkSizeOfMemPool;
            break;
        }
    }
1
2
3
4
5
6
7
8
9
10
11

void* MemPool::getChunk() ：

void* MemPool::getChunk() noexcept
{
    uint32_t l_index{0U};
    if (!m_freeIndices.pop(l_index))
    {
        std::cerr << "Mempool [m_chunkSize = " << m_chunkSize << ", numberOfChunks = " << m_numberOfChunks
                  << ", used_chunks = " << m_usedChunks << " ] has no more space left" << std::endl;
        return nullptr;
    }

    /// @todo: verify that m_usedChunk is not changed during adjustMInFree
    ///         without changing m_minFree
    m_usedChunks.fetch_add(1U, std::memory_order_relaxed);
    adjustMinFree();

    return m_rawMemory + l_index * m_chunkSize;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

关键在 m_freeIndices.pop(l_index) 这里传入的 l_index 。

看一下freelist的pop

bool LoFFLi::pop(Index_t& index) noexcept
{
    Node oldHead = m_head.load(std::memory_order_acquire);
    Node newHead = oldHead;

    do
    {
        // we are empty if next points to an element with index of Size
        if (oldHead.indexToNextFreeIndex >= m_size)
        {
            return false;
        }

        // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) upper limit of index set by m_size
        newHead.indexToNextFreeIndex = m_nextFreeIndex[oldHead.indexToNextFreeIndex]; //<1>
        newHead.abaCounter += 1;
    } while (!m_head.compare_exchange_weak(oldHead, newHead, std::memory_order_acq_rel, std::memory_order_acquire)); //<2>

    /// comes from outside, is not shared and therefore no synchronization is needed
    index = oldHead.indexToNextFreeIndex;  //<3>
    /// @todo what if interrupted here an another thread guesses the index and
    ///         calls push
    /// @brief murphy case: m_nextFreeIndex does not require any synchronization since it
    ///         either is used by the same thread in push or it is given to another
    ///         thread which performs the cleanup and during this process a synchronization
    ///         is required
    // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic)
    m_nextFreeIndex[index] = m_invalidIndex; //<4>

    /// we need to synchronize m_nextFreeIndex with push so that we can perform a validation
    /// check right before push to avoid double free's
    std::atomic_thread_fence(std::memory_order_release);

    return true;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

先看一下 LoFFLi 类的成员：

class LoFFLi
{
  private:
    struct alignas(8) Node
    {
        Index_t indexToNextFreeIndex;
        uint32_t abaCounter;
    };
	std::atomic<Node> m_head{{0U, 1U}};
}
1
2
3
4
5
6
7
8
9
10

它有一个m_head{0,1};
因此 <1>处 newHead.indexToNextFreeIndex = m_nextFreeIndex[0]；
由init()知，m_nextFreeIndex[0] = 1；
<2> 如果 m_head 当前的值 == oldHead ，则将 m_head = newHead，并返回 true;
反之 ，oldHead = m_head ，返回false。
这里主要是担心 执行 do 下面的操作时被打断，考虑以下情况：
此时 index = 0；若此时有一线程抢占CPU并将 index = 0 的chunk取走，则此时 m_head = {1，2};
oldHead = {1，2}，newHead.indexToNextFreeIndex = m_nextFreeIndex[1]，去get index = 1 的 chunk。

后面操作 index = 0 , m_nextFreeIndex[0] = 10001; 然后将index 返回。

MemPool::freeChunk(const void* chunk)

当需要free该chunk时，调用 m_freeIndices.push(index)：

bool LoFFLi::push(const Index_t index) noexcept
{
    /// we synchronize with m_nextFreeIndex in pop to perform the validity check
    std::atomic_thread_fence(std::memory_order_release);

    /// we want to avoid double free's therefore we check if the index was acquired
    /// in pop and the push argument "index" is valid
    // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) index is limited by capacity
    if (index >= m_size || m_nextFreeIndex[index] != m_invalidIndex) //<1>
    {
        return false;
    }

    Node oldHead = m_head.load(std::memory_order_acquire);
    Node newHead = oldHead;

    do
    {
        // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) index is limited by capacity
        m_nextFreeIndex[index] = oldHead.indexToNextFreeIndex; //<2>
        newHead.indexToNextFreeIndex = index; //<3>
        newHead.abaCounter += 1;
    } while (!m_head.compare_exchange_weak(oldHead, newHead, std::memory_order_acq_rel, std::memory_order_acquire));

    return true;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

假定index = 0
<1>处判断 传入的 m_nextFreeIndex[0] 是否 = m_invalidIndex，因为pop时已置为 m_invalidIndex；
<2> oldHead.indexToNextFreeIndex = (上面 push 时) 1 ，因此 m_nextFreeIndex[0] = 1；
<3> newHead.indexToNextFreeIndex =0 ，即 m_head ={0，3} ；
这样就把 index = 0 的chunk free了。