C++标准项目---高并发内存池

大筒木老辈子

941人浏览 · 2025-10-17 16:18:18

大筒木老辈子 · 2025-10-17 16:18:18 发布

1. 项目介绍

该项目的原型是google的一个开源项目tcmalloc，tcmalloc全称Thread-Caching Malloc，即线程缓存的malloc，实现了高效的多线程内存管理，用于替代内存分配相关的函数（例如：malloc/free）。

博主的这个项目是把tcmalloc最核心的框架简化后拿出来，模拟实现出一个自己的高并发内存池，目的就是学习tcamlloc的精华。

相比于tcmalloc，在该项目的具体实现当中，为了项目的可读性使用了不少的C++11新特性以及STL的容器，所以实际上我们的项目并没有完全脱离开malloc/free。

有兴趣的读者可以在详细了解下文之后，尝试使用数组以及项目中封装的系统调用来进行改造，使其完全脱离malloc/free。

项目源码：https://gitee.com/da-guan-mu-lao-sheng/c-learning/tree/master/MemoryPool/MemoryPool

1.1 什么是内存池

内存池是一种预先分配一块连续内存空间，并在程序运行中从该空间内高效分配、回收小块内存的管理技术，核心目的是避免频繁向操作系统申请和释放内存。

1.1.1 内存池的基本工作原理

内存池的运作流程可分为 3 个核心步骤：

初始化（创建池）：程序启动时，向操作系统申请一块固定大小的连续内存，作为内存池的 “池空间”。
内存分配：当程序需要内存时，不直接调用系统接口，而是从内存池中找到一块 “空闲” 的、大小匹配的内存块，标记为 “已使用” 并返回给程序。
内存回收：当程序释放内存时，不将内存还给操作系统，而是将对应的内存块标记为 “空闲”，放回内存池，供后续分配重复使用。

1.1.2 内存池的作用

减少内存碎片：频繁分配 / 回收不同大小的内存，会导致内存空间出现大量无法利用的 “碎片”，内存池通过统一管理连续空间，大幅降低碎片产生。
降低系统开销：内存池每次申请内存都会申请一大块的内存，此后始终维护这块内存，减少了与系统交互的次数。
提升分配效率：内存池预先划分好内存块结构，分配时直接返回现成的块，回收时只需标记块为 “可用”，无需复杂的查找和计算。

1.1.3 常见应用场景

内存池在对性能和稳定性要求高的场景中广泛使用：

高频内存操作场景：如服务器处理海量请求、游戏引擎渲染（频繁创建 / 销毁对象）。
资源受限场景：如嵌入式系统（内存小、不允许频繁系统调用）。
实时性要求场景：如工业控制、金融交易系统（需避免内存分配耗时波动）。

1.2 项目架构

现代很多的开发环境都是多核多线程，在这种场景下，必然存在激烈的锁竞争问题。

实际上malloc本身就是一个内存池，但是它是一个通用的内存池，虽然在任何环境下都可以使用，但“通用”也意味着它的效率在特殊场景下的效率并不顶尖。

而我们项目的原型tcmalloc就是在多线程、高并发的场景下更胜一筹的内存管理方案。因此，在项目的实现中我们需要考虑如下三方面的问题：

多线程环境下，锁竞争问题。
性能问题。
内存碎片问题。

基于上面的考虑，tcmalloc设计了三层缓存结构：

Thread Cache：线程缓存。每个线程独占一个 Thread Cache，当需要分配内存时就向 Thread Chache 进行申请。当 Thread Cache 内部缓存的内存块不足，而无法满足请求时，才向公共内存池进行申请。Thread Cache 可以减少线程对公共内存池的互斥访问。
Central Cache：中心缓存。公共内存池，向上负责切分内存，为 Thread Cache 分配内存并及时回收（确保某个 Thread Cache 中不会积攒太多空闲内存，达到负载均衡）；向下可向Page Cache 申请以页为单位申请内存，并在某块申请的内存被全部回收时，归还给 Page Cache 进行合并。
Page Cache：页缓存。负责以页为单位的大块内存的申请、切分与合并。

2 具体实现

2.1 组织小块内存

内存池在申请到一大块内存之后如何进行管理呢？显然，我们的基本逻辑是：将大块内存切分为小块内存，当用户申请内存时就将小块内存分配出去；当用户释放内存时，我们就将小块内存组织起来，并在合适的时机合并为大块内存。

首先我们要解决的问题，就是小块内存如何组织起来。内存块的数量是未知的，所以线性表并不是一个好选择，我们只能选择链表。

传统的链表结点至少需要两个字段：val和next，也就是两个指针，在这里我们还需要记录内存块的大小。当小块内存过小时（例如4字节），为其专门设置一个链表结点就显得十分浪费空间。

因此，用传统的链表来管理小块内存是不合用的，我们注意到控制信息（val、next，size）只有在小块内存没有被分配出去时才有意义。

所以，我们可以将小块内存的空间利用起来，将小块内存本身的空间看作是一个结点，并将next指针放到小块内存头部进行存放：

这时候就有聪明的读者要问了：小块内存的大小不够一个指针怎么办呢？

那还不简单吗，我给你补够不就完了。首先大小小于指针类型的内存块用到的本来就不多，再者浪费一两个字节也是完全可以接受的。

这样的链表还存在一个问题，那就是同一个链表只能管理固定大小的内存块，因为我们并没有记录内存块的大小。

但是这正符合我们的需求：将不同大小的内存块分开管理才好进行查找和合并。

这就是所谓的“自由链表”，“自由”指的是内存块空闲，该链表用于管理被用户释放回来的空闲内存块。

有了上面的认知之后，我们可以尝试来设计一个固定大小的内存池，即只分配特定大小的内存块。

同时，我们不考虑多线程的安全问题，仅仅做一个简单的尝试：

template<typename T>
class FixedSizePool
{
private:
	// 单次分配的数据块
	union block_t
	{
		block_t* next;
		T memory;
	};
public:
	FixedSizePool()
	{
		// 去除内碎片
		_memorySize -= (_memorySize % sizeof(block_t));
	}
	T* New()
	{
		// 优先使用回收的内存块
		if (_freeList != nullptr)
		{
			block_t* block = _freeList;
			_freeList = _freeList->next;
			return (T*)block;
		}

		// 无可分配内存，重新申请一块
		if (_begin == _end)
		{
			_begin = (char*)malloc(_memorySize);
			if (_begin == nullptr)
				throw std::bad_alloc();
			_end = _begin + _memorySize;
		}

		// 从可分配内存中分配一块内存
		block_t* block = (block_t*)_begin;
		_begin += sizeof(block_t);
		// 定位new初始化T对象
		new(block)T;
		return (T*)block;
	}

	void Delete(T* blockPtr)
	{
		block_t* block = (block_t*)blockPtr;
		// Free掉的空间本质未被释放，需要显式调用析构
		block->memory.~T();
		// 将释放掉的内存块头插到_freeList
		block->next = _freeList;
		_freeList = block;
	}
private:
	char* _begin = nullptr; // 可分配内存起始
	char* _end = nullptr; // 可分配内存结尾
	size_t _memorySize = 128 * 1024; // 一次申请的大块内存的字节数
	block_t* _freeList = nullptr; // 回收的内存块链表，头部字节作为指针
};

2.2 Thread Cache

2.2.1 内存块对齐

首先要明确，我们的内存池是可以分配可变长度的内存块的，所以我们需要多个链表来管理不同大小的内存块。

我们将这些链表放到一个哈希表中存放，每一个链表称为一个哈希桶。

显然，我们不可能将每一种大小都管理起来，我们只能选取某些大小来进行管理。当用户申请 size 大小的内存时，我们将size向上对齐，分配给用户足以容纳 size 的最小内存块。

在这个过程中必定存在着空间的浪费，但是我们可以将空间的浪费控制在一个可接受的范围内，例如 10% 左右。

假设 size 在两个大小梯度 n1，n2 之间（n1 < size <= n2），那么浪费率的计算公式如下： $WasteRate = (n_2 - size) / size$
当 size = n1 + 1 时，浪费率达到最大：
$MaxWasteRate = (n_2 - n_1 - 1) / (n_1 + 1)$

在保持 MaxWasteRate 不变的情况下，随着 $n_1$ 的增大，可接受的对齐值 $n_2 - n_1$ 也就越大。

据此，我们可以设计出如下的对齐方案：

大小区间	对齐方式	哈希桶个数
[1B，128B]	8B	16
(128B，1KB]	16B	56
(1KB，8KB]	128B	56
(8KB，64KB]	1KB	56
(64KB，256KB]	8KB	24

我们可以设计一个SianHandler类来帮助我们按照上面的规则对齐 size ，并计算出对应哈希桶的下标：

// 块大小处理
class SizeHandler
{
private:
	// 初始化_bucket_num
	static size_t InitBucketNum()
	{
		size_t bucket_num = 0;
		for (auto& range : _align_rule)
		{
			bucket_num += range.num;
		}
		return bucket_num;
	}
	// 初始化_max_block_size
	static size_t InitMaxSize()
	{
		return _align_rule.back().ceiling;
	}
public:
	// 确保size在规定范围内
	static inline void CheckSize(size_t size)
	{
		assert(0 < size && size <= _max_size);
	}

	// 使申请的内存大小对齐桶大小		
	static size_t AlignSize(size_t size)
	{
		// 让size与2的exponent次幂对齐
		static auto Align = [](size_t size, size_t exponent) {
			return ((size + (1 << exponent) - 1) & ~((1 << exponent) - 1));
		};
		
		CheckSize(size);
		
		size_t aligned_size = 0;
		for (auto& range : _align_rule)
		{
			if (size <= range.ceiling)
			{
				aligned_size = Align(size, range.exponent);
				break;
			}
		}
		return aligned_size;
	}

	// 根据块大小计算桶下标
	static size_t GetIndex(size_t size)
	{
		// 获取区间下标
		static auto RangeIndex = [](size_t size, size_t exponent) {
			return ((size + (1 << exponent) - 1) >> exponent) - 1;
		};

		CheckSize(size);

		size_t index = 0, prev_ceiling = 0;
		for (auto& range : _align_rule)
		{
			if (size <= range.ceiling)
			{
				index += RangeIndex(size - prev_ceiling, range.exponent);
				break;
			}
			prev_ceiling = range.ceiling;
			index += range.num;
		}
		return index;
	}

	static size_t BucketNum()
	{
		return _bucket_num;
	}

	static size_t MaxSize()
	{
		return _max_size;
	}

private:
	struct range_data
	{
		size_t ceiling; // 区间最大值
		size_t exponent; // 对齐数 = 1 << exponent
		size_t num; // 区间桶数
	};
	static std::vector<range_data> _align_rule; // 对齐规则
	static size_t _bucket_num; // 桶的个数
	static size_t _max_size; // 内存块最大大小
};

std::vector<SizeHandler::range_data> SizeHandler::_align_rule = {
	{128, 3, 16}, {1024, 4, 56}, {8 * 1024, 7, 56}, {64 * 1024, 10, 56}, {256 * 1024, 13, 24}
};

size_t SizeHandler::_bucket_num = SizeHandler::InitBucketNum(); // 桶的个数
size_t SizeHandler::_max_size = SizeHandler::InitMaxSize(); // 内存块最大大小

2.2.2 FreeList

为了方便对自由链表进行操作、管理，我们还是先将其封装为一个类：

// 空闲块链表
class FreeList
{
public:
	static void*& Next(void* block)
	{
		return *(void**)block;
	}
	void Push(void* block)
	{
		Next(block) = _head;
		_head = block;
		_size++;
		if (_tail == nullptr)
			_tail = block;
	}
	void PushBack(void* block)
	{
		Next(block) = nullptr;
		if (_tail == nullptr)
			Push(block);
		else
		{
			Next(_tail) = block;
			_tail = block;
			_size++;
		}
	}
	size_t PushRange(void* begin, void* end)
	{
		Next(end) = _head;
		_head = begin;
		size_t count = 0;
		while (begin != Next(end))
		{
			begin = Next(begin);
			count++;
		}
		_size += count;
		if (_tail == nullptr)
			_tail = end;
		return count;
	}
	void* Pop()
	{
		if (Empty())
			return nullptr;
		void* block = _head;
		_head = Next(_head);
		_size--;
		if (_head == nullptr)
			_tail = nullptr;
		return block;
	}
	// 弹出至多n个结点，返回实际弹出数量
	size_t PopRange(void*& begin, void*& end, size_t n)
	{
		assert(!Empty());
		begin = end = _head;
		size_t count = 1;
		for (count = 1; count < n && Next(end); count++)
		{
			end = Next(end);
		}
		_head = Next(end);
		_size -= count;
		Next(end) = nullptr;
		if (_head == nullptr)
			_tail = nullptr;
		return count;
	}
	bool Empty()
	{
		return _head == nullptr;
	}
	size_t Size()
	{
		return _size;
	}
	// FreeList中的连接关系手动清理，因为连接关系是保存在内存块上的，FreeList中没有保存控制信息
	// 这些内存块在下次被申请到时，连接关系会自动被重构
	void Clear()
	{
		_head = _tail = nullptr;
		_size = 0;
	}
private:
	void* _head = nullptr;
	void* _tail = nullptr;
	size_t _size = 0; // 链表中结点个数
};

2.2.3 ThreadCache类的定义

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

private:
	// 从CentralCache获取内存块
	void* FetchFromCentralCache(size_t index, size_t size);

	// 将内存块还给CentralCache
	void ReturnToCentralCache(size_t index, size_t size);

	std::vector<FreeList> _list_bucket;
	std::vector<size_t> _apply_num; // 一次向CentralCache申请的内存块的个数（慢启动）
};

关于 _apply_num 字段，我们在下文再做详细解释。

2.2.4 TLS无锁访问

如何让一个线程访问到独属于自己的Thread Cache呢？

我们希望内存池最终提供给线程的接口的使用方式是和malloc一样的，这样我们才能无缝替换掉以往项目中的malloc。

这样一来，Thread Cache就必须要在全局范围内可被访问，这就需要一个全局的数据结构来管理，并在接口内部确定哪一个Thread Cache是属于当前线程的。

但是，这样做存在两个缺点：

不符合线程隔离的要求，一个线程可以访问到另一个线程的Thread Cache；
各个线程在访问同一个全局的数据结构时需要加锁，极大降低了效率。

有没有什么办法能让某个Thread Cache对应的线程全局可访问，而其他线程却访问不到呢？

有，我们可以使用线程本地存储（Thread-Local Storage，简称 TLS）—— 这是多线程编程中用于实现 “线程私有全局变量” 的核心技术。

简单来说，TLS 允许你定义一个 “全局变量”，但这个变量在每个线程中都有独立的副本：

线程 A 对该变量的修改，只会影响线程 A 自己的副本；
线程 B 访问的是自己的副本，完全不受线程 A 的影响；
从线程内部看，它就像一个普通的全局变量，但本质上是线程私有的。

在C/C++中，我们有两个用于定义TLS变量的的关键字：

C11 标准引入 _Thread_local 关键字（GCC 等编译器支持）；
C++11 及以上支持 thread_local 关键字，更通用。

static thread_local ThreadCache* pTLSThreadCache = nullptr;

static void* ConcurrentAlloc(size_t size)
{
	// 通过TLS 每个线程无锁的获取自己的专属的ThreadCache对象
	if (pTLSThreadCache == nullptr)
	{
		pTLSThreadCache = new ThreadCache;
	}

	// cout << std::this_thread::get_id() << ":" << pTLSThreadCache << endl;

	return pTLSThreadCache->Allocate(size);
}

static void ConcurrentFree(void* ptr)
{
	assert(pTLSThreadCache);

	pTLSThreadCache->Deallocate(ptr);
}

这样一来，每个线程在使用上面的两个接口（对应malloc/free）进行内存的申请释放时，使用的就可以无锁地访问到属于自己的Thread Cache，并且线程间完全隔离。

2.2.5 慢启动方式获取内存块

当Thread Cache缓存的内存块不够满足用户需求时，就需要向Central Cache申请新的内存块。

但是一次申请多少个内存块合适呢？

假如一次申请一块满足用户需求的内存块：可能在有内存块被释放之前，又出现新的请求，此时又需要向Central Cache进行申请，效率低下。
假如一次申请多块满足用户需求的内存块：用户可能长期、甚至一直都不对该类内存块进行新的请求，这时多申请的内存块就浪费了，随着程序的运行可能又需要还给Central Cache。

解决问题的核心在于预测用户的需求量。我们可以借鉴TCP协议拥塞控制的核心机制---慢启动。

即，随着用户向Central Cache的申请次数增多，我们每次向Central Cache的申请量也逐渐增多，因为越频繁的申请可能预示着用户对该类内存块的需求量越大。

我们前面定义的 _apply_num 就是为每一个哈希桶一次申请的量进行计数，最开始一次申请一个内存块，每申请一次，该计数值加1。

size_t apply_num = _apply_num[index];
if (apply_num < SizeHandler::SlowStartCeiling(size))
	_apply_num[index]++;

void* begin = nullptr;
void* end = nullptr;
size_t actual_n = CentralCache::GetInstance().NewBlocks(begin, end, apply_num, size);

然而，我们不能让这个数字无限制地增长，否则可能会导致一次申请过量的内存。所以我们需要给这个值设置一个上限，但是上限值取多少呢？

我们希望一次向Central Cache申请的内存总量不要超过单个内存块的最大大小：

// 慢启动一次分配块数上限
static size_t SlowStartCeiling(size_t size)
{
	size_t ceiling = _max_size / size;
	if (ceiling < 2) ceiling = 2;
	if (ceiling > 512) ceiling = 512;
	return ceiling;
}

2.3 Central Cache

2.3.1 跨度（Span）

我们前面说，Central Cache向PageCache申请的是以页为单位的大块内存。在这里，我们将这样的大块内存定义为一个Span：

struct Span
{
	page_id_t page_id = 0; // 起始页号
	size_t num = 0; // 包含的页数

	// 使用带头双向循环链表管理
	Span* prev = this;
	Span* next = this;

	size_t use_count = 0; // 使用计数，为0时代表该Span的内存块全部回收
	FreeList list; // 切分好的内存块链表

	bool is_use = false; // 是否被PageCache分配给了CentralCache
};

2.3.1.1 为什么需要Span

为什么需要Span呢？或者说为什么大块内存需要以页为单位呢？

首先，操作系统本身就是以页为单位分配内存的。

直接调用操作系统接口（如 Windows 的VirtualAlloc、Linux 的mmap）申请内存，必须指定 “页大小的整数倍” 的大小，否则操作系统会自动向上对齐到最近的页大小（例如申请 5KB，会实际分配 8KB，即 2 个 4KB 页）。而系统调用接口返回的地址也一定是某个页的首地址。

所以Page Cache每次向系统申请的大块内存就是以页为单位的，以页为单位分配给Central Cache可以最大程度地利用申请来的连续空间。

其次，以页为单位也方便Page Cache进行内存块的合并。Page Cache可能会将一个大块的内存按页切分成多个Span交付给Central Cache，当Central Cache返还Span给Page Cache时，Page Cache就可以根据Span的页号和页数来将相邻的Span进行合并，以此减少碎片。

2.3.1.2 页操作宏

既然要以页为单位进行操作，我们就可以将常用的页操作定义为宏，方便使用。

#ifdef _WIN64
	typedef unsigned long long page_id_t;
#elif _WIN32
	typedef size_t page_id_t;
#else
	// linux
#endif

// 页大小，2的指数形式，4K
static const size_t page_exponent = 12;

// 地址转页号
#define PAGE_ID(addr) ((page_id_t)(addr) >> page_exponent)
// 页号转地址
#define ADDR(page_id) ((void*)((size_t)(page_id) << page_exponent))
// 页数转大小
#define PAGE_SIZE(page_num) ((size_t)(page_num) << page_exponent)
// 容纳size所需要的page数
#define PAGE_NUM(size) (((size_t)(size) + ((1 << page_exponent) - 1)) >> page_exponent)

2.3.2 SpanList

Central Cache也有一个哈希表，每个桶中存放的都是和Thread Cache相同大小的内存块。但是为了将Span完整地还给Page Cache，Central Cache的哈希桶中是以Span为单位存放内存块的（内存块存放在Span的list字段中）。

因为对齐规则完全相同，所以也可以使用SizeHandler类中的方法来计算下标。

所以，我们需要一个SpanList来管理span，这里我们选取的数据结构是带头双向循环链表（上图只是示意图）：

// CentralChache哈希桶，Span带头双向循环链表
class SpanList
{
public:
	void Insert(Span* pos, Span* new_span)
	{
		assert(pos && new_span);
		Span* prev = pos->prev;
		prev->next = new_span;
		new_span->prev = prev;
		new_span->next = pos;
		pos->prev = new_span;
	}
	void Erase(Span* pos)
	{
		assert(pos && pos != &_head);
		pos->prev->next = pos->next;
		pos->next->prev = pos->prev;
		
		pos->prev = pos;
		pos->next = pos;
	}
	void PushFront(Span* new_span)
	{
		Insert(Begin(), new_span);
	}
	Span* PopFront()
	{
		Span* front = _head.next;
		Erase(front);
		return front;
	}
	Span* Begin()
	{
		return _head.next;
	}
	Span* End()
	{
		return &_head;
	}
	bool Empty()
	{
		return _head.next == &_head;
	}

private:
	Span _head;
};

2.3.3 CentralCache类的定义

Central Cache是所有线程共用的，所以这里我们使用了“单例模式”来进行设计。

同时，为了减少各个线程互斥访问Central Cache的冲突，我们以桶为单位进行加锁。

class CentralCache
{
public:
	// 获取CentralCache单例
	static CentralCache& GetInstance()
	{
		// 懒汉方式实现
		// 由于CentralCache的实例化依赖于SizeHandler::BucketNum()
		// 所以使用饿汉方式的话，CentralCache实例可能会在SizeHandler::_bucket_num之前完成实例化，导致出错
		// 否则就需要将_bucket_num定义为全局变量或是宏
		static CentralCache instance;
		return instance;
	}

	// 获取大小为size的n个内存块
	size_t NewBlocks(void*& begin, void*& end, size_t n, size_t size);

	// 从ThreadCache回收内存块
	void FreeBlock(void* begin, size_t size);

private:
	CentralCache() 
		: _list_bucket(SizeHandler::BucketNum())
		, _bucket_mutex(SizeHandler::BucketNum())
	{}
	CentralCache(const CentralCache&) = delete;

	// 获取一个有空闲块的Span，若无则从PageChache获取并插入到List
	Span* GetOneSpan(size_t index, size_t n, size_t size);

	// 向PageCache获取大小大于等于size的Span
	Span* FetchFromPageCache(size_t index, size_t n, size_t size);

	std::vector<SpanList> _list_bucket; // 哈希桶
	std::vector<std::mutex> _bucket_mutex; // 桶锁
};

2.3.4 什么时候将Span还给Page Cache

在NewBlocks函数当中，Span每分出去n个内存块，其use_count就会加n：

size_t actual_n = span->list.PopRange(begin, end, n);
span->use_count += actual_n;

在FreeBlocks函数当中，Span每回收一个内存块，其use_count就会--。当减到0时，就说明该Span的所有内存块都被回收了，它是一个完整的页对齐的大块内存，此时就可以将其交还给Page Cache尝试合并：

if (--(span->use_count) == 0)
{
	_list_bucket[index].Erase(span);

	_bucket_mutex[index].unlock();
	PageCache::GetInstance().FreeSpan(span);
	_bucket_mutex[index].lock();
}

2.3.5 通过页号查找Span

在FreeBlocks中，我们如何通过block的地址来查找Span呢？

在不借助其他数据结构的情况下，我们可以根据size找到block所在的桶。然后遍历桶，进而查找到这个block所在的Span。

但问题是，这样的效率太低下了，显然时不可取的。其实，我们可以记录一下页号与Span的映射关系，以便于我们查找。

std::unordered_map<page_id_t, Span*> _pageid_span; // 页号与span的对应关系，方便查找

实际上，在Page Cache中也存在这样的查找需求（页合并），并且所有的Span都是由Page Cache申请或切分出来的，所以我们实际上是将这样的数据结构定义在了Page Cache中，并开放了响应的接口以方便上层使用：

Span* PageCache::FindSpan(page_id_t page_id)
{
	// 性能瓶颈，每次查找都需要对PageCache加锁
	std::lock_guard<std::recursive_mutex> guard(_mtx);
	auto span = _pageid_span.find(page_id);
	if (span != _pageid_span.end())
	{
		return span->second;
	}
	return nullptr;
}

2.4 Page Cache

2.4.1 向系统申请/释放内存

既然是要替代malloc/free，我们自然是要直接使用系统调用来申请内存：

// 直接去堆上按页申请空间（n页）
inline static void* SystemAlloc(size_t n)
{
#ifdef _WIN32
	void* ptr = VirtualAlloc(0, PAGE_SIZE(n), MEM_COMMIT | MEM_RESERVE, PAGE_READWRITE);
#else
	// linux下brk mmap等
#endif
	if (ptr == nullptr)
		throw std::bad_alloc();
	return ptr;
}

// 释放空间
inline static void SystemFree(void* ptr)
{
#ifdef _WIN32
	VirtualFree(ptr, 0, MEM_RELEASE);
#else
	// sbrk unmmap等
#endif
}

我们之前写的固定大小内存池也就可以借此脱离malloc/free了：

// 无可分配内存，重新申请一块
if (_begin == _end)
{
	_begin = (char*)SystemAlloc(PAGE_NUM(_memorySize));
	if (_begin == nullptr)
	    throw std::bad_alloc();
	_end = _begin + _memorySize;
}

2.4.2 Span的申请与释放

显然，Span的申请与释放十分符合我们固定大小内存池的使用场景：

// 定长内存池，用于申请Span
FixedSizePool<Span> _span_pool;

2.4.3 Span的切分

我们定义Page Cache一次向系统申请的量为128个页。Central Cache需要多少页，我们就切分多少页的Span给Central Cache。

但无论是最开始未被切分的Span，还是之后被切分留下来的Span，我们都需要依据页面数来将这些Span分别管理起来，这样才方便我们之后继续切分。

所以，与Thread Cache和Central Cache不同，Page Cache需要128个哈希桶（实际定义129个），下标从1 ~ 128，分别代表1 ~ 128页的Span：

切分的方法也很简单，只需要新申请出一个Span，然后分别修改被切分的Span和新Span的page_id，page_num即可：

2.4.4 Span的合并

当Page Cache从Central Cache中回收一个Span时，我们需要检查其前后相邻的Span是否存在，以及是否可用，如果存在且可用，我们就将其进行合并。

相邻的Span是什么意思？就是其包含的页相邻。

2.4.4.1 存在性检查

要检查Span的前一个Span是否存在，就是检查page_id - 1所对应的Span是否存在。
要检查Span的后一个Span是否存在，就是检查page_id + page_num - 1所对应的Span是否存在。

实际上，只有被分配给Central Cache的Span需要将其所有的页都映射到该Span，因为Central Cache会将自己获得的Span拆开来用，其中的每个block都需要能找到自己所属的Span。

// 将分配出去的span的每一页注册到_pageid_span
for (int i = 0; i < span->num; i++)
{
	_pageid_span[span->page_id + i] = span;
}

// 要分配出去的span的页数是大于目标的，将剩下来的页存放到新span
if (k > n)
{
	Span* left = _span_pool.New();
	left->page_id = span->page_id + span->num;
	left->num = k - n;
	// 插入桶
	_list_bucket[left->num].PushFront(left);
	// 首尾进行注册
	_pageid_span[left->page_id] = left;
	_pageid_span[left->page_id + left->num - 1] = left;
}

而对于未分配给Central Cache的Span，我们只需要将其首尾页映射到自身即可，因为我们只需要检查其首尾是否存在就可以与回收的Span合并。

2.4.4.2 可用性检查

Central Cache和Page Cache都通过FindSpan来进行查找，我们如何知道查找出来的Span当前是属于Central Cache，还是Page Cache呢？

此时，我们就需要用到Span的is_used字段来判断了。当Page Cache把Span分配给Central Cache时，将is_used置为true，当Page Cache回收Span时，将is_used置为false。

如果通过FindSpan查找到了前一页或后一页对应的Span，并且其is_used为false，我们就可以将新回收的Span与改Span进行合并。

// 向前合并
Span* prev = FindSpan(span->page_id - 1);
while (prev && !prev->is_use)
{
	// 判断合并后的大小是否合法
	if (span->num + prev->num > _page_bucket_num)
		break;
	span->page_id = prev->page_id;
	span->num += prev->num;
	
	_list_bucket[prev->num].Erase(prev);
	// delete prev;
	_span_pool.Delete(prev);

	prev = FindSpan(span->page_id - 1);
}

// 向后合并
Span* next = FindSpan(span->page_id + span->num);
while (next && !next->is_use)
{
	if (span->num + next->num > _page_bucket_num)
		break;
	span->num += next->num;

	_list_bucket[next->num].Erase(next);
	// delete next;
	_span_pool.Delete(next);

	next = FindSpan(span->page_id + span->num);
}

2.4.5 Page Cache类的定义

同样地，Page Cache是需要互斥访问的，但是我们不能以桶锁的方式来加锁，而需要对整个Page Cache加锁。

因为在Page Cache的各个方法当中，我们的某次操作通常要涉及到多个桶，若采用桶锁，我们就需要频繁加锁解锁，甚至可能导致死锁。

同时，对于FindSpan也需要互斥调用，因为该方法的底层为unordered_map，在插入数据的过程当中，是可能会发生扩容的。这就导致我们在并发访问时可能访问到错误的数据，甚至导致程序崩溃。

#pragma once
#include "Common.h"
#include "FixedSizePool.h"

class PageCache
{
public:
	static PageCache& GetInstance()
	{
		static PageCache instance;
		return instance;
	}
	// 获取一个新Span
	Span* NewSpan(size_t n);
	// 释放一个Span
	void FreeSpan(Span* span);
	// 通过页号查找Span
	Span* FindSpan(page_id_t page_id);

private:
	PageCache() : _list_bucket(_page_bucket_num + 1)
	{ }
	PageCache(const PageCache&) = delete;

	// Span不够时向系统申请新的Span
	Span* CreateSpan(size_t n);

	std::vector<SpanList> _list_bucket;
	std::recursive_mutex _mtx;
	std::unordered_map<page_id_t, Span*> _pageid_span; // 页号与span的对应关系，方便查找

	// 定长内存池，用于申请Span
	FixedSizePool<Span> _span_pool;

	// PageCache桶数量：1页 到 128页
	static const size_t _page_bucket_num = 128;
};

这里我们使用递归锁是因为NewSpan内部的设计，读者可以参考源代码。

2.5 大内存申请问题

当用户要申请大于SizeHandler::_max_size(256KB)的内存时，我们的内存池总不能直接报错吧？那么，如何处理这样的大内存申请呢？

事实上，虽然Thread Cache和Central Cache能处理的内存申请最大值就是_max_size，但是Page Cache能处理的内存申请最大值为128页（128 * 4KB）。

对于大于256KB的申请，Thread Cache就不再通过正常流程分配内存了，而是直接向Page Cache进行申请（不进行管理，释放时发现内存块大小过大，也直接转交给Page Cache）：

// 申请内存对象
void* ThreadCache::Allocate(size_t size)
{
	// 申请的内存过大，不在ThreadCache和CentralCache的职责之内
	if (size > SizeHandler::MaxSize())
	{
		// 直接向PageCache申请页对齐的内存
		size_t page_num = PAGE_NUM(size);
		Span* span = PageCache::GetInstance().NewSpan(page_num);
		void* addr = ADDR(span->page_id);
		return addr;
	}

	// ...
}

// 释放内存对象
void ThreadCache::Deallocate(void* ptr)
{
	assert(ptr);

	Span* span = PageCache::GetInstance().FindSpan(PAGE_ID(ptr));
	size_t size = PAGE_SIZE(span->num);

	if (size > SizeHandler::MaxSize())
	{
		PageCache::GetInstance().FreeSpan(span);
		return;
	}

	// 。。。
}

对于大于512KB的申请，Page Cache也无法处理，此时直接向系统申请（同样地，在释放时发现内存块过大，也直接还给系统）：

Span* PageCache::NewSpan(size_t n)
{
	assert(n > 0);
	// 页数过大，PageCache不在PageCache的管理范围内
	// 向系统申请空间
	if (n > _page_bucket_num)
	{
		Span* new_span = CreateSpan(n);
		// _pageid_span[new_span->page_id] = new_span;
		_page_map.set(new_span->page_id, new_span);
		return new_span;
	}

	// ...
}

void PageCache::FreeSpan(Span* span)
{
	// 回收的Span超过128页，未被管理，直接释放
	if (span->num > _page_bucket_num)
	{
		void* addr = ADDR(span->page_id);
		SystemFree(addr);
		// delete span;
		_span_pool.Delete(span);
		return;
	}

	// ...
}

2.5 使用基数树进行优化

2.5.1 效率瓶颈

此时，项目当中存在一个非常明显的效率瓶颈，那就是PageCache::FindSpan。

我们在项目当中查找FindSpan的相关引用，有如下结果：

可以看到非常的多，无论是ThreadCache还是CentralCache，都会对该函数进行引用。

尤其是在释放内存时，ThreadCache每次释放内存都需要调用FindSpan，这就导致所有线程在释放内存时都会产生激烈的锁竞争，远不及申请内存的效率。

2.5.2 基数树

为了缓解上面的问题，我们可以使用基数树来代替unordered_map。如此一来，我们就可以无锁地调用FindSpan。

因为相比于红黑树与哈希表，基数树在不断插入数据的过程中虽然也会修改结构，但是其修改结构时不会影响到已存在的数据所在的路径。

所以我们在对基数树进行查询时，无需加锁（修改时仍旧需要加锁）。

并且，基数树可以看作时红黑树和哈希表的结合体，查询效率比红黑树高，比哈希表稳定，这一原因也会提升我们的效率。

关于基数树的具体信息，这里就不过多介绍，读者可以自行查阅相关资料。

我们简单地改造了TCMalloc中的基数树并用到了我们的项目当中（详见TCMalloc_PageMap.h）：

#ifdef _WIN64
	typedef unsigned long long page_id_t;
	static const size_t bit = 64;
#elif _WIN32
	typedef size_t page_id_t;
	static const size_t bit = 32;
#else
	// linux
#endif

// 32位: 32 - page_exponent  64位: 64 - page_exponent
TCMalloc_PageMap2<bit - page_exponent> _page_map;

Span* PageCache::FindSpan(page_id_t page_id)
{
	// 性能瓶颈，每次查找都需要对PageCache加锁
	// std::lock_guard<std::recursive_mutex> guard(_mtx);
	// auto span = _pageid_span.find(page_id);
	// if (span != _pageid_span.end())
	// {
	//		return span->second;
	// }
	return (Span*)_page_map.get(page_id);
}

这里的非类型模板参数代表需要管理的元素个数：页位数 = 地址空间位数 - 页内位数。

上面我们使用的是2层基数树，相比3层基数树来说更加简单，相比1层基数树来说更加节省空间（一层基数树需要一次性为所有元素开辟空间）。

在32位环境下，基数树最多需要开辟 $2^{(32-13)}$ 个单元；而在64位环境下，基数树最多需要开辟 $2^{(64 - 13)}$ 个单元。

单层基数树本质上就是把页号当作数组下标的哈希表，需要一次性开辟出上面的这些空间；而多层基数树本质上就是将这个哈希表进行了分组，可以在使用过程中动态开辟空间。

显然，在64位环境下，至少需要三层基数树才能使程序正常运行；而在32位环境下则可选择使用1 ~ 2层的基数树。

2.6 测试

我们先启动四个线程，让他们使用ConcurrentAlloc和ConcurrentFree进行10轮申请和释放，每轮100000次；
然后再启动四个线程，让他们使用malloc和free进行相同的操作；
最后比较两组操作的用时。

2.6.1 测试代码

#include"ConcurrentAlloc.h"
#include "stdio.h"

// ntimes 一轮申请和释放内存的次数
// rounds 轮次
void BenchmarkMalloc(size_t ntimes, size_t nworks, size_t rounds)
{
	std::vector<std::thread> vthread(nworks);
	std::atomic<size_t> malloc_costtime = 0;
	std::atomic<size_t> free_costtime = 0;

	for (size_t k = 0; k < nworks; ++k)
	{
		vthread[k] = std::thread([&, k]() {
			std::vector<void*> v;
			v.reserve(ntimes);

			for (size_t j = 0; j < rounds; ++j)
			{
				size_t begin1 = clock();
				for (size_t i = 0; i < ntimes; i++)
				{
					v.push_back(malloc(16));
					// v.push_back(malloc((16 + i) % 8192 + 1));
				}
				size_t end1 = clock();

				size_t begin2 = clock();
				for (size_t i = 0; i < ntimes; i++)
				{
					free(v[i]);
				}
				size_t end2 = clock();
				v.clear();

				malloc_costtime += (end1 - begin1);
				free_costtime += (end2 - begin2);
			}
			});
	}

	for (auto& t : vthread)
	{
		t.join();
	}

	printf("%u个线程并发执行%u轮次，每轮次malloc %u次: 花费：%u ms\n",
		nworks, rounds, ntimes, malloc_costtime.load());

	printf("%u个线程并发执行%u轮次，每轮次free %u次: 花费：%u ms\n",
		nworks, rounds, ntimes, free_costtime.load());

	printf("%u个线程并发malloc&free %u次，总计花费：%u ms\n",
		nworks, nworks * rounds * ntimes, malloc_costtime.load() + free_costtime.load());
}


// 单轮次申请释放次数 线程数 轮次
void BenchmarkConcurrentMalloc(size_t ntimes, size_t nworks, size_t rounds)
{
	std::vector<std::thread> vthread(nworks);
	std::atomic<size_t> malloc_costtime = 0;
	std::atomic<size_t> free_costtime = 0;

	for (size_t k = 0; k < nworks; ++k)
	{
		vthread[k] = std::thread([&]() {
			std::vector<void*> v;
			v.reserve(ntimes);

			for (size_t j = 0; j < rounds; ++j)
			{
				size_t begin1 = clock();
				for (size_t i = 0; i < ntimes; i++)
				{
					v.push_back(ConcurrentAlloc(16));
					// v.push_back(ConcurrentAlloc((16 + i) % 8192 + 1));
				}
				size_t end1 = clock();

				size_t begin2 = clock();
				for (size_t i = 0; i < ntimes; i++)
				{
					ConcurrentFree(v[i]);
				}
				size_t end2 = clock();
				v.clear();

				malloc_costtime += (end1 - begin1);
				free_costtime += (end2 - begin2);
			}
			});
	}

	for (auto& t : vthread)
	{
		t.join();
	}

	printf("%u个线程并发执行%u轮次，每轮次concurrent alloc %u次: 花费：%u ms\n",
		nworks, rounds, ntimes, malloc_costtime.load());

	printf("%u个线程并发执行%u轮次，每轮次concurrent dealloc %u次: 花费：%u ms\n",
		nworks, rounds, ntimes, free_costtime.load());

	printf("%u个线程并发concurrent alloc&dealloc %u次，总计花费：%u ms\n",
		nworks, nworks * rounds * ntimes, malloc_costtime.load() + free_costtime.load());
}

int main()
{
	size_t n = 100000;
	cout << "==========================================================" << endl;
	BenchmarkConcurrentMalloc(n, 4, 10);
	cout << endl << endl;

	BenchmarkMalloc(n, 4, 10);
	cout << "==========================================================" << endl;

	return 0;
}

2.6.2 每次申请不同大小的内存块

v.push_back(ConcurrentAlloc((16 + i) % 8192 + 1));
v.push_back(malloc((16 + i) % 8192 + 1));

可以看到使用ConcurrentAlloc/ConcurrentFree的组成功完成了申请释放，但是使用malloc/free的组迟迟给不出结果。

我们将每轮的数据量调整到10000，再次进行测试：

可以看到我们的项目有明显的效率优势。

2.6.3 每次申请大小相同的内存块

v.push_back(ConcurrentAlloc(16));
v.push_back(malloc(16));

此时，我们的项目在效率上依然存在优势，但是已经相差不多了。

这是因为在申请相同大小的内存块时，Central Cache中桶锁的竞争会非常激烈。

2.6.4 总结

我们的项目在多线程环境下，频繁申请释放不同大小的内存块时具有明显的效率优势。

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

程序员学 AI（八）：AI Agent 到底是什么？从回答问题到完成任务

AI Agent技术社区

22道AI Agent 工程师必会的面试题，别啥都不会就出去找工作

AI Agent技术社区

基于AI agent的智能图书馆

AI Agent技术社区

所有评论(0)

查看更多评论

大筒木老辈子

@2302_80372340

已为社区贡献6条内容

C++标准项目---高并发内存池

大筒木老辈子

1. 项目介绍

1.1 什么是内存池

1.1.1 内存池的基本工作原理

1.1.2 内存池的作用

1.1.3 常见应用场景

1.2 项目架构

2 具体实现

2.1 组织小块内存

2.2 Thread Cache

2.2.1 内存块对齐

2.2.2 FreeList

2.2.3 ThreadCache类的定义

2.2.4 TLS无锁访问

2.2.5 慢启动方式获取内存块

2.3 Central Cache

2.3.1 跨度（Span）

2.3.1.1 为什么需要Span

2.3.1.2 页操作宏

2.3.2 SpanList

2.3.3 CentralCache类的定义

2.3.4 什么时候将Span还给Page Cache

2.3.5 通过页号查找Span

2.4 Page Cache

2.4.1 向系统申请/释放内存

2.4.2 Span的申请与释放

2.4.3 Span的切分

2.4.4 Span的合并

2.4.4.1 存在性检查

2.4.4.2 可用性检查

2.4.5 Page Cache类的定义

2.5 大内存申请问题

2.5 使用基数树进行优化

2.5.1 效率瓶颈

2.5.2 基数树

2.6 测试

2.6.1 测试代码

2.6.2 每次申请不同大小的内存块

2.6.3 每次申请大小相同的内存块

2.6.4 总结

所有评论(0)

温馨提示：您尚未绑定手机号

大筒木老辈子