C++项目 – 高并发内存池(四)Page Cache
文章目录
- C++项目 -- 高并发内存池(四)Page Cache
- 一、PageCache设计
- 1.PageCache的整体结构
- 2.Common.h的更新
- 3.CentralCache中的GetOneSpan函数的实现
- 4.PageCache类设计
- 5.代码实现
- 二、申请内存过程联调
一、PageCache设计
1.PageCache的整体结构
PageCache也是哈希桶结构,每个哈希桶下挂载的是Span,但是与Central Cache挂载的Span有两点不同:
- 哈希桶的映射规则不同:PageCache的映射规则为第一个哈希桶挂载的Span的size是一页,第二个哈希桶挂载的Span的size是两页,一直到最后一个哈希桶挂载的Span的size是128页;
- Span结构不同:PageCache挂载的Span不切分成小内存块,因为其是直接分给CentralCache的;
这样的设计方便以页为单位的分配和回收;
申请内存:
- 当central cache向page cache申请内存时,page cache先检查对应位置有没有span,如果没有则向更大页寻找一个span,如果找到则分裂成两个。比如:申请的是4页page,4页page后面没有挂span,则向后面寻找更大的span,假设在10页page位置找到一个span,则将10页page span分裂为一个4页page span和一个6页page span。
- 如果找到_spanList[128]都没有合适的span,则向系统使用mmap、brk或者是VirtualAlloc等方式申请128页page span挂在自由链表中,再重复1中的过程。
- 需要注意的是central cache和page cache 的核心结构都是spanlist的哈希桶,但是他们是有本质区别的,central cache中哈希桶,是按跟thread cache一样的大小对齐关系映射的,他的spanlist中挂的span中的内存都被按映射关系切好链接成小块内存的自由链表。而page cache 中的spanlist则是按下标桶号映射的,也就是说第i号桶中挂的span都是i页内存。
释放内存:
- 如果central cache释放回一个span,则依次寻找span的前后page id的没有在使用的空闲span,看是否可以合并,如果合并继续向前寻找。这样就可以将切小的内存合并收缩成大的span,减少内存碎片。
2.Common.h的更新
- 定义全局静态变量
NPAGES
,是Page Cache中的哈希桶的大小,定义为129是因为Page Cache是按照页数进行映射的,一页的span映射到下标为1的桶,因此桶的有效下标从1开始比较偏方便; - 定义全局静态变量
PAGE_SHIFT
,用于页与字节单位之间的转换,一页是8K Byte,也就是2 ^ 13 Byte
; - 将ObjectPool.h中从堆中直接申请空间的代码移动到这里:
- 在
SizeClass
类中增加NumMovePage
函数,用于根据当前对象size计算central cache一次向page cache获取多少页的span:
先根据一个对象的size获取该对象批量申请内存块的上限值,再计算一次申请内存上限的字节大小,转换成页为单位(/=8k);
- 完善SpanList类的功能,增加类似迭代器(返回Span*)、头插、头删、判空功能;
3.CentralCache中的GetOneSpan函数的实现
特别注意加解锁问题
- 先查看当前的spanList中有没有非空的span,有就返回;如果没有就需要向PageCache申请Sapn;
- 申请下的Span是页为单位的大块内存,需要切分,获取大块内存的起始地址和终止地址;
- 大块内存切分成小内存块,尾插进freeList:地址是连续的,缓存利用率高;
- 所有内存切分完成后,再将该span头插进spanList;
- 在执行
GetOneSpan
函数之前,就已经加了桶锁,如果central cache需要向page cache申请span,就先将桶锁解除,因为下面的程序需要到page cache中申请内存,如果不解锁,其他线程无法访问这个桶,就会造成释放空间阻塞; - page cache的
NewSpan
函数涉及递归,加锁比较麻烦,因此直接在调用这个函数之前就加上锁,就相当于加上了全局锁; - 在得到新的span之后,不用立即加上桶锁,因为此时这个span还没有挂载到spanList,其他线程访问不到;
- 在切分好span并挂载到spanList之前再加上桶锁;
Span* CentralCache::GetOneSpan(SpanList& spanList, size_t size) {//先检查该SpanList有没有未分配的SpanSpan* it = spanList.Begin();while (it != spanList.End()) {if (it->_freeList != nullptr) {return it;}else {it = it->_next;}}//先把central cache 的桶锁解掉,这样如果其他线程释放对象回来,就不会被阻塞spanList._mtx.unlock();//SpanList中没有空闲的Span,需要向page cache申请//在此处加上page cache的全局锁,NewSpan的所有操作都是加锁进行的PageCache::GetInstance()->_pageMtx.lock();Span* span = PageCache::GetInstance()->NewSpan(SizeClass::NumMovePage(size));PageCache::GetInstance()->_pageMtx.unlock();//从page cache获取到了新的span,需要进行切分//无需在此加上桶锁,因为该span还没有放到spanList中,其他线程访问不到//计算span大块内存的起始地址和大块内存的大小(字节数)char* start = (char*)(span->_pageID << PAGE_SHIFT);size_t bytes = span->_n << PAGE_SHIFT;char* end = start + bytes;//把大块内存切成自由链表链接起来//先切一块下来做头,方便尾插span->_freeList = start;start += size;void* tail = span->_freeList;while (start < end) {NextObj(tail) = start;tail = start;start += size;}//在span挂载到spanList之前加上桶锁spanList._mtx.lock();spanList.PushFront(span);return span;
}
4.PageCache类设计
- PageCache类设计为单例模式,与CentralCache类似,
- PageCache按span的页数进行映射,下标为1的_spanList下挂载的都是1页大小的Span;
- 需要加锁,这里不能使用桶锁了,因为如果当前size的SapnList没有span对象,就需要从更大size的spanlist中获取,再进行拆分,使用桶锁会造成频繁的加减锁,效率更低下,因此需要加全局锁;
- 最开始page cache是没有挂载任何span的,第一次是向系统申请一个128页的span的,然后再向下分割给更小的span;
例如:把128页的span切分成2页的span和126页的span,2页的span返回给central cache使用,126页的span挂到126号桶中; NewSpan
函数的功能是返回一个k页的span;- 先检查第k个桶里面有没有span,如果有就pop第一个span;如果为空,就检查后面的桶有没有span,如果有,可以进行切分;
- 切分成一个k页的span和一个n-k页的span,k页的span返回给central cache,n-k页的span挂到第n-k桶中去;
- 如果没有更大的page,就需要向堆申请内存;知道页的地址就可以计算出页的id;
- 在向堆申请了128页的内存后,直接插入到最有一个桶,然后递归调用本函数,就可以完成切分;
- 加解锁:使用普通锁是无法解决递归的加锁的,递归进去的时候程序已经被锁住,无法向下进行,造成死锁,可以使用递归互斥锁;
也可以将重复调用的部分设计成子函数,就可以使用普通锁; - 我们选择直接在central cache的
GetOneSpan
函数调用NewSpan
函数之前就加锁,而不是在NewSpan
函数中加锁;
Span* PageCache::NewSpan(size_t k)
{assert(k > 0 && k < NPAGES);//先检查第k个桶里面有没有spanif (!_spanLists[k].Empty()) {//有就返回return _spanLists[k].PopFront();}//没有就需要检查后面的桶有没有更大的span,如果有可以拆分for (size_t i = k + 1; i < NPAGES; i++) {if (!_spanLists[i].Empty()) {Span* nspan = _spanLists[i].PopFront();Span* kspan = new Span;//在nspan头部且下一个k页的span//kspan返回//nspan剩下的部分挂载到相应的桶上kspan->_pageID = nspan->_pageID;kspan->_n = k;nspan->_pageID += k;nspan->_n -= k;_spanLists[nspan->_n].PushFront(nspan);return kspan;}}//走到这里说明没有更大的span了,需要向堆申请一个128页的大块内存Span* bigSpan = new Span;void* ptr = SystemAlloc(NPAGES - 1);bigSpan->_pageID = (PAGE_ID)ptr >> PAGE_SHIFT;bigSpan->_n = NPAGES - 1;_spanLists[NPAGES - 1].PushFront(bigSpan);//此时需要将_spanLists中的128页的内存切分,递归调用一下return NewSpan(k);
}
5.代码实现
common.h
#pragma once
//公共头文件#include <iostream>
#include <vector>
#include <assert.h>
#include <thread>
#include <mutex>
#include <algorithm>
using std::cout;
using std::endl;
using std::vector;static const size_t MAX_BYTES = 256 * 1024; //ThreadCache能分配对象的最大字节数
static const size_t NFREELIST = 208; //central cache 最大的哈希桶数量
static const size_t NPAGES = 129; //page cache 哈希桶的数量
static const size_t PAGE_SHIFT = 13; //页与字节的转换#ifdef _WIN32#include<windows.h>
#else
//linux
#endif#ifdef _WIN64typedef unsigned long long PAGE_ID;
#elif _WIN32typedef size_t PAGE_ID;
#elif//linux#endif//直接去堆上申请空间
inline static void* SystemAlloc(size_t kpage) {
#ifdef _WIN32void* ptr = VirtualAlloc(0, kpage << 13, MEM_COMMIT | MEM_RESERVE, PAGE_READWRITE);
#else#endif // _WIN32if (ptr == nullptr) {throw std::bad_alloc();}return ptr;
}// 访问obj的前4 / 8字节地址空间
static void*& NextObj(void* obj) {return *(void**)obj;
}//自由链表类,用于管理切分好的小内存块
class FreeList {
public:void Push(void* obj) {assert(obj);//头插NextObj(obj) = _freeList;_freeList = obj;}//范围插入void PushRange(void* start, void* end) {assert(start);assert(end);NextObj(end) = _freeList;_freeList = start;}void* Pop() {assert(_freeList);//头删void* obj = _freeList;_freeList = NextObj(obj);return obj;}bool Empty() {return _freeList == nullptr;}//用于实现thread cache从central cache获取内存的慢开始算法size_t& MaxSize() {return _maxSize;}
private:void* _freeList = nullptr;size_t _maxSize = 1;
};// 管理对齐和哈希映射规则的类
class SizeClass {
public://对齐规则// 整体控制在最多10%左右的内碎片浪费// [1,128] 8byte对齐 freelist[0,16)// [128+1,1024] 16byte对齐 freelist[16,72)// [1024+1,8*1024] 128byte对齐 freelist[72,128)// [8*1024+1,64*1024] 1024byte对齐 freelist[128,184)// [64*1024+1,256*1024] 8*1024byte对齐 freelist[184,208)//RoundUp的子函数,根据对象大小和对齐数,返回对象对齐后的大小static inline size_t _RoundUp(size_t size, size_t align) {//if (size % align == 0) {// return size;//}//else {// return (size / align + 1) * align;//}//使用位运算能够得到一样的结果,但是位运算的效率很高return ((size + align - 1) & ~(align - 1));}//计算当前对象size字节对齐之后对应的sizestatic inline size_t RoundUp(size_t size) {assert(size <= MAX_BYTES);if (size <= 128) {//8字节对齐return _RoundUp(size, 8);}else if (size <= 1024) {//16字节对齐return _RoundUp(size, 16);}else if (size <= 8 * 1024) {//128字节对齐return _RoundUp(size, 128);}else if (size <= 64 * 1024) {//1024字节对齐return _RoundUp(size, 1024);}else if (size <= 256 * 1024) {//8KB字节对齐return _RoundUp(size, 8 * 1024);}else {assert(false);}return -1;}//Index的子函数,用于计算映射的哈希桶下标static inline size_t _Index(size_t size, size_t alignShift) {//if (size % align == 0) {// return size / align - 1;//}//else {// return size / align;//}//使用位运算能够得到一样的结果,但是位运算的效率很高//使用位运算需要将输入参数由对齐数改为对齐数是2的几次幂、return ((size + (1 << alignShift) - 1) >> alignShift) - 1;}//计算对象size映射到哪一个哈希桶(freelist)static inline size_t Index(size_t size) {assert(size <= MAX_BYTES);//每个区间有多少个哈希桶static int groupArray[4] = { 16, 56, 56, 56 };if (size <= 128) {return _Index(size, 3);}else if (size <= 1024) {//由于前128字节不是16字节对齐,因此需要减去该部分,单独计算16字节对齐的下标//再在最终结果加上全部的8字节对齐哈希桶个数return _Index(size - 128, 4) + groupArray[0];}else if (size <= 8 * 1024) {return _Index(size - 1024, 7) + groupArray[0] + groupArray[1];}else if (size <= 64 * 1024) {return _Index(size - 8 * 1024, 10) + groupArray[0] + groupArray[1] + groupArray[2];}else if (size <= 256 * 1024) {return _Index(size - 64 * 1024, 13) + groupArray[0] + groupArray[1] + groupArray[2] + groupArray[3];}else {assert(false);}return -1;}//thread cache一次从central cache中获取多少内存块static size_t NumMoveSize(size_t size) {//一次获取的内存块由对象的大小来决定assert(size > 0);//将获取的数量控制在[2, 512]size_t num = MAX_BYTES / size;if (num < 2) {num = 2;}if (num > 512) {num = 512;}return num;}//计算central cache一次向page cache获取多少页的spanstatic size_t NumMovePage(size_t size) {assert(size > 0);//先计算该对象一次申请内存块的上限值size_t num = NumMoveSize(size);//计算上限的空间大小size_t npage = num * size;//转换成page单位npage >>= PAGE_SHIFT;if (npage == 0) {npage = 1;}return npage;}
};struct Span
{PAGE_ID _pageID = 0; // 大块内存起始页的页号size_t _n = 0; // 页的数量Span* _next = nullptr; // 双向链表的结构Span* _prev = nullptr;size_t _objSize = 0; // 切好的小对象的大小size_t _useCount = 0; // 切好小块内存,被分配给thread cache的计数void* _freeList = nullptr; // 切好的小块内存的自由链表bool _isUse = false; // 是否在被使用
};class SpanList {
public:SpanList() {_head = new Span;_head->_next = _head;_head->_prev = _head;}void Insert(Span* pos, Span* newSapn) {assert(pos);assert(newSapn);Span* prev = pos->_prev;prev->_next = newSapn;newSapn->_prev = prev;newSapn->_next = pos;pos->_prev = newSapn;}void Erase(Span* pos) {assert(pos);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;}void PushFront(Span* newSapn) {Insert(Begin(), newSapn);}Span* PopFront() {Span* front = _head->_next;Erase(front);return front;}private:Span* _head; //头节点
public:std::mutex _mtx; //桶锁
};
CentralCache.cpp
#define _CRT_SECURE_NO_WARNINGS 1
#include "CentralCache.h"//单例模式静态成员的定义
CentralCache CentralCache::_sInstance;//从CentralCache获取一定数量的内存对象给ThreadCache
size_t CentralCache::FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size) {//先根据对象size获取对应的spanList下标size_t index = SizeClass::Index(size);//每个线程访问spanList时需要加锁_spanLists[index]._mtx.lock();//获取非空的spanSpan* span = GetOneSpan(_spanLists[index], size);assert(span);assert(span->_freeList);//从span中获取batchNum个对象,若不够,就有多少拿多少start = span->_freeList;end = start;size_t i = 0;size_t actualNum = 1; // 实际拿到的对象数量while (i < batchNum - 1 && NextObj(end) != nullptr) {end = NextObj(end);actualNum++;i++;}//在span中去掉这一段对象span->_freeList = NextObj(end);NextObj(end) = nullptr;span->_useCount += actualNum;_spanLists[index]._mtx.unlock();return actualNum;
}Span* CentralCache::GetOneSpan(SpanList& spanList, size_t size) {//先检查该SpanList有没有未分配的SpanSpan* it = spanList.Begin();while (it != spanList.End()) {if (it->_freeList != nullptr) {return it;}else {it = it->_next;}}//先把central cache 的桶锁解掉,这样如果其他线程释放对象回来,就不会被阻塞spanList._mtx.unlock();//SpanList中没有空闲的Span,需要向page cache申请//在此处加上page cache的全局锁,NewSpan的所有操作都是加锁进行的PageCache::GetInstance()->_pageMtx.lock();Span* span = PageCache::GetInstance()->NewSpan(SizeClass::NumMovePage(size));PageCache::GetInstance()->_pageMtx.unlock();//从page cache获取到了新的span,需要进行切分//无需在此加上桶锁,因为该span还没有放到spanList中,其他线程访问不到//计算span大块内存的起始地址和大块内存的大小(字节数)char* start = (char*)(span->_pageID << PAGE_SHIFT);size_t bytes = span->_n << PAGE_SHIFT;char* end = start + bytes;//把大块内存切成自由链表链接起来//先切一块下来做头,方便尾插span->_freeList = start;start += size;void* tail = span->_freeList;while (start < end) {NextObj(tail) = start;tail = start;start += size;}//在span挂载到spanList之前加上桶锁spanList._mtx.lock();spanList.PushFront(span);return span;
}
PageCache.h
#pragma once
#include "Common.h"//单例模式
class PageCache {
public:static PageCache* GetInstance() {return &_sInstance;}std::mutex _pageMtx; //全局锁//获取一个k页的SpanSpan* NewSpan(size_t k);private:SpanList _spanLists[NPAGES];PageCache() {}PageCache(const PageCache&) = delete;static PageCache _sInstance;
};
PageCache.cpp
#define _CRT_SECURE_NO_WARNINGS 1
#include "PageCache.h"PageCache PageCache::_sInstance;Span* PageCache::NewSpan(size_t k)
{assert(k > 0 && k < NPAGES);//先检查第k个桶里面有没有spanif (!_spanLists[k].Empty()) {//有就返回return _spanLists[k].PopFront();}//没有就需要检查后面的桶有没有更大的span,如果有可以拆分for (size_t i = k + 1; i < NPAGES; i++) {if (!_spanLists[i].Empty()) {Span* nspan = _spanLists[i].PopFront();Span* kspan = new Span;//在nspan头部且下一个k页的span//kspan返回//nspan剩下的部分挂载到相应的桶上kspan->_pageID = nspan->_pageID;kspan->_n = k;nspan->_pageID += k;nspan->_n -= k;_spanLists[nspan->_n].PushFront(nspan);return kspan;}}//走到这里说明没有更大的span了,需要向堆申请一个128页的大块内存Span* bigSpan = new Span;void* ptr = SystemAlloc(NPAGES - 1);bigSpan->_pageID = (PAGE_ID)ptr >> PAGE_SHIFT;bigSpan->_n = NPAGES - 1;_spanLists[NPAGES - 1].PushFront(bigSpan);//此时需要将_spanLists中的128页的内存切分,递归调用一下return NewSpan(k);
}
二、申请内存过程联调
#include "ConcurrentAlloc.h"
#include "ThreadCache.h"void TestConcurrentAlloc1() {void* p1 = ConcurrentAlloc(5);void* p2 = ConcurrentAlloc(78);void* p3 = ConcurrentAlloc(9);void* p4 = ConcurrentAlloc(12);void* p5 = ConcurrentAlloc(100);void* p6 = ConcurrentAlloc(58);cout << p1 << endl;cout << p2 << endl;cout << p3 << endl;cout << p4 << endl;cout << p5 << endl;cout << p6 << endl;
}void TestConcurrentAlloc2() {//6字节申请1024次for (int i = 0; i < 1024; i++) {void* p1 = ConcurrentAlloc(6);cout << p1 << endl;}//第1025次会不会向page cache申请内存void* p2 = ConcurrentAlloc(8);cout << p2 << endl;}int main() {TestConcurrentAlloc2();return 0;
}