C++项目高并发内存池

紧跟C++项目高并发内存池_三层缓存

文章目录

C++代码Commen.h：添加了向直接向堆中释放空间SystemFree函数ConcurrentAlloc.h：添加了线程申请超过256KB空间情况PageCache：申请Span和释放内存过程添加了大于32页的情况单线程测试代码位置

当申请内存小于256KB时，走前三层缓存获得内存。

当申请内存大小大于256KB时就需要新的方法进行申请。

256KB=32页（一页定义为8KB），直接向PageCache申请，并且记录申请的内存页号与span的对于关系。

如果大于PageCache中哈希桶的大小128页时，直接找系统堆申请内存

ThreadCache中哈希桶的个数为208个，最大可以申请到256KB

static const size_t MAX_BYTE = 256 * 1024;//如果线程申请超过256KB不能直接向ThreadCache申请空间

static const size_t NUMLIST = 208;//ThreadCache中哈希桶的个数

static const size_t NPAGE = 129;//PageCache中哈希桶的个数,0号桶空开从1号桶开始

static const size_t PAGESIZE = 13;//定义一页的大小为2^13（8K）

注意：直接向PageCache或堆申请内存时需要加锁。

对于释放内存，如果大于128页，说明直接向系统堆上开辟的空间，先找到这个地址对应的span，再复用之前的PageCache释放空间的函数释放内存。

释放直接向堆申请的空间为：

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

所以直接涉及到线程调用申请函数方面以及PageCache方面，其余层不需要改变代码。

C++代码 Commen.h：添加了向直接向堆中释放空间SystemFree函数

#pragma once

#include
#include
#include
#include
#include
#include
#include
#include

#include

// 直接去堆上按页申请空间
inline static void* SystemAlloc(size_t kpage)
{
#ifdef _WIN32
	void* ptr = VirtualAlloc(0, kpage << 13, 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
}

using std::cout; using std::endl;

static const size_t MAX_BYTE = 256 * 1024;//如果线程申请超过256KB不能直接向ThreadCache申请空间

static const size_t NUMLIST = 208;//ThreadCache中哈希桶的个数

static const size_t NPAGE = 129;//PageCache中哈希桶的个数,0号桶空开从1号桶开始

static const size_t PAGESIZE = 13;//定义一页的大小为2^13（8K）

#ifdef _WIN64
typedef unsigned long long PAGE_ID;//64位下的页号
#elif _WIN32
typedef size_t PAGE_ID;
#else 
	//Linux
#endif // _WIN32


//获取自由链表下一个节点
static void*& NextObj(void* obj) {
	return *(void**)obj;
}

//管理切分好内存的自由链表
class FreeList {
public:
	FreeList() :_freeList(nullptr), MaxSize(1), size(0) {}
	void Push(void* obj) {
		//头插
		assert(obj);
		NextObj(obj) = _freeList;
		_freeList = obj;
		++size;
	}

	void* Pop() {
		//头删
		assert(_freeList);
		void* obj = _freeList;
		_freeList = NextObj(obj);
		--size;
		return obj;
	}

	void PushList(void* begin, void* end, size_t len) {
		//头插链表
		NextObj(end) = _freeList;
		_freeList = begin;
		size += len;
	}

	void PopList(void*& begin, void*& end, size_t len) {//获得len个内存节点
		assert(len <= size);//要弹出的节点个数小于现在freeList中的节点个数
		begin = _freeList; end = begin;
		for (size_t i = 0; i < len - 1; i++) {
			end = NextObj(end);
		}
		_freeList = NextObj(end);
		NextObj(end) = nullptr;
		size -= len;
	}

	bool Empty() { return _freeList == nullptr; }

	size_t GetSize() { return size; }
	size_t& GetMaxSize() { return MaxSize; }
private:
	void* _freeList;
	size_t MaxSize;//记录freeList链表要向中心缓存层申请多少个内存节点

	size_t size;
};

//计算ThreadCache中哈希桶的桶个数
class SizeClass {
public:
	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{
			return _RoundUp(size, 1 << PAGESIZE);//8KB对其
		}
	}
	//计算在哈希桶的那个位置
	static inline size_t Index(size_t size) {
		assert(size <= MAX_BYTE);
		static int Group[4] = { 16,56,56,56 };
		if (size <= 128) return _Index(size, 8);
		else if (size <= 1024) return _Index(size - 128, 16) + Group[0];
		else if (size <= 8 * 1024) return _Index(size - 1024, 128) + Group[1] + Group[0];
		else if (size < 64 * 1024) return _Index(size - 8 * 1024, 1024) + Group[2] + Group[1] + Group[0];
		else if (size < 256 * 1024) return _Index(size - 64 * 1024, 8 * 1024) + Group[3] + Group[2] + Group[1] + Group[0];
		else {
			assert(false);
			return -1;
		}
	}

	//ThreadCache一个Span中链表挂多少个空间节点
	static size_t ForMemory(size_t size) {
		assert(size > 0);
		size_t num = MAX_BYTE / size;
		if (num < 2) {
			num = 2;//如果对象很大，一次少给ThreadCache一些。
		}
		else if (num > 512) {//如果对象很小，一次多给ThreadCache一些
			num = 512;
		}
		return num;
	}

	//计算PageCache向系统申请页的数目
	static size_t NumForPage(size_t size) {
		size_t NumForMemory = ForMemory(size);//计算中心缓存层一个Span最多要多少节点

		size_t Byte = NumForMemory * size;//总共的字节数

		size_t NumPage = (Byte >> PAGESIZE);//计算申请几页
		if (NumPage == 0) {
			NumPage = 1;//至少给一页空间
		}
		return NumPage;
	}
private:
	static inline size_t _RoundUp(size_t size, size_t AlignNum) {
		size_t AligSize = size;
		if (size % AlignNum != 0) {
			AligSize = (size / AlignNum + 1) * AlignNum;
		}
		return AligSize;
	}
	static inline size_t _Index(size_t size, size_t AlignNum) {
		size_t Pos = 0;
		if (size % AlignNum == 0) {
			Pos = size / AlignNum - 1;
		}
		else {
			Pos = size / AlignNum;
		}
		return Pos;
	}
};

//管理以页为单位的大块空间结构
struct Span {
	PAGE_ID _PageID;//记录是第几页
	size_t _Num;//记录Span里面有多少页
	Span* _next;//双向链表
	Span* _prev;

	size_t use_count;//记录分配了多少个对象给ThreadCahce

	void* FreeList;//切好的小块内存空间

	bool IsUse;//标记这块Span是否被使用

	Span() :_PageID(0), _Num(0), _next(nullptr), _prev(nullptr), use_count(0), FreeList(nullptr), IsUse(false) {}
};

class SpanList {
private:
	Span* _head;
public:
	std::mutex _mtx;//桶锁
	SpanList() {//带头双向循环链表
		_head = new Span;
		_head->_next = _head;
		_head->_prev = _head;
	}

	void Insert(Span* pos, Span* NewSpan) {
		//pos位置前插入NewSpan
		assert(pos != nullptr && NewSpan != nullptr);
		Span* prev = pos->_prev;//前一个
		prev->_next = NewSpan;
		NewSpan->_prev = prev;
		NewSpan->_next = pos;
		pos->_prev = NewSpan;
	}

	Span* Pop() {
		Span* front = _head->_next;//带头双向循环链表
		Erase(_head->_next);
		return front;
	}

	void Erase(Span* pos) {//删除SpanList上的节点
		assert(pos != _head);
		Span* prev = pos->_prev;
		Span* next = pos->_next;
		prev->_next = next;
		next->_prev = prev;
	}

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

	bool Empty() { return _head->_next == _head; }
};

ConcurrentAlloc.h：添加了线程申请超过256KB空间情况

#pragma once

#include"Common.h"
#include"ThreadCache.h"
#include"PageCache.h"

//线程调用申请ThreadCache空间
static void* ConcurrentAlloc(size_t size) {
	if (size > MAX_BYTE) {//大于256KB内存
		size_t AlignSize = SizeClass::RoundUp(size);//计算对其大小
		//直接向PageCache索要K页的内存
		size_t K = AlignSize >> PAGESIZE;
		PageCache::GetInst()->_PageMtx.lock();
		Span*span=PageCache::GetInst()->NewSpan(K);
		PageCache::GetInst()->_PageMtx.unlock();
		void* ptr = (void*)(span->_PageID << PAGESIZE);//获取这块内存的地址
		return ptr;
	}
	else {
		//获取线程自己的ThreadCache
		if (tls_threadcache == nullptr) {
			tls_threadcache = new ThreadCache;
		}
		cout << std::this_thread::get_id() << " " << tls_threadcache << endl;
		return tls_threadcache->ApplySpace(size);
	}
}

static void ConcurrentFree(void* ptr, size_t size) {
	if(size>MAX_BYTE){
		Span* span = PageCache::GetInst()->MapObjToSpan(ptr);//计算要释放的大空间属于那个Span

		PageCache::GetInst()->_PageMtx.lock();
		PageCache::GetInst()->RetrunPageCache(span);//将内存挂到PageCache桶上，需要修改桶，所以要加锁
		PageCache::GetInst()->_PageMtx.unlock();
	}
	else {
		//释放时每个线程一定有tls_threadcache
		assert(tls_threadcache != nullptr);
		tls_threadcache->ReleaseSpace(ptr, size);
	}
}

PageCache：申请Span和释放内存过程添加了大于32页的情况

#pragma once

#include"Common.h"

class PageCache {
private:
	SpanList _SpanList[NPAGE];
	static PageCache _sInst;

	//页号与Span链表的映射,方便归还内存时直接通过内存找页号找到这块内存是那个Span
	std::unordered_mapIdSpanMap;

	PageCache() {}
public:
	std::mutex _PageMtx;

	PageCache(const PageCache&) = delete;

	//获取这个内存是那个Span
	Span* MapObjToSpan(void* obj);

	static PageCache* GetInst() {
		return &_sInst;
	}

	//将CentralCache的Span回收,合并相邻页
	void RetrunPageCache(Span* span);

	//获取NumPage页的Span
	Span* NewSpan(size_t NumPage);
};

#include"PageCache.h"

PageCache PageCache::_sInst;

Span* PageCache::NewSpan(size_t NumPage) {//NumPage是页数
	assert(NumPage > 0);

	if (NumPage >= NPAGE) {//如果要申请的内存大于桶的个数，直接向堆申请空间
		void* ptr = SystemAlloc(NumPage);
		Span* span = new Span;
		span->_PageID = (PAGE_ID)ptr >> PAGESIZE;
		span->_Num = NumPage;
		//保存起始页号，方便释放内存
		IdSpanMap[span->_PageID] = span;
		return span;
	}
	else {
		//看当前位置桶中是否有Span
		if (!_SpanList[NumPage].Empty()) {
			return _SpanList[NumPage].Pop();
		}
		//对应位置的桶是空，检查后面桶里有没有Span,将大Span切分成小Span
		for (size_t i = NumPage + 1; i < NPAGE; i++) {
			if (!_SpanList[i].Empty()) {//有一个桶存在，切分大Span成NumPage页Span和N-NumPage页的Span
				Span* NumPageSpan = new Span;
				Span* NSpan = _SpanList[i].Pop();
				//头切
				NumPageSpan->_PageID = NSpan->_PageID;
				NumPageSpan->_Num = NumPage;
				NSpan->_PageID += NumPage;
				NSpan->_Num -= NumPage;

				//将切下的Span挂到其他桶上
				_SpanList[NSpan->_Num].Insert(_SpanList[NSpan->_Num].begin(), NSpan);
				//保存NSpan前后页的映射关系，方便回收
				IdSpanMap[NSpan->_PageID] = NSpan;
				IdSpanMap[NSpan->_PageID + NSpan->_Num - 1] = NSpan;//中间页没必要加入映射中，前后两页为了方便回收

				//填入PAGE_ID与Span*映射中
				for (PAGE_ID i = 0; i < NumPageSpan->_Num; i++) {//切了Num页
					IdSpanMap[NumPageSpan->_PageID + i] = NumPageSpan;
				}

				return NumPageSpan;
			}
		}
		//所有桶都没有Span,直接向堆中申请一大块空间，将这块空间挂到最后一个桶上
		Span* BigSpan = new Span;
		void* ptr = SystemAlloc(NPAGE - 1);
		BigSpan->_PageID = (PAGE_ID)ptr >> PAGESIZE;
		BigSpan->_Num = NPAGE - 1;
		_SpanList[BigSpan->_Num].Insert(_SpanList[BigSpan->_Num].begin(), BigSpan);
		//在调用自己向Central Cache发送空间
		return NewSpan(NumPage);
	}
}

Span* PageCache::MapObjToSpan(void* obj) {
	//计算obj的页号
	PAGE_ID pageId = (PAGE_ID)obj >> PAGESIZE;
	//获取这个内存是那个Span
	std::unordered_map::iterator ret = IdSpanMap.find(pageId);
	if (ret != IdSpanMap.end()) {
		return ret->second;
	}
	else {
		assert(false);//不可能找不到
		return nullptr;
	}
}

//将CentralCache的Span回收,合并相邻页
void PageCache::RetrunPageCache(Span* span) {
	if (span->_Num >= NPAGE) {
		//直接向堆申请的空间大于128页的还给系统，其余的挂在桶上
		void* ptr = (void*)(span->_PageID << PAGESIZE);
		SystemFree(ptr);
		delete span;
	}
	else {
		//对Span前后页进行合并，缓解内存碎片外碎片问题
		while (true) { //向前合并
			PAGE_ID prevId = span->_PageID - 1;
			std::unordered_map::iterator ret = IdSpanMap.find(prevId);
			if (ret == IdSpanMap.end()) {
				break;
			}
			else {
				if (ret->second->IsUse == true) {
					//前面相邻页的Span正在使用的内存不合并
					break;
				}
				//和前Span合并大小超过128KB则停止合并,无法管理
				else if (ret->second->_Num + span->_Num >= NPAGE) {
					break;
				}
				else {
					//合并
					span->_PageID = ret->second->_PageID;
					span->_Num += ret->second->_Num;
					_SpanList[ret->second->_Num].Erase(ret->second);
					delete ret->second;
				}
			}
		}
		//向后合并
		while (true) {
			PAGE_ID nextId = span->_PageID + span->_Num;
			std::unordered_map::iterator ret = IdSpanMap.find(nextId);
			if (ret == IdSpanMap.end()) {
				break;
			}
			else {
				Span* nextSpan = ret->second;
				if (nextSpan->IsUse == true) {
					break;
				}
				else if (nextSpan->_Num + span->_Num >= NPAGE) {
					break;
				}
				else {
					span->_Num += nextSpan->_Num;
					_SpanList[nextSpan->_Num].Erase(nextSpan);
					IdSpanMap.erase(nextSpan->_Num);
					delete nextSpan;
				}
			}
		}
		//将合并的Span挂起并添加映射
		_SpanList[span->_Num].Insert(_SpanList[span->_Num].begin(), span);
		span->IsUse = false;
		IdSpanMap[span->_PageID] = span;
		IdSpanMap[span->_PageID + span->_Num - 1] = span;
	}
}

单线程测试

#include"ConcurrentAlloc.h"
void TestBigAlloc() {
	void* ptr = ConcurrentAlloc(32 * 1024 * 8 + 1);//大于32页小于128页
	void* ptr1 = ConcurrentAlloc(128 * 1024 * 8 + 1);//大于128页
	ConcurrentFree(ptr, 32 * 1024 * 8 + 1);
	ConcurrentFree(ptr1, 128 * 1024 * 8 + 1);
}

int main()
{
	TestBigAlloc();
	return 0;
}

至此，项目全部写完，后序过程就是对项目进行性能测试及其优化。

代码位置

Github
Gitee

C++项目高并发内存池

C/C++/C#相关栏目本月热门文章