[数据结构与算法] C++项目高并发内存池

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 数据结构与算法 -> C++项目高并发内存池_大于256KB的内存申请 -> 正文阅读

[数据结构与算法]C++项目高并发内存池_大于256KB的内存申请

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

文章目录

当申请内存小于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<iostream>
#include<vector>
#include<time.h>
#include<assert.h>
#include<thread>
#include<mutex>
#include<algorithm>
#include<unordered_map>

#include<windows.h>

// 直接去堆上按页申请空间
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_map<PAGE_ID, Span*>IdSpanMap;

	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<PAGE_ID,Span*>::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<PAGE_ID, Span*>::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<PAGE_ID, Span*>::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;
}