C++从零实现一个高并发内存池

Posted 蓝乐

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了C++从零实现一个高并发内存池相关的知识，希望对你有一定的参考价值。

从零实现一个高并发内存池

🎄项目介绍

◎项目的内容介绍

这个项目是实现一个高并发内存池，参考了google的开源项目tcmalloc实现的简易版；其功能就是实现高效的多线程内存管理。由功能可知，高并发指的是高效的多线程，而内存池则是实现内存管理的，关于内存池接下来会进行详细介绍
tcmalloc源码

◎要求的知识储备

实现这个项目要求掌握C/C++、数据结构（链表与哈希桶）、操作系统的内存管理、单例模式、多线程以及互斥锁等方面的知识。

◎内存池的介绍

1、池化技术

我们知道，向系统申请和释放资源都有较大的开销，而池化技术就是程序先向系统申请过量的资源，而这些资源由我们自己管理，这样就能避免频繁的申请和释放资源导致的开销。
其实，在计算机中，除了我们上面所说的内存池，还有我们之前提到过的数据结构池，以及线程池、对象池、连接池等等，都利用了池化技术。

2、内存池

内存池指的是程序预先向操作系统申请足够大的一块内存空间；此后，程序中需要申请内存时，不需要直接向操作系统申请，而是直接从内存池中获取；同理，程序释放内存时，也不是将内存直接还给操作系统，而是将内存归还给内存池。当程序退出（或者特定时间）时，内存池才将之前申请的内存真正释放。

3、内存池主要解决的问题

由上可知，内存池首要解决的是效率问题，其次从系统的内存分配器角度出发，还需要解决内存碎片的问题。那么什么是内存碎片问题呢？
内存碎片分为外碎片和内碎片。

  • 外碎片由下图所示：对于我们申请的内存，可能因为频繁的申请和释放内存导致内存空间不连续，那么就会出现明明由足够大的内存空间，但是我们却申请不出连续的空间出来，这便是外碎片问题了。

  • 内碎片则是由于一些对齐的需求，导致分配出去的内存空间无法被利用，关于内碎片我们项目后面中会涉及，到时候会有更准确的理解。

4、malloc解析

我们在C语言中动态申请内存时，是通过malloc函数去申请内存的，但是实际上malloc并不是直接向堆申请内存的，而malloc本质上就是一个内存池，即向操作系统申请一大块内存，当程序将malloc管理的内存池中内存全部申请完后或者有大量内存需求时，malloc函数就会继续向操作系统申请空间。

🎄设计思路

◎第一阶段–设计一个定长的内存池

我们直到malloc函数申请内存空间是通用的，即任何场景下都可以使用，但是各方面都会就表示各方面都不顶尖，那么我们可以设计一个定长内存池来保证特定场景下的内存申请效率要高于malloc函数。

适应平台的指针方案

在这里，我们想取出一块对象内存中的前4个字节（32位系统）或者8个字节（64位系统）的内存来存储一个指针指向下一块释放回来的自由对象内存，那么在这里为了不作平台系统的判断，可以使用一个小技巧，即将对象内存强转成void**的类型，那么再对这个二级指针类型解引用就可以取出当前对象的前4个字节（32位系统）或8个字节（64位系统）。
由于这个操作之后会频繁使用，因此定义为内敛函数放在common.h头文件中方便调用：

static inline void*& NextObj(void* obj)

	return *(void**)obj;

由此，我们就可以设计出定长内存池的对象：

template <class T>//一次申请T大小的内存空间
class ObjectPool

public:
	T* New()
	
		T* obj;
		//若自由链表有对象，则直接从自由链表中取
		if (_freeList != nullptr)
		
			//头删
			obj = (T*)_freeList;
			_freeList = *(void**)_freeList;
		
		else
		
			if (_remainedBytes < sizeof(T))//当前内存池中没有足以分配的内存，需要申请
			
				_remainedBytes = 8 * 1024;	
				//_memory = (char*)malloc(_remainedBytes);//申请定长（8Kb）的内存空间
				_memory = (char*)SystemAlloc(_remainedBytes >> PAGE_SHIFT);//申请定长（8Kb）的内存空间

			
			//保证一次分配的空间够存放下当前平台的指针
			size_t unity = sizeof(T) < sizeof(void*) ? sizeof(void*) : sizeof(T);
			obj = (T*)_memory;
			_memory += unity;
			_remainedBytes -= unity;
		
		//定位new显式调用T类型构造函数
		new(obj)T;
		return obj;
	
	void Delete(T* obj)//将obj这块内存链接到_freeList中
	
		//显式调用obj对象的析构函数,清理空间
		obj->~T();
		//将obj内存块头插
		*(void**)obj = _freeList;
		_freeList = obj;
	
private:
	char* _memory = nullptr;//存储一次申请一大块的内存,char*类型便于分配内存
	void* _freeList = nullptr;//将释放的内存回收链接
	size_t _remainedBytes;//_memory中剩余的内存空间
;

对于我们设计的定长内存池，我们可以通过下面的测试代码来比较一下malloc与定长内存池的效率：

void TestObjectPool()

	// 申请释放的轮次
	const size_t Rounds = 5;
	// 每轮申请释放多少次
	const size_t N = 1000000;
	size_t begin1 = clock();
	std::vector<TreeNode*> v1;
	v1.reserve(N);
	for (size_t j = 0; j < Rounds; ++j)
	
		for (int i = 0; i < N; ++i)
		
			v1.push_back(new TreeNode);
		
		for (int i = 0; i < N; ++i)
		
			delete v1[i];
		
		v1.clear();
	
	size_t end1 = clock();
	ObjectPool<TreeNode> TNPool;
	size_t begin2 = clock();
	std::vector<TreeNode*> v2;
	v2.reserve(N);
	for (size_t j = 0; j < Rounds; ++j)
	
		for (int i = 0; i < N; ++i)
		
			v2.push_back(TNPool.New());
		
		for (int i = 0; i < 100000; ++i)
		
			TNPool.Delete(v2[i]);
		
		v2.clear();
	
	size_t end2 = clock();
	cout << "new cost time:" << end1 - begin1 << endl;
	cout << "object pool cost time:" << end2 - begin2 << endl;


可以明显的看出，定长内存池的开销是要低于malloc的，由此可见，在特定场景下，定长内存池的效率高于malloc函数。

◎第二阶段–高并发内存池整体框架设计

随着开发环境逐渐多核多线程，那么在申请内存的场景下，必然存在激烈的锁竞争问题。其实，malloc本身就已经足够优秀了，但我们项目的原型tcmalloc将在多线程高并发的场景下更胜一筹。而我们这次实现的内存池将考虑以下几方面的问题：

  • 1.性能问题
  • 2.多线程场景下的锁竞争问题
  • 3.内存碎片问题
    concurrent memory pool（并发内存池），主要有以下3个部分组成：

1.线程缓存（thread cache）

线程缓存是每个线程独有的，用于小于256kb内存的分配。那么对于每一个线程从thread cache申请资源，就无需考虑加锁问题，每个线程独享一个缓存（cache），这也是并发线程池高效的地方。

2.中心缓存（central cache）

中心缓存有所有线程所共享，thread cache 按需从central cache处获取对象，而central cache在合适的时机从thread cache处回收对象从而避免一个线程占用太多资源，导致其他线程资源吃紧，进而实现内存分配在多个线程更加均衡的按需调度。由于所有thread cache都从一个central cache中取内存对象，故central cache是存在竞争的，也就是说从central cache中取内存对象需要加锁，但我们在central cache这里用的是桶锁，且只有thread cache中没有对象后才会来central cache处取对象，因此锁的竞争不会很激烈。

3.页缓存（page cache）

页缓存是中心缓存上一级的缓存，存储并分配以页为单位的内存，central cache中没有内存对象时，会从page cache中分配出一定数量的page，并切割成定长大小的小块内存，给central cache。当page cache中一个span的几个跨度页都回收以后，page cache会回收central cache中满足条件的span对象，并且合并相邻的页，组成更大的页，从而缓解内存碎片（外碎片）的问题。

◎第三阶段–三级缓存的具体实现

1.Thread Cache框架构建及核心实现

thread cache是哈希桶结构，每个桶是一个根据桶位置映射的挂接内存块的自由链表，每个线程都会有一个thread cache对象，这样就可以保证线程在申请和释放对象时是无锁访问的。

🌕申请与释放内存的规则及无锁访问

  • 申请内存
  1. 当内存申请大小size不超过256KB，则先获取到线程本地存储的thread cache对象，计算size映射的哈希桶自由链表下标i。
  2. 如果自由链表_freeLists[i]中有对象，则直接Pop一个内存对象返回。
  3. 如果_freeLists[i]中没有对象时，则批量从central cache中获取一定数量的对象，插入到自由链表并返回一个对象。
  • 释放内存
    1.当释放内存小于256kb时将内存释放回thread cache，计算size映射自由链表桶位置i，将对象Push到_freeLists[i]。
    2.当链表的长度过长，则回收一部分内存对象到central cache。
  • tls - thread local storage
    线程局部存储（tls），是一种变量的存储方法，这个变量在它所在的线程内是全局可访问的，但是不能被其他线程访问到，这样就保持了数据的线程独立性。而熟知的全局变量，是所有线程都可以访问的，这样就不可避免需要锁来控制，增加了控制成本和代码复杂度。
//TLS: thread local storage，实现线程的无锁访问
static _declspec(thread) ThreadCache* pTLSThreadCache = nullptr;

🌕管理内存对齐和映射等关系

▶计算对齐大小映射的规则

thread cache中的哈希桶映射比例比非均匀的，如果将内存大小均匀划分的话，则会划分出大量的哈希桶，比如256kb如果按照8byte划分，则会创建32768个哈希桶，这就有较大的内存开销；而如果按照更大的字节划分，那么内存开销虽然减少了，但照顾到的场景也少了，且会产生内碎片问题。
那么参考tcmalloc项目，为了保证内碎片的浪费整体控制在10%左右进行的区间映射，同时没有那么大的开销，对于内存的哈希映射关系如下：

	// 整体控制在最多10%左右的内碎片浪费
	// [1,128] 8byte对齐       freelist[0,16)
	// [128+1,1024] 16byte对齐   freelist[16,72)
	// [1024+1,8*1024] 128byte对齐   freelist[72,128)
	// [8*1024+1,64*1024] 1024byte对齐     freelist[128,184)
	// [64*1024+1,256*1024] 8*1024byte对齐   freelist[184,208)

也就是说，对于1byte到128byte的内存对象，按照8byte对齐，划分为下标0-15号的哈希桶，而129byte到1kb的内存对象，按照16byte对齐，划分下标16-71号的哈希桶，以此类推，最终划分为0-207号总共208个哈希桶，这样就保证了内存较小的开销，同时各个对齐关系中内碎片浪费控制在10%左右，比如129byte到144byte区间，取144byte的内存对象，浪费率为(144 - 129) / 144 = 10.42%,当然对于最开始的1byte申请8byte内存对象，虽然浪费高达87.5%，但考虑到最终内碎片浪费了7byte，对比后续内碎片一次浪费7kb来说可以忽略不计了。
这便是申请的内存对象大小对齐的映射关系，这个关系在后续central cache及page cache中仍有应用，因此可以将其定义在头文件common.h中，以后内存大小对齐的管理。

▶计算相应内存映射在哪一个哈希桶中

既然我们已经有了对齐内存的映射关系，那么我们需要通过这个映射关系去计算相应内存对象挂接在哪一个哈希桶中，在这里可以直接使用相关映射关系去进行加减乘除运算：

	//计算对应链桶的下标
	//static inline size_t _Index(size_t bytes, size_t alignNum)
	//
	//	if (bytes % alignNum == 0)
	//	
	//		return bytes / alignNum - 1;
	//	
	//	else
	//	
	//		return bytes / alignNum;
	//	
	//

但是参考tcmalloc源码，考虑到位移运算更加接近底层，效率更高，而实际应用中映射对应关系的计算是相当频繁的，因此使用位运算来改进算法。

▶代码实现
//管理对齐和映射关系
class SizeClass

public:
	// 整体控制在最多10%左右的内碎片浪费
	// [1,128] 8byte对齐       freelist[0,16)
	// [128+1,1024] 16byte对齐   freelist[16,72)
	// [1024+1,8*1024] 128byte对齐   freelist[72,128)
	// [8*1024+1,64*1024] 1024byte对齐     freelist[128,184)
	// [64*1024+1,256*1024] 8*1024byte对齐   freelist[184,208)

	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);//按8byte对齐
		
		else if(size <= 1024)
		 
			return _RoundUp(size, 16);//按16byte对齐
		
		else if (size <= 8 * 1024)
		
			return _RoundUp(size, 128);//按128byte对齐
		
		else if(size <= 64 * 1024)
		
			return _RoundUp(size, 1024);//按1024byte对齐
		
		else if (size <= 256 * 1024)
		
			return _RoundUp(size, 8 * 1024);//按8*1024byte对齐
		
		else//大于256KB
		
			//assert(false);
			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 size)
	
		assert(size <= MAX_BYTE);
		static int group_array[4] =  16,56,56,56 ;//打表，每个区间有多少个链
		if (size <= 128)
		
			return _Index(size, 3);
		
		else if (size <= 1024)
		
			return _Index(size - 128, 4) + group_array[0];
		
		else if(size <= 8 * 1024)
		
			return _Index(size - 1024, 7) + group_array[1] + group_array[0];
		
		else if (size <= 64 * 1024)
		
			return _Index(size - 8 * 1024, 10) + group_array[2] + group_array[1] + group_array[0];
		
		else if (size <= 256 * 1024)
		
			return _Index(size - 64 * 1024, 13) + group_array[3] + group_array[2] + group_array[1] + group_array[0];
		
		else
		
			assert(false);
		
	

	//计算向CentralCache一次获取批次的数量
	static size_t NumMoveSize(size_t size)
	
		assert(size > 0);//获取的内存对象总大小必须大于0
		size_t num = MAX_BYTE / size;//计算合理的内存对象数量
		//根据批次计算获取的数量会在[1,32768]，范围过大，
		//因此控制获取的对象数量范围在区间[2, 512],较为合理
		//小对象获取的批次多，大对象获取的批次少
		if (num < 2)
			num = 2;
		if (num > 512)
			num = 512;
		return num;
	

	//计算一次向PageCache申请多少页
	//size: 单个对象从8byte - 256kb
	static size_t NumMovePage(size_t size)
	
		size_t num = NumMoveSize(size);//当前size所需批次数量
		size_t nPage = num * size;
		nPage >>= PAGE_SHIFT;
		if (nPage == 0)
			nPage = 1;
		return nPage;
	
;

🌕自由链表的设计

在有了上面的基础之后，我们来设计自由链表，其实就是实现一个单链表，方便插入删除，同时标识链表长度_size以方便后续释放流程，以及定义_maxSize来保住thread cache一次申请对象批次的下限。

class FreeList

public:
	void Push(void* obj)
	
		//将归还的内存块对象头插进自由链表
		NextObj(obj) = _freeList;
		_freeList = obj;
		++_size;
		
	void PushRange(void* start, void* end, size_t size)
	
		NextObj(end) = _freeList;
		_freeList = start;
		_size += size;
	
	void* Pop()
	
		assert(_freeList);
		//将自由链表中的内存块头删出去
		void* obj = _freeList;
		_freeList = NextObj(obj);
		--_size;
		return obj;
	

	void PopRange(void*& start, void*& end, size_t n)
	
		assert(n >= _size);
		start = _freeList;
		end = start;
		for (size_t i = 0; i < n - 1; i++)
		
			end = NextObj(end);
		
		_freeList = NextObj(end);
		_size -= n;
		NextObj(end) = nullptr;
	

	bool Empty()
	
		return _freeList == nullptr;
	
	size_t& MaxSize()
	
		return _maxSize;
	
	size_t& Size()
	
		return _size;
	
private:
	void* _freeList = nullptr;
	size_t _maxSize = 1;//慢增长用于保住申请批次下限
	size_t _size = 0;//计算链表长度
;

🌕thread cache框架构建

在有了上述基础后，我们来搭建thread cache的框架，其实就是一个哈希桶，每个桶中挂接着自由链表对象。
不过在thread cache中采用tls（thread local storage）即线程局部存储技术，来保证一个线程仅拥有一个全局的thread cache对象，从而可以使线程在向thread cache申请内存对象的时候实现无锁访问。

class ThreadCache

public:
	//申请和释放内存对象
	void* Allocate(size_t size);//从Thread Cache中申请size个Bytes的空间
	void DeAllocate(void* ptr, size_t size);

	void* FetchFromCentralCache(size_t index, size_t size);//向CentralCache中申请空间
	void ListTooLong(FreeList& list, size_t size);//释放对象导致链表过长时，归还内存到CentralCache
private:
	FreeList _freeLists[NFREELIST];//ThreadCache的自由链表
;
//TLS: thread local storage，实现线程的无锁访问
static _declspec(thread) ThreadCache* pTLSThreadCache = nullptr;

🌕thread cache核心实现

代码部分：thread cache核心方法实现

2.central cache框架构建及核心实现

central cache也是一个哈希表结构，其映射关系与thread cache是一样的，不同的是central cache的哈希桶位置所挂接的是SpanList链表结构，不过每个桶下的span对象被切分成了一块块小内存挂接在span对象的自由链表freeList中。

🌕申请与释放内存规则

  • 申请内存
    1.当thread cache中没有内存后，就会向central cache中申请大块内存。这里的申请过程采用的是类似网络tcp协议拥塞控制的慢开始算法，而central cache中哈希映射的spanlist下挂着的span则向thread cache提供大块内存，而从span中取出对象给thread cache是需要加锁的，这里为了保证效率，提供的是桶锁。
▶慢开始算法
	//预计获取的批次数量
	//慢开始反馈调节算法
	//1、刚开始向CentralCache获取的批次数量较小，申请太多可能用不完
	//2、当size需求逐渐增大时，batchNum就不断增大，直到上限
	//3、size越小，一次向CentralCache申请的批次数量就较大
	//4、siz
以上是关于C++从零实现一个高并发内存池的主要内容，如果未能解决你的问题，请参考以下文章
 

 C++从零实现一个高并发内存池


 C++实现的高并发内存池


 C++实现的高并发内存池


 实现高并发内存池


 实现高并发内存池


 高并发内存池