<1>Central Cache类设计

根据上图我们可以知道central cache也是一个哈希桶结构，通中挂的是一个个的span并且是双链表，而sapn中又包含了freellist，如下定义：

首先定义一个SpanNode的节点来表示一个个的Span对象

struct SpanNode {
	PAGE_ID _pageId = 0; //大块内存起始页号
	size_t _n = 0; //页的数量

	SpanNode* _next = nullptr;
	SpanNode* _prev = nullptr;

	void* _freeList = nullptr; //自由链表
	size_t _size = 0; //切好小对象大小
	size_t _useCount = 0; //分配给thread cache小块内存数量
	
	bool _isUse = false; //是否正在被使用

};

接着实现一个双链表结构用来将Span挂接起来

class SpanList {
private:
	SpanNode* _head = nullptr; //头节点
	std::mutex _mtx; //桶锁
public:
	SpanList() {
		_head = new SpanNode;
		_head->_next = _head;
		_head->_prev = _head;
	}
	~SpanList() {
		delete _head;
	}
	std::mutex& getMutex() {
		return _mtx;
	}
	SpanNode* Begin() {
		return _head->_next;
	}
	SpanNode* end() {
		return _head;
	}
	bool Empty() {
		return _head->_next == _head;
	}
	void Insert(SpanNode* pos, SpanNode* newSpan) {
		assert(pos && newSpan);
		SpanNode* prev = pos->_prev;

		prev->_next = newSpan;
		newSpan->_prev = prev;

		pos->_prev = newSpan;
		newSpan->_next = pos;	

	}

	void Erase(SpanNode* pos) {
		assert(pos);
		assert(pos != _head);
		SpanNode* prev = pos->_prev;
		SpanNode* next = pos->_next;

		prev->_next = next;
		next->_prev = prev;
	}

	SpanNode* PopFront() {
		SpanNode* ret = _head->_next;
		Erase(ret);
		return ret;
	}

	void PushFront(SpanNode* spanNode) {
		Insert(Begin(), spanNode);
	}
};

Central Cache类定义

//单例模式（懒汉）
class CentralCache {
private:
	SpanList _SpanLists[NFREELIST];
private:
	CentralCache() {}
	CentralCache(const CentralCache&) = delete;
	static CentralCache _sInst; //类外初始化
public:
	static CentralCache* GetInStance() {
		return &_sInst;
	}

	// 获取一个非空的spanNode
	SpanNode* GetOneSpan(SpanList& list, size_t size);

	// 从中心缓存获取一定数量的对象给thread cache
	size_t FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size);

	// 将一定数量的对象释放到span
	void ReleaseListToSpans(void* start, size_t byte_size);
};

接口实现：

// 获取一个非空的spanNode
SpanNode* CentralCache::GetOneSpan(SpanList& list, size_t size) {
	//查看当前桶中的每个SpanNode节点是否有未分配的对象
	SpanNode* it = list.Begin();
	while (it != list.end()) {
		if (it->_freeList != nullptr) {
			return it;
		}
		else {
			it = it->_next;
		}
	}
	//走到这里说明当前桶中每个节点没有未分配的对象了,找pageCache要
	//...
	
}

// 从中心缓存获取一定数量的对象给thread cache
size_t CentralCache::FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size) {
	size_t index = SizeClass::Index(size);

	//这里会涉及多个线程同时从中心缓存获取内存的情况，故应该加锁
	_SpanLists[index].getMutex().lock();
	SpanNode* spanNode = GetOneSpan(_SpanLists[index], size);
	assert(spanNode);
	assert(spanNode->_freeList);

	//从spanNode中获取batchNum个对象,如果不够有多少拿多少
	start = spanNode->_freeList;
	end = start;
	int i = 0;
	int ActualNum = 1;
	while (i < batchNum - 1 && CurNext(end) != nullptr) {
		end = CurNext(end);
		++i;
		++ActualNum;
	}
	//从spanNode将这段链表删除,并统计出该spanNode中自由链表已经使用的数量
	spanNode->_freeList = CurNext(end);
	CurNext(end) = nullptr;
	spanNode->_useCount += ActualNum;
	_SpanLists[index].getMutex().unlock();

	return ActualNum;
}

关于当链表太长时要将内存还给SpanNode时情况稍微有点复杂，需要好好思考一下。因为还回来的链表中的每个节点的地址我们没办法确定是否连续的。这就导致有可能换回来一个链表但其中的节点分布于_SpanList的多个SpanNode节点中，这就需要对其筛选让其进入不同的桶中（这部分代码我们放到Page Cache中来实现，因为Central Cache中的SpanNode节点都是Page Cache分配给他的），这里我们先把大概逻辑顺一下如下图：

测试代码如下： 

void TestAddressPage() {
	PAGE_ID id1 = 3000;
	PAGE_ID id2 = 3001;
	char* p1 = (char*)(id1 << PAGE_SHIFT);
	char* p2 = (char*)(id2 << PAGE_SHIFT);
	while (p1 < p2) {
		cout << (void*)p1 << " : " << ((PAGE_ID)p1 >> PAGE_SHIFT) << endl;
		p1 += 8;
	}
}

 执行结果：

由上图我们可以看出，事实和我们的推论是一样的。释放内存代码如下图：

//将一定数量的对象释放到span
void CentralCache::ReleaseListToSpans(void* start, size_t byte_size) {
	size_t index = SizeClass::Index(byte_size);

	_SpanLists[index].getMutex().lock();
	//将传过来的链表一个个头插进对应的spanNode中
	while (start) {
		void* next = CurNext(start);
		//通过链表每个节点的地址来获取对应的spanNode地址
		SpanNode* SpanNode = PageCache::GetInstance()->MapObjectToSpan(start);
		CurNext(start) = SpanNode->_freeList;
		SpanNode->_freeList = start;
		SpanNode->_useCount--; //每头插回来一个已使用数量减一

		//当SpanNode已使用数量为0时说明该节点中的自由链表节点都还回来了,继续归还给PageCache处理
		if (SpanNode->_useCount == 0) {
			//归还给PageCache
			//...
		}

		start = next;
	}
	_SpanLists[index].getMutex().unlock();
}

<1>Page Cache类设计

通过上面的分析我们可以知道Page Cache和Central Cache一样是被多个线程共享的故也应该将其设计为单例模式，上面在讲Central Cache释放逻辑是提到我们要根据自由链的节点地址找到其对应SpanNode节点故我们还应该添加其映射关系这里我们采用unordered_map进行映射。此外我们还需要对Central Cache提供对象的接口、回收SpanNode的接口、获取映射关系的接口，如下定义：

static const size_t NPAGES = 129; //下标是从0开始的

class PageCache {
private:
	SpanList _pageLists[NPAGES];
	std::unordered_map<PAGE_ID, SpanNode*> _idSpanNodeMap;
	std::mutex _PageMtx;
private:
	PageCache() {};
	PageCache(const PageCache&) = delete;
	static PageCache _sInst;
public:
	static PageCache* GetInstance() {
		return &_sInst;
	}

	std::mutex& GetMutex() {
		return _PageMtx;
	}
	//获取n页SpanNode
	SpanNode* NewSpan(size_t n);

	//获取从对象到SpanNode的映射
	SpanNode* MapObjectToSpanNode(void* obj);

	//释放空闲SpanNode回到PageCache，并合并相邻的SpanNode
	void ReleaseSpanNodeToPageCache(SpanNode* SpanNode);
};

代码实现：

当Page Cache中无内存时我们需要向系统取申请内存，故我们应该提供一个向系统申请内存的接口，如下

windows和Linux下如何直接向堆申请页为单位的大块内存：

// 去堆上按页申请空间
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
	// linux下sbrk unmmap等
#endif
}

static const size_t NPAGES = 129; //下标是从0开始的

PageCache PageCache::_sInst;

//获取n页SpanNode
SpanNode* PageCache::NewSpan(size_t n) {
	assert(n > 0);
	//大于128页直接向堆申请内存
	if (n > NPAGES - 1) {
		void* ptr = SystemAlloc(n);
		SpanNode* spanNode = new SpanNode;
		spanNode->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT; //右移13位相当于除8k
		spanNode->_n = n;

		_idSpanNodeMap[spanNode->_pageId] = spanNode; //建立映射方便释放内存
		return spanNode;
	}

	//检查对应桶中是否有sapnNode
	if (!_pageLists[n].Empty()) {
		SpanNode* nSpanNode = _pageLists[n].PopFront();
		//将其中节点建立映射关系映射
		for (PAGE_ID i = 0; i < nSpanNode->_n; i++) {
			_idSpanNodeMap[nSpanNode->_pageId + i] = nSpanNode;
		}
		return nSpanNode;
	}

	//检查后面桶中是否有SpanNode节点，有则进行切分
	for (size_t i = n + 1; i < NPAGES; i++) {
		if (!_pageLists[i].Empty()) {
			//找到第i个桶不为空时先将其取出，切分出n个后在放回相应桶中
			SpanNode* ISpanNode = _pageLists[i].PopFront();
			//开辟出一个节点进行切分,然后返回
			SpanNode* NSpanNode = new SpanNode;
			NSpanNode->_n = n;
			NSpanNode->_pageId = ISpanNode->_pageId;

			ISpanNode->_pageId += n;
			ISpanNode->_n -= n;

			_pageLists[ISpanNode->_n].PushFront(ISpanNode);

			//存储ISpanNode起始页号映射方便回收
			_idSpanNodeMap[ISpanNode->_pageId] = ISpanNode;
			_idSpanNodeMap[ISpanNode->_pageId + ISpanNode->_n - 1] = ISpanNode;

			for (PAGE_ID i = 0; i < NSpanNode->_n; i++) {
				_idSpanNodeMap[NSpanNode->_pageId + i] = NSpanNode;
			}

			return NSpanNode;
		}
	}

	//走到这里说明后面的桶中没有剩余的SpanNode节点,找系统申请一页
	SpanNode* bigSpanNode = new SpanNode;
	void* ptr = SystemAlloc(NPAGES - 1);
	bigSpanNode->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT; 
	bigSpanNode->_n = NPAGES - 1;

	_pageLists[bigSpanNode->_n].PushFront(bigSpanNode); //将申请的大页span放入哈希桶中

	return NewSpan(n); //重新调用进行切分

}
//获取从对象到SpanNode的映射
SpanNode* PageCache::MapObjectToSpanNode(void* obj) {
	PAGE_ID id = (PAGE_ID)obj >> PAGE_SHIFT;

	std::unique_lock<std::mutex> lock(PageCache::GetInstance()->GetMutex()); //出作用域自动解锁
	auto ret = _idSpanNodeMap.find(id);
	if (ret != _idSpanNodeMap.end()) {
		//找到了进行返回
		return ret->second;
	}
	else {
		//找不到说明出问题了直接报错
		assert(false); 
		return nullptr;
	}
}

//释放空闲SpanNode回到PageCache，并合并相邻的SpanNode
void PageCache::ReleaseSpanNodeToPageCache(SpanNode* spanNode) {

	//大于128页直接还给堆
	if (spanNode->_n > NPAGES - 1) {
		//根据页号算出响应的地址,然后进行释放
		void* ptr = (void*)(spanNode->_pageId << PAGE_SHIFT);
		SystemFree(ptr);
		return;
	}

	//向前合并
	while (1) {
		PAGE_ID prev = spanNode->_pageId - 1;
		//找不到跳出循环
		auto ret = _idSpanNodeMap.find(prev);
		if (ret == _idSpanNodeMap.end()) {
			break;
		}
		//prevSpanNode正在被使用跳出循环
		SpanNode* prevSpanNode = ret->second;
		if (prevSpanNode->_isUse == true) {
			break;
		}
		//合并出超出128kb的spanNode不进行管理
		if (prevSpanNode->_n + spanNode->_n > NPAGES - 1) {
			break;
		}

		//进行合并
		spanNode->_pageId = prevSpanNode->_pageId;
		spanNode->_n += prevSpanNode->_n;

		_pageLists[prevSpanNode->_n].Erase(prevSpanNode);

		delete prevSpanNode;

	}

	//向后合并
	while (1) {
		PAGE_ID next = spanNode->_pageId - 1;
		//找不到跳出循环
		auto ret = _idSpanNodeMap.find(next);
		if (ret == _idSpanNodeMap.end()) {
			break;
		}
		//prevSpanNode正在被使用跳出循环
		SpanNode* nextSpanNode = ret->second;
		if (nextSpanNode->_isUse == true) {
			break;
		}

		//合并出超出128kb的spanNode不进行管理
		if (nextSpanNode->_n + spanNode->_n > NPAGES - 1) {
			break;
		}

		//进行合并
		spanNode->_n += nextSpanNode->_n;
		_pageLists[nextSpanNode->_n].Erase(nextSpanNode);
		delete nextSpanNode;
	}

	//放回哈希桶中
	_pageLists[spanNode->_n].PushFront(spanNode);
	spanNode->_isUse = false;

	存储ISpanNode起始页号映射方便回收
	_idSpanNodeMap[spanNode->_pageId] = spanNode;
	_idSpanNodeMap[spanNode->_pageId + spanNode->_n - 1] = spanNode;
}

6、性能分析

参考了一段别人的测试代码，进行测试： 测试代码链接