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

CentralCache整体设计思路

当线程缓存没有内存时，要中心缓存申请内存。
CentralCache被所有线程共有，需要加锁，为了减少竞争锁资源的力度，选择采用哈希桶锁的方式管理中心缓存层的内存。
eg:如果线程1需要1号桶，线程2需要2号桶。桶的编号不同，锁不同所以不存在竞争。减少的锁的竞争。
哈希中每个桶的大小和ThreadCache规则相同

中心缓存中的以页为单位的大块空间靠span管理。与ThreadCache不同，CentralCache哈希桶保存的不是内存块，而是span链表。

当ThreadCache桶中没有内存向CentralCache要时，先根据要申请的大小(eg：8byte)找到CentralCache中的那个桶，再将8byte桶中的span切分成多个8byte，用链表管理起来。给ThreadCache空间一次给一个span。当8byte桶的第一份span被线程申请完毕后再根据链表找下一个span.

由上可知，随着申请内存大小的不同，span的大小也不同

当CentralCache某个桶内存使用完毕，会再向PageCache申请空间

ThreadCache释放空间
当ThreadCache中桶的内存太大，CentralCache还负责回收内存。
CentralCache的中挂的span中use_count记录分配了多少个对象出去，分配一个对象给Threadache，就++use_count（Span中的成员变量）。

当ThreadCache过长或者线程销毁，则会将内存释放回CentralCache中的，释放回来时use_count。当use_count减到0时则表示所有对象都回到了span，则将span释放回PageCache，PageCache中会对前后相邻的空闲页进行合并。解决内存碎片问题。

CentralCache结构设计

Span设计
设一页的大小是8K
在32位单进程内存最多分成页数为：
2 ^ 32 / 2 ^ 13(8K) =2 ^ 19个页，size_t类型可以存下

在64位单进程内存最多分成页数为：
2 ^ 64 / 2 ^ 13=2 ^ 51 个页，size_t类型存不下，unsigned long long 存

这时为了解决可以使用条件编译解决。

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

注意这里不能调换64与32的关系。因为在32位下_WIN32定义，_WIN64没有定义
x64下_WIN32与_WIN64都有定义。所以要把_WIN64放在最前，有_WIN64定义一定是64位。

X64

X86

span要设计成带头双向循环链表，Span与SpanList的基本结构为

//管理以页为单位的大块空间结构
struct Span {
	PAGE_ID _PageID;//大块内存页号
	size_t _Num;//页数量
	Span* _next;//双向链表
	Span* _prev;

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

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

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

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;
	}

	void Erase(Span* pos) {
		assert(pos != _head);
		Span* prev = pos->_prev;
		Span* next = pos->_next;
		prev->_next = next;
		next->_prev = prev;
		//还给下一层的PageCache
	}
};

中心缓存只有一个，每个线程都可以看见，所以中心缓存类采用单例模式

C++ CentralCache代码

1.公共资源头文件（添加了Span类以及SpanList类，补充了计算中心缓存要给线程缓存多少空间块 ForMemory）

#pragma once

#include<iostream>
#include<vector>
#include<time.h>
#include<assert.h>
#include<thread>
#include<mutex>
#include<algorithm>

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

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

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

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


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

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

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

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

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

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

//计算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{
			assert(false);
			return -1;
		}
	}
	//计算在哈希桶的那个位置
	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向中心缓存申请的页的数量
	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;
	}
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* _next;//双向链表
	Span* _prev;

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

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

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

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;
	}

	void Erase(Span* pos) {
		assert(pos != _head);
		Span* prev = pos->_prev;
		Span* next = pos->_next;
		prev->_next = next;
		next->_prev = prev;
		//还给下一层的PageCache
	}
};

2.线程申请空间ConcurrentAlloc.h（每个线程都会有一个独立的ThreadCache对象，实现无锁）

#pragma once

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

//线程调用申请ThreadCache空间
static void* ConcurrentAlloc(size_t size) {
	//获取线程自己的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) {
	//释放时每个线程一定有tls_threadcache
	assert(tls_threadcache != nullptr);
	tls_threadcache->ReleaseSpace(ptr,size);
}

3.线程缓存层ThreadCache.h（补充了线程缓存层向中心缓存层申请空间的代码）

#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);
	}

	//向中心缓存申请空间
	void* RequestFromCentralCache(size_t index, size_t size) { 
		//慢开始调节算法
		size_t Num = std::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);//将链表的下一个节点插入桶中，头节点返回给ThreadCache
			return begin;
		}
	}
};

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

4.中心缓存层核心代码

CentralCache.h

#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);
};

CentralCache.cpp

#include"CentralCache.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;

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

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

代码位置

Github
Gitee