【iceoryx】 Freelist

iceoryx 里的 freeList

FreeList的init()
publisher getChunk()
- void* MemPool::getChunk() ：
- 看一下freelist的pop
MemPool::freeChunk(const void* chunk)

在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_sizem_nextFreeIndex[i] = i + 1;}}
}

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;}}

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_minFreem_usedChunks.fetch_add(1U, std::memory_order_relaxed);adjustMinFree();return m_rawMemory + l_index * m_chunkSize;
}

关键在 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 Sizeif (oldHead.indexToNextFreeIndex >= m_size){return false;}// NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) upper limit of index set by m_sizenewHead.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 neededindex = 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'sstd::atomic_thread_fence(std::memory_order_release);return true;
}

先看一下 LoFFLi 类的成员：

class LoFFLi
{private:struct alignas(8) Node{Index_t indexToNextFreeIndex;uint32_t abaCounter;};std::atomic<Node> m_head{{0U, 1U}};
}

它有一个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 checkstd::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 capacityif (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 capacitym_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;
}

假定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了。