高并发内存池

2025/5/14 编程实战高并发内存管理

内存池

内存池是一种预分配内存并进行重复利用的技术，通过减少频繁的动态内存分配和释放操作，从而提高程序的运行效率，内存池通常预先分配一块大的内存区域，将其划分很多小块。需要时直接从这块区域中分配，而不是直接调用系统的动态分配函数，可以有效的减少系统调用带来的开销

整体框架

三层缓存构成

thread cache

线程缓存是每个线程独有的，每个线程独享一个cache，这也是这个内存池高效的地方，
thread cache是一个哈希桶拉链法+自由链表的结构

#pragma once

#include "Common.h"

class ThreadCache
{
public:
	// 申请和释放内存对象
	void* Allocate(size_t size);
	void Deallocate(void* ptr, size_t size);

	// 从中心缓存获取对象
	void* FetchFromCentralCache(size_t index, size_t size);

	// 释放对象时，链表过长时，回收内存回到中心缓存
	void ListTooLong(FreeList& list, size_t size);
private:
	FreeList _freeLists[NFREELIST];
};

// TLS thread local storage
static _declspec(thread) ThreadCache* pTLSThreadCache = nullptr;

因为trheadcache使用的是哈希桶结构
不分段的话若全是按8字节分，则会需要32768个桶，但但是每个桶都是16Byte则会差生较多的内存碎片。因此这里采用了不同段的内存使用不同的内存对齐规则，既控制了桶的数量不会太多，又整体将内存碎片浪费控制在10%左右

内存对齐的函数的接口

//管理对齐和映射等关系
class SizeClass
{
public:
	//获取向上对齐后的字节数
	static inline size_t RoundUp(size_t bytes);
	//获取对应哈希桶的下标
	static inline size_t Index(size_t bytes);
};

对齐映射规则

static inline size_t _RoundUp(size_t bytes, size_t alignNum)
	{
		return ((bytes + alignNum - 1) & ~(alignNum - 1));
	}
	//内联函数：调用频繁，因此写成内联函数
	//向上对齐
	static inline size_t RoundUp(size_t size)
	{
		if (size <= 128)
		{
			return _RoundUp(size, 8);
		}
		else if (size <= 1024)
		{
			return _RoundUp(size, 16);
		}
		else if (size <= 8*1024)
		{
			return _RoundUp(size, 128);
		}
		else if (size <= 64*1024)
		{
			return _RoundUp(size, 1024);
		}
		else if (size <= 256 * 1024)
		{
			return _RoundUp(size, 8*1024);
		}
		else//size > 256 * 1024 byte 
		{
			return _RoundUp(size, 1<<PAGE_SHIFT);
		}
	}

	static inline size_t _Index(size_t bytes, size_t align_shift)
	{
		return ((bytes + (1 << align_shift) - 1) >> align_shift) - 1;
	}
	//计算映射的哪一个自由链表桶中
	static inline size_t Index(size_t bytes)
	{
		assert(bytes <= MAX_BYTES);
		static int group_array[4] = { 16, 56, 56, 56 };
		if (bytes <= 128)
		{
			return _Index(bytes, 3);
		}
		else if (bytes <= 1024)
		{
			return _Index(bytes - 128, 4) + group_array[0];
		}
		else if (bytes <= 8 * 1024)
		{
			return _Index(bytes - 1024, 7) + group_array[0] + group_array[1];
		}
		else if (bytes <= 64 * 1024)
		{
			return _Index(bytes - 8 * 1024, 10) + group_array[0] + group_array[1] + group_array[2];
		}
		else if (bytes <= 256 * 1024)
		{
			return _Index(bytes - 64 * 1024, 13) + group_array[0] + group_array[1] + group_array[2] + group_array[3];
		}
		else
		{
			assert(false);
			return -1;
		}
	}

central cache

中心缓存所有线程共享，当我们的thread cache没有内存块时会向central cache申请，central cahce也是一个哈希桶的结构，不过它挂载的是Span List链表结构，Span List中则有一个个相同大小的内存块按照哈希桶的映射后，通过双向链表的形式挂在相应的Span上，

Central Cache本质是由一个哈希映射的span对象自由双向链表构成
为了保证全局只有唯一的Central Cache,这个类因此可以被设计称单例模式(这里使用的是饿汉模式)
饿汉模式：构造函数私有，对象设为静态私有。拷贝构造和赋值重载设为delete（防拷贝）


#pragma once

#include "Common.h"

// 单例模式
class CentralCache
{
public:
	static CentralCache* GetInstance()
	{
		return &_sInst;
	}
	// 获取一个非空的span
	Span* GetOneSpan(SpanList& list, size_t byte_size);
	// 从中心缓存获取一定数量的对象给thread cache
	size_t FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size);
	// 将一定数量的对象释放到span跨度
	void ReleaseListToSpans(void* start, size_t byte_size);
private:
	SpanList _spanLists[NFREELIST];
private:
	CentralCache()
	{}
	CentralCache(const CentralCache&) = delete;
	static CentralCache _sInst;
};

span的结构

//管理多个连续页的大块内存跨度结构
struct Span
{
	PAGE_ID _pageId = 0;//大块内存起始页的页号
	size_t _n = 0;//页的数量

	Span* _next = nullptr;//双向链表的结构
	Span* _prev = nullptr;

	size_t _objSize = 0;//切好的小对象的大小
	size_t _usecount = 0;//切好的小块内存，被分配给thread cache的计数
	void* _freeList = nullptr;//切好的小块内存的自由链表

	bool _isUse = false;//是否在被使用
};

使用双向链表来管理，下面是双向链表的结构
spanList

//带头双向循环链表
class SpanList
{
public:
	SpanList()
	{
		_head = new Span;
		_head->_next = _head;
		_head->_prev = _head;
	}

	void Insert(Span* pos, Span* newSpan)
	{
		assert(pos);
		assert(newSpan);
		Span* prev = pos->_prev;
		prev->_next = newSpan;
		newSpan->_prev = prev;
		newSpan->_next = pos;
		pos->_prev = newSpan;
	}

	Span* Begin()
	{
		return _head->_next;
	}

	Span* End()
	{
		return _head;
	}

	bool Empty()
	{
		return _head->_next == _head;
	}
	void PushFront(Span* span)
	{
		Insert(Begin(), span);
	}

	Span* PopFront()
	{
		Span* front = _head->_next;
		Erase(front);
		return front;
	}
	void Erase(Span* pos)
	{
		assert(pos);
		assert(pos != _head);

		Span* prev = pos->_prev;
		Span* next = pos->_next;
		prev->_next = next;
		next->_prev = prev;
	}
public:
	std::mutex _mtx;//桶锁
private:
	Span* _head;
};

page cache

存储的内存以page为单位存储以及分配的，当central cache没有内存对象时，我们从page cache分配出一定数量的page，并切割成定长大小的小块内存，分配给central cache，当一个span的几个跨度页的对象都收回以后，page cache会回收central cache满足条件的span对象，合并相邻的页，组成更大的页，这样缓解内存碎片问题

page的任务是
检查central cache申请的相应位置是否合法，如果没有合适的span,需要从较大的span中分裂两个
如果找到链表的最后都没有合适的span，则在系统中使用VirtualAlloc申请一个较大的span。
pagecache中的spanlist与前两个采用不同的内存对齐原则，直接就是1page,2page,3page
框架

#pragma once

#include "Common.h"
#include "ObjectPool.h"
#include "PageMap.h"

class PageCache
{
public:
	static PageCache* GetInstance()
	{
		return &_sInst;
	}

	// 获取从对象到span的映射
	Span* MapObjectToSpan(void* obj);

	// 释放空闲span回到Pagecache，并合并相邻的span
	void ReleaseSpanToPageCache(Span* span);

	// 获取一个K页的span
	Span* NewSpan(size_t k);

	std::mutex _pageMtx;
private:
	SpanList _spanLists[NPAGES];
	ObjectPool<Span> _spanPool;

	//std::unordered_map<PAGE_ID, Span*> _idSpanMap;
	//std::map<PAGE_ID, Span*> _idSpanMap;
	TCMalloc_PageMap1<32 - PAGE_SHIFT> _idSpanMap;

	PageCache()
	{}
	PageCache(const PageCache&) = delete;


	static PageCache _sInst;
};

LOADING

高并发内存池

内存池

整体框架

thread cache

central cache

page cache