高并发内存池 - Page Cache 中关于 Span 的获取项目 gitee 链接 高并发内存池项目项目 github 链接 高并发内存池项目在上一篇文章中最终Page Cache可实现的代码目前如下Page Cache.h#pragmaonce#includeComm.hclassPageCache{public:staticPageCache*GetInstance(){return_sInst;}Span*NewSpan(size_t k);//要的 Span 长度为几页private:std::mutex _pageMtx;//桶锁在 PageCache 中是行不通的需要整体上锁SpanList _spanLists[NPAGES];PageCache(){}PageCache(constPageCache)delete;staticPageCache _sInst;};Page Cache.cpp#includePageCache.hstaticPageCache _sInst;在本篇文章中我们将完善Page Cache中的部分相关逻辑使项目进一步推进。首先我们要回到Central Cache层中去填一填我们挖下的坑。Central Cache.h#pragmaonce#includeComm.hclassCentralCache//单例模式{public:staticCentralCache*GetInstance(){return_sInst;}size_tFetchRangeObj(void*start,void*end,size_t batchNum,size_t size);Span*GetOneSpan(SpanListlist,size_t size);staticCentralCache _sInst;private:CentralCache(){}CentralCache(constCentralCacheCc)delete;SpanList _spanlists[NFREELIST];//和 ThreadCache 的自由链表是一一对应的};这是我们已经实现的Central Cache的头文件其中 FetchRangeObj() 的整体逻辑我们已经实现完毕size_tCentralCache::FetchRangeObj(void*start,void*end,size_t batchNum,size_t size){size_t indexSizeClass::Index(size);_spanlists[index]._mtu.lock();//每个桶里都有一个锁只对桶里的锁进行申请提高并发效率Span*spanGetOneSpan(_spanlists[index],size);assert(span);assert(span-_freelist);//从 span 中获取 batchNum 个对象//如果不够 batchNum 个返回尽量多个startspan-_freelist;endstart;size_t actualNum1;//至少会返回一个对象size_t i0;while(ibatchNum-1ObjNext(end)!nullptr){endObjNext(end);i;actualNum;}span-_freelistObjNext(end);ObjNext(end)nullptr;_spanlists[index]._mtu.unlock();returnactualNum;}对于其中的GetOneSpan()那时我们并未实现而是说这部分内容需要我们对Page Cache的整体结构明晰后才可以进一步实现那么现在是时候了。GetOneSpan()函数的核心任务是在SpanList中查找Span在自由链表中是否还有空间可以被分配可分配直接返回不可分配向Page Cache申请Span所以第一步就是遍历SpanList判断其中的Span是否还有可分配的空间。在设计SpanList时我们将其设计为带头循环双向链表所以实现遍历功能只需要再添加返回头节点Begin()和尾节点End()两个接口即可。在这里将 PushFront() 接口也一并带出后面会提到classSpanList{public:SpanList(){_headnewSpan;_head-_next_head;_head-_prev_head;}voidInsert(Span*pos,Span*newSpan){assert(pos);//使用 assert 断言提高调试效率assert(newSpan);newSpan-_prevpos;newSpan-_nextpos-_next;pos-_nextnewSpan;pos-_next-_prevnewSpan;}voidPushFront(Span*span){Insert(Begin(),span);}Span*Begin()//设计返回头尾指针的结构是为了实现循环遍历{return_head-_next;}Span*End(){return_head;}voidErase(Span*pos){assert(pos);assert(pos!_head);//不能对头结点进行删除pos-_prev-_nextpos-_next;pos-_next-_prevpos-_prev;}private:Span*_head;public:std::mutex _mtu;//桶锁};将接口完善后我们便可实现GetOneSpan()以下部分的逻辑Span*CentralCache::GetOneSpan(SpanListlist,size_t size){Span*curlist.Begin();while(cur!list.End()){if(cur-_freelist!nullptr)//如果一个 Span 中仍有空间可以分配那自由链表一定不为空returncur;curcur-_next;}}如果此时代码跳出循环发现SpanList中已经没有可以再分配的空间时此时便要向Page Cache申请Span关键是申请大小为几页的 Span 呢这部分内容和之前一样TcMalloc给了我们一个解决方案staticsize_tNumMovePage(size_t size)//传入的参数为要申请的字节数{size_t numNumMoveSize(size);size_t npagenum*size;npagePAGE_SHIFT;//PAGE_SHIFT 为页大小为 2 的几次幂if(npage0)npage1;//不满一页补到一页returnnpage;}它利用了之前的NumMoveSize()此函数限制了一次最多Thread Cache向Central Cache申请多少个小空间个数的多少由小空间的大小决定空间越大越少越小越多。//thread cache 一次从 central cache 中获取多少个staticsize_tNumMoveSize(size_t size){if(size0)return0;intnumMAX_BYTES/size;if(num2)num2;if(num512)num512;returnnum;}所以在NumMovePage()中利用了这个函数申请的空间大小 ✖️NumMoveSize(size)最后得出的结果再除以一页的空间大小如果不足一页补为一页这样就能做到空间越大的申请的页数越多空间越小的申请的页数越小算法十分巧妙值得学习。所以接下来便可以向Page Cache去申请空间了Span*CentralCache::GetOneSpan(SpanListlist,size_t size){Span*curlist.Begin();while(cur!list.End()){if(cur-_freelist!nullptr)//如果一个 Span 中仍有空间可以分配那自由链表一定不为空returncur;curcur-_next;}//此时 Spanlist 已经为空需要向 PageCache 申请Span*newSpanPageCache::GetInstance()-NewSpan(SizeClass::NumMovePage(size));}当成功申请出空间后拿到的newSpan中的自由链表为空它的整块大空间并没有被切割过所以要将其分成小块填满它自己的自由链表。其中要注意拿到的newSpan现在并未插入到Central Cache对应的哈希桶中所以也要进行插入操作用到上面的PushFront()。Span*CentralCache::GetOneSpan(SpanListlist,size_t size){Span*curlist.Begin();while(cur!list.End()){if(cur-_freelist!nullptr)//如果一个 Span 中仍有空间可以分配那自由链表一定不为空returncur;curcur-_next;}//此时 Spanlist 已经为空需要向 PageCache 申请Span*newSpanPageCache::GetInstance()-NewSpan(SizeClass::NumMovePage(size));//从 PageCache 中获得的是一整个 Span需要分割成小块塞入自由链表中char*start(char*)(newSpan-_pageIdPAGE_SHIFT);//使用页号去计算 Span 的起始地址size_t bytesnewSpan-_nPAGE_SHIFT;char*endstartbytes;newSpan-_freeliststart;startsize;void*tailnewSpan-_freelist;while(startend){ObjNext(tail)start;tailObjNext(tail);startsize;}list.PushFront(newSpan);returnnewSpan;}关于GetOneSpan()的逻辑讲解完毕也算是把这个坑填上了此时我们再回看Cantral Cache.cpp的内容#includeCentralCache.h#includePageCache.hCentralCache CentralCache::_sInst;Span*CentralCache::GetOneSpan(SpanListlist,size_t size){Span*curlist.Begin();while(cur!list.End()){if(cur-_freelist!nullptr)//如果一个 Span 中仍有空间可以分配那自由链表一定不为空returncur;curcur-_next;}//此时 Spanlist 已经为空需要向 PageCache 申请Span*newSpanPageCache::GetInstance()-NewSpan(SizeClass::NumMovePage(size));//从 PageCache 中获得的是一整个 Span需要分割成小块塞入自由链表中char*start(char*)(newSpan-_pageIdPAGE_SHIFT);//使用页号去计算 Span 的起始地址size_t bytesnewSpan-_nPAGE_SHIFT;char*endstartbytes;newSpan-_freeliststart;startsize;void*tailnewSpan-_freelist;while(startend){ObjNext(tail)start;tailObjNext(tail);startsize;}list.PushFront(newSpan);returnnewSpan;}size_tCentralCache::FetchRangeObj(void*start,void*end,size_t batchNum,size_t size){size_t indexSizeClass::Index(size);_spanlists[index]._mtu.lock();//每个桶里都有一个锁只对桶里的锁进行申请提高并发效率Span*spanGetOneSpan(_spanlists[index],size);assert(span);assert(span-_freelist);//从 span 中获取 batchNum 个对象//如果不够 batchNum 个返回尽量多个startspan-_freelist;endstart;size_t actualNum1;//至少会返回一个对象size_t i0;while(ibatchNum-1ObjNext(end)!nullptr){endObjNext(end);i;actualNum;}span-_freelistObjNext(end);ObjNext(end)nullptr;_spanlists[index]._mtu.unlock();returnactualNum;}会发现其实现在一切都解释的通了代码逻辑我们十分清晰。那现在的代码中还有一个我们尚未实现的地方Span*newSpanPageCache::GetInstance()-NewSpan(SizeClass::NumMovePage(size));NewSpan函数还未实现我们只实现了函数声明所以接下来我们讲解NewSpan函数的实现逻辑。NewSpan会先尝试在对应页数大小 Span 的 SpanList 中查找是否还有剩余的Span可供分配所以可以先写出这部分逻辑Span*PageCache::NewSpan(size_t k){assert(k0kNPAGES);//先检查第 k 个桶中是否还有 Spanif(!_spanLists[k].Empty())return_spanLists-PopFront();}这里的PopFront()接口需要在SpanList类中添加只是简单的节点操作。Span*PopFront(){Span*spanPop_head-_next;Erase(spanPop);returnspanPop;}这也恰恰说明了在完成一个项目时不要要求一开始就尽善尽美而是随着不断的深入如果有需求我们再实现不断补充、不断完善。如果走到代码的最后说明此时没有剩余的Span可供分配此时就要去找是否有更大页数的Span可供我们分割满足我们的需求所以补充逻辑Span*PageCache::NewSpan(size_t k){assert(k0kNPAGES);//先检查第 k 个桶中是否还有 Spanif(!_spanLists[k].Empty())return_spanLists-PopFront();//走到这里说明桶中为空要遍历 Page Cache 是否有大页 Span 可分割for(intik;iNPAGES;i){if(_spanLists[i].Empty()){Span*spanN_spanLists[i].PopFront();Span*spanK;spanK-_pageIdspanN-_pageId;spanK-_nk;spanN-_pageIdk;spanN-_n-k;_spanLists[spanN-_n].PushFront(spanK);returnspanK;}}这里设计的非常巧妙仅仅是对于页号的变更与变化存储页数就完成了Span的分割最后将分割出的Span返回剩余的插入到对应页数的SpanList中。此时的代码如果走到最后说明连大块的Span都没有此时PageCahe就要使用最后的手段了向堆申请内存。申请内存时我们要申请大块内存因为向系统频繁地申请内存会造成不小的性能消耗比如用户态、系统态切换上下文切换所以最好一次申请比较大块的内存然后我们自行对内存管理这样可以减少向系统申请的频率提高效率。就像大学生跟父母要钱一样如果每天买一瓶水、吃一顿饭都要打一个电话那父母只会不胜其烦所以不如一次性向父母拿足够一个月的生活费自己支配这样效率就很高。补上最后的逻辑Span*PageCache::NewSpan(size_t k){assert(k0kNPAGES);//先检查第 k 个桶中是否还有 Spanif(!_spanLists[k].Empty())return_spanLists-PopFront();//走到这里说明桶中为空要遍历 Page Cache 是否有大页 Span 可分割for(intik;iNPAGES;i){if(_spanLists[i].Empty()){Span*spanN_spanLists[i].PopFront();Span*spanK;spanK-_pageIdspanN-_pageId;spanK-_nk;spanN-_pageIdk;spanN-_n-k;_spanLists[spanN-_n].PushFront(spanK);returnspanK;}}//走到这里说明也没有大页 Span所以此时需要向堆申请空间Span*bigSpannewSpan;void*ptrSystemAlloc(NPAGES-1);//申请一个最大页数的 SpanbigSpan-_pageId(PAGE_ID)ptrNPAGES;bigSpan-_nNPAGES-1;_spanLists[bigSpan-_n].PushFront(bigSpan);returnNewSpan(k);//递归调用自己此时一定有大页 Span 可供分割}SystemAlloc的具体实现根据不同的系统使用不同的系统接口以下为 MAC 的实现inlinestaticvoid*SystemAlloc(size_t kpage){// 获取系统页大小通常是 4096即 112// 如果 kpage 已经是页数直接计算总字节数size_t sizekpagePAGE_SHIFT;// 假设一页 4096 2^12// MAP_ANON: 匿名映射不关联文件// MAP_PRIVATE: 私有映射// PROT_READ | PROT_WRITE: 可读可写void*ptrmmap(nullptr,size,PROT_READ|PROT_WRITE,MAP_PRIVATE|MAP_ANON,-1,0);if(ptrMAP_FAILED){throwstd::bad_alloc();}returnptr;}所以整个NewSpan()函数的逻辑都已讲解完毕可以观察到设计的结构十分巧妙思路逻辑清晰安排合理正如侯捷老师在**《STL 源码解析》**中说的观摩大家风范的代码有助于我们提高。现在我们思考下一个问题在Page Cache层中的大锁被申请后是否需要释放Central Cache中的桶锁答案是要释放支持不能释放的读者可能会这么想释放了后如果别的线程申请资源一样会没有Span可供使用那这样释放了也完全没有意义所以不如不释放保证前程安全。但这恰恰忽略了一个重点只考虑了申请的情况没有考虑有线程要释放资源的情况。不对桶锁进行释放那线程无法将空间资源释放到Span中这样会影响效率。所以最终的最佳实践是对桶锁进行释放好让Central Cache可以回收资源方便别的资源申请。所以优化后的代码如下Span*CentralCache::GetOneSpan(SpanListlist,size_t size){Span*curlist.Begin();while(cur!list.End()){if(cur-_freelist!nullptr)//如果一个 Span 中仍有空间可以分配那自由链表一定不为空returncur;curcur-_next;}//先将 Central Cache 的桶锁解开其他线程有归还空间的需求list._mtu.unlock();//此时 Spanlist 已经为空需要向 PageCache 申请PageCache::GetInstance()-_pageMtx.lock();Span*newSpanPageCache::GetInstance()-NewSpan(SizeClass::NumMovePage(size));PageCache::GetInstance()-_pageMtx.unlock();//从 PageCache 中获得的是一整个 Span需要分割成小块塞入自由链表中char*start(char*)(newSpan-_pageIdPAGE_SHIFT);//使用页号去计算 Span 的起始地址size_t bytesnewSpan-_nPAGE_SHIFT;char*endstartbytes;newSpan-_freeliststart;startsize;void*tailnewSpan-_freelist;while(startend){ObjNext(tail)start;tailObjNext(tail);startsize;}list._mtu.lock();list.PushFront(newSpan);returnnewSpan;}首先这里的实现要注意的第一点是1. 对桶锁释放后在进行涉及桶的操作时我们还要进行上锁list._mtu.lock();list.PushFront(newSpan);PushFront()操作中设计对桶进行变动所以要对锁进行申请那上面将Span分割成小块的内存为何用保证线程安全呢因为这一块Span还未链入哈希桶中只有申请的线程能看见它所以它本身就为线程安全的对它进行分割不用加锁。这样对于GetOneSpan的优化已经全然明晰。