[数据结构与算法] 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++项目高并发内存池_三层缓存释放内存过程设计与联调 -> 正文阅读

[数据结构与算法]C++项目高并发内存池_三层缓存释放内存过程设计与联调

文章目录

紧跟高并发内存池申请内存联调接下来就是高并发内存池的内存释放过程

ThreadCache释放内存，将一部分多余的空间节点链表给CentralCache

如果ThreadCache中的FreeList中空间节点超过要申请的长度，回收内存。
这些回收的内存节点给到Central Cache，Central Cache将这些空间节点合并成Span挂到Span哈希桶的对应位置。

#pragma once

#include"Common.h"
#include"CentralCache.h"

class ThreadCache {
private:
	FreeList _FreeList[NUMLIST];
public:
	//申请释放空间
	void* ApplySpace(size_t size) {
		assert(size <= MAX_BYTE);
		size_t AligSize = SizeClass::RoundUp(size);
		//计算桶位置
		size_t index = SizeClass::Index(AligSize);
		if (!_FreeList[index].Empty()) {
			return _FreeList[index].Pop();
		}
		else {
			//thread cache没有内存向central cache要空间
			return RequestFromCentralCache(index, AligSize);
		}
	}

	//释放的空间
	void ReleaseSpace(void* ptr, size_t size) {
		assert(size <= MAX_BYTE && ptr != nullptr);
		size_t index = SizeClass::Index(size);//找到第几个桶
		//将这个空间头插到自由链表上
		_FreeList[index].Push(ptr);

		//如果FreeList中空间节点超过要申请的长度，回收内存
		if (_FreeList[index].GetMaxSize() < _FreeList[index].GetSize()) {
			RetrunCentralCache(_FreeList[index], size);//size是每个节点的大小
		}
	}

	//释放空间到CentralCache
	void RetrunCentralCache(FreeList& List, size_t size) {
		void* start = nullptr; void* end = nullptr;
		List.PopList(start, end, List.GetMaxSize());//一次获取List.GetMaxSize()个节点的链表

		CentralCache::GetCentralCache()->MergeSpan(start, size);
	}


	//向中心缓存申请空间
	void* RequestFromCentralCache(size_t index, size_t size) { 
		//慢开始调节算法
		size_t Num = min(SizeClass::ForMemory(size),_FreeList[index].GetMaxSize());//计算中心层给线程缓存多少个空间节点;
		if (Num == _FreeList[index].GetMaxSize()) {
			_FreeList[index].GetMaxSize() += 2;//慢增长,每次申请下次会多给ThreadCache空间
		}
		//依次递增SizeClass::ForMemory(size)是申请上限，一次申请数量不会比它还大
		void* begin = nullptr; void* end = nullptr;
		size_t actualNum = CentralCache::GetCentralCache()->GiveThreadCache(begin, end, Num, size);
		assert(actualNum >= 1);//至少会申请到一个空间节点。
		if (actualNum == 1) {//实际就获得了一个节点,直接将这个节点返回
			return begin;
		}
		else {
			//获得了多个节点，要把这些节点都插入到ThreadCache的哈希桶上
			_FreeList[index].PushList(NextObj(begin), end, actualNum - 1);//将链表的下一个节点插入桶中，头节点返回给ThreadCache
			return begin;
		}
	}
};

//每个线程都会拥有自己的ThreadCache
static __declspec(thread) ThreadCache* tls_threadcache = nullptr;//用static修饰只在当前文件可见

自由链表类需要添加的成员函数

获取自由链表的节点数
自由链表弹出len个节点给CenralCache

//管理切分好内存的自由链表
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;//记录FreeList有几个节点
};

CentralCache将从ThreadCache获取的空间节点合并为Span（MergeSpan）

需要注意的是，归还内存的时候需要修改CentralCache的Span链表，需要加桶锁。

此外，线程归还内存时这块内存属于Span链表中的那个Span不确定。

如果直到一块空间的页号，这块空间的地址为页号×页大小（这块内存任意位置÷页大小都是页号）用来判断这块内存属于那一页
eg:

void TestPageId() {
	PAGE_ID id = 2020;
	PAGE_ID id2 = 2021;//1页大小
	char* start = (char*)(id << PAGESIZE);
	char* end = (char*)(id2 << PAGESIZE);
	cout << (void*)start << " " << (void*)end << endl;
	while (start <= end) {
		cout << (void*)start << " PAGEID:" << ((PAGE_ID)start >> PAGESIZE) << endl;
		start += 8;//一块空间8字节
	}
}

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

在这里插入图片描述
综上，知道一块内存的地址，可以确定这块内存的页号。为了方便找到那个Span，这里选择添加页号与Span链表的映射std::unordered_map<PAGE_ID, Span*>IdSpanMap;这个映射放到PageCache上

CentralCache

#pragma once

#include"Common.h"

//中心缓存只有一个，每个线程都可以看见，所以中心缓存类采用单例模式(饿汉模式)
class CentralCache {
private:
	SpanList _SpanList[NUMLIST];//与ThreadCache桶的大小相同
	CentralCache(const CentralCache&) = delete;
	static CentralCache _sInst;
public:
	CentralCache() {}
	static CentralCache* GetCentralCache() {
		return &_sInst;
	}

	//获取一个非空Span,需要向PageCache申请空间
	Span* GetSpan(SpanList& List, size_t size);

	size_t GiveThreadCache(void*& start, void*& end, size_t Num, size_t Size);

	//将一条空间链表合并成Span挂到桶上，实现释放内存的过程
	void MergeSpan(void* start, size_t size);
};

#include"CentralCache.h"
#include"PageCache.h"

CentralCache CentralCache::_sInst;//定义了一个对象

size_t CentralCache::GiveThreadCache(void*& start, void*& end, size_t Num, size_t size) {
	size_t index = SizeClass::Index(size);
	//要求num个内存节点，需要计Span中FreeList空节点个数
	_SpanList[index]._mtx.lock();//加锁

	Span* span = GetSpan(_SpanList[index], size);
	assert(span != nullptr && span->FreeList != nullptr);

	start = span->FreeList; end = start;
	//变量Num次，找Num个节点
	size_t actualNum = 1;//实际获得几个内存节点
	for (int i = 0; i < Num - 1; i++) {
		if (NextObj(end) == nullptr) break;//如果走到span中FreeList的空，说明span中内存不够Num个，这个时候有多少返回多少
		actualNum += 1;
		end = NextObj(end);
	}
	span->FreeList = NextObj(end);
	//为分出去的内存节点链表添加nullptr
	NextObj(end) = nullptr;

	span->use_count += actualNum;

	_SpanList[index]._mtx.unlock();
	return actualNum;
}

Span* CentralCache::GetSpan(SpanList& List, size_t size) {
	//在哈希桶对应位置Span链表中找是否有Span，没有就向PageCache申请空间

	//遍历桶的Span链表
	Span* it = List.begin();
	while (it != List.end()) {
		if (it->FreeList != nullptr) {
			return it;//这个Span有空间
		}
		else {
			//Span没有空间，继续找下一个链表Span
			it = it->_next;
		}
	}
	//先把CentralCache的桶锁解开，如果其他线程释放内存不会阻塞
	List._mtx.unlock();
	//没有空闲的Span只能找PageCache,需要加锁，PageCache只能由一个线程访问
	//size是单个对象的大小
	PageCache::GetInst()->_PageMtx.lock();
	Span* span=PageCache::GetInst()->NewSpan(SizeClass::NumForPage(size));
	span->IsUse = true;
	PageCache::GetInst()->_PageMtx.unlock();

	//获得了一块大Span，这块Span这时被线程单独看到，不需要加锁（没有挂到桶上）
	//Span起始地址
	char* start = (char*)((span->_PageID) << PAGESIZE);
	size_t ByteSize = (span->_Num) << PAGESIZE;
	char* end = start + ByteSize;
	//把Span内部大块内存切成自由链表链接起来
	span->FreeList = start;
	start += size;//自由链表的头节点
	void* tail = span->FreeList;
	while (start < end) {
		NextObj(tail) = start;
		tail = NextObj(tail);
		start += size;
	}
	List._mtx.lock();
	List.Insert(List.begin(), span);//将Span挂到桶上，此时需要加桶锁
	return span;
}

void CentralCache::MergeSpan(void* start, size_t size) {
	size_t index = SizeClass::Index(size);
	_SpanList[index]._mtx.lock();

	while (start != nullptr) {
		void* next = NextObj(start);
		Span* span = PageCache::GetInst()->MapObjToSpan(start);
		//将这个节点头插到这个Span上
		NextObj(start) = span->FreeList;
		span->FreeList = start;

		span->use_count--;//每回收一块内存，记录分出去的内存use_count就-1

		if (span->use_count == 0) {
			//Span所有小块内存都回来了，这个Span就可以给Page Cache回收。PageCache做前后页的合并
			_SpanList[index].Erase(span);

			//通过页号就可以再次找到这块内存
			span->FreeList = nullptr;
			span->_next = nullptr;
			span->_prev = nullptr;
			//页号与页数不能动，PageCache通过这两个数据找到这大块内存,此时可以解开桶锁，让其他线程访问这个桶
			_SpanList[index]._mtx.unlock();
			//此时要访问PageCache需要加PageCache的锁
			PageCache::GetInst()->_PageMtx.lock();
			PageCache::GetInst()->RetrunPageCache(span);
			PageCache::GetInst()->_PageMtx.unlock();
			_SpanList[index]._mtx.lock();
		}
		start = next;
	}

	_SpanList[index]._mtx.unlock();
}

PageCache添加映射

#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 && NumPage < NPAGE);

	//看当前位置桶中是否有Span
	if (!_SpanList[NumPage].Empty()) {
		return _SpanList->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);

			//填入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) {

}

PageCache将前后页号的从CentralCache获取的Span合并成一个大Span挂起

获得CentralCache的Span后，对Span前后页进行合并，缓解内存碎片外碎片问题。

为了找前后页对应的Span还需要页号和Span的映射，这个映射在上一层已经介绍。
如果前后页在PageCache桶上就可以合并，如果还在被CentralCache使用则不能被合并。
注意：这时不能看Span中use_cout因为在对线程的情况下，可能线程1还没有改变use_cout时，线程2来将这块空间又合并到PageCache。存在线程安全问题

这时添加一个状态IsUse标记这个Span是否被使用

//管理以页为单位的大块空间结构
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) {}
};

当遇到正在使用的Span时停止合并。（前后合并）
和前Span合并大小超过128KB则停止合并（最大桶的大小时128KB）

PageCache

#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 && NumPage < NPAGE);

	//看当前位置桶中是否有Span
	if (!_SpanList[NumPage].Empty()) {
		return _SpanList->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) {
	//对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 f1() {
	std::vector<void*>vet;
	for (int i = 0; i < 10; i++) {
		void* ptr = ConcurrentAlloc(6);
		vet.push_back(ptr);
	}
	for (auto e : vet) {
		ConcurrentFree(e, 6);
	}
}

void f2() {
	std::vector<void*>vet;
	for (int i = 0; i < 10; i++) {
		void* ptr = ConcurrentAlloc(7);
		vet.push_back(ptr);
	}
	for (auto e : vet) {
		ConcurrentFree(e, 7);
	}
}

void TestConcurrentAllocThread() {
	std::thread t1(f1);
	t1.join();
	std::thread t2(f2);
	t2.join();
}

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