C++项目——高并发内存池（3）--central cache整体设计

1.central cache的介绍

1.1框架思想

1.1.1哈希映射

centralcache其实也是哈希桶结构的，并且central cache和thread cacha的哈希映射关系是一致的。目的为了，当thread cache某一个哈希桶下没有内存块时，可以利用之前编写的SizeClass::Index() 直接访问centralcache对应的哈希桶结构以拿到内存空间。

不同的是thread cache桶结构下面挂的是一个一个切好的定长内存块，而central cache桶结构下面挂的是一个一个的SpanList结构，其中的Span是管理以页为单位的大块内存，Span中大块内存被按照映射关系切成一个个小内存块，挂在Span自由链表中。

1.1.2 单例模式

因为，在高并发线程池的整体项目框架下，所有的thread cache都共享一个central cache,所以将它设计成单例模式是和逻辑且安全的。

1.1.3 加桶锁

但是central cache是线程之间共有的，所以线程从这里申请内存时，是需要加锁的。为了减少锁的竞争，central cache使用的是桶锁，意思是，当线程1和线程2同时向同一个桶申请内存时，才会有竞争，可以减少阻塞。

1.2 cenctral cache的作用

• 分配更多的内存给thread cache

因为仅在thread cache给予内存是无锁的，效率更高。所以当它对应哈希映射的哈希桶内没有内存块时，就需要central cache来提供，那如果一次仅仅提供一个对应的内存块，冲突和阻塞现象会加重，即需要一次分配给一串。

• 中央调度工作

由于thread cache一直向centralcache申请某一字节的内存空间，而当这些内存释放时，就会挂在thread cache下的哈希桶结构中，而导致其他线程再向centralcache申请时，没有对应的内存空间，所以centralcache还需要起到调度的作用，当thread cache桶结构挂的内存块过多时，需要拿回来。

 2.Span SpanList

 Span

struct Span
{PAGE_ID _pageId=0;// 大块内存起始页的页号size_t _n=0;//数量Span* _next=nullptr;Span* _prev=nullptr;size_t _useCount=0;//被使用的个数void* _freeList=nullptr;//切好的小块内存的自由链表
};

给默认值的意义是可以偷懒不写构造函数。

---

SpanList

之前提到central cache需要桶锁，总不能208个一个一个都显示的写吧。而且锁作为公有成员变量没有什么安全问题。所以直接在哈希桶下挂的SpanList中封装一把锁，这样每个桶就都有一把锁。

//每个桶下面挂着的就是SpanList
class SpanList
{ 
public:std::mutex _mtx;//桶锁SpanList(){_head = new Span;_head->_prev = _head;_head->_next = _head;}void Insert(Span* pos, Span* newSpan){assert(pos);assert(newSpan);Span* prev = pos->_prev;prev->_next = newSpan;newSpan->_prev =prev;newSpan->_next = pos;pos->_prev = newSpan;}void Erase(Span* pos){assert(pos);assert(pos != _head);Span* prev = pos->_prev;Span* next = pos->_next;prev->_next = next;next->_prev = prev;//注意不需要free 因为我用完要放到自由链表里面 供以后使用//从当前剔除就行}
private:Span* _head=nullptr;};

3.Central Cache的实现

 声明

class CentralCache
{
public:static CentralCache* GetInstance(){return &_sInst;}//从中心缓冲中获取多少数量的对象给thread cachesize_t FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size);// 获取一个非空的spanSpan* GetOneSpan(SpanList& list, size_t byte_size);
private:CentralCache(){}CentralCache(const CentralCache& abc) = delete;
private:static CentralCache _sInst;//记得类外初始化SpanList _spanLists[NFREELISTS]; 
};

 部分实现

#include "CentralCache.h"
CentralCache CentralCache::_sInst;Span* CentralCache::GetOneSpan(SpanList& list, size_t size)
{// 需要在自身判断 也需要在page cache中判断 所以以后在写// 这里只是为了通过编译return nullptr;
}

3.1 ThreadCache::FetchFromCentralCache()

记得在thread cache中我们没有实现的这个函数吗，现在我们既然有了CentralCache类型了，现在来构思一下这个函数。

void* ThreadCache::FetchFromCentralCache(size_t index, size_t size)

首先这个函数是需要返回一个地址的，就是我们将分割好的内存的起始位置当做返回值返回。其次我们通过上面得知CentralCache一次可不仅仅分给我们一个内存块，而是多个。

因此

• 我们需要知道当空间充足时，分配几个。空间不充足时，分配几个。
• 第二我们也需要知道这群内存块的最后一个的起始位置，方便我们把这一串挂到threadcache下的哈希桶结构内。
• 如何将一串连续的空间都挂到桶上呢？难道每次都一个个的放吗？

3.1.1 SizeClass::NumMoveSize()

该用于确定分配数量的上限，即不可能分配出比该函数返回值更大的数量了。

class SizeClass
{
public://... 之前写过的略static size_t NumMoveSize(size_t size){assert(size > 0);int num = MAX_BYTES / size;if (num < 2) num = 2;//如果内存块比较大 最少一次拿走两个if (num > 512) num = 512;//如果内存块很小 最多一次拿走512个return num;}
};

假设size=8，难道就真的按照返回值512，一次性给一个thread cache分配出512个内存块吗？首先如果是这样，那么只会允许两次向central cache申请，其次给的空间过多，都在某一线程的自由链表挂着，其他线程想要用的时候拿不到，会使central cache的调度次数大大增加。

 

所以我们还需要一个缓慢增加申请数量的代码。

如果一个桶经常向central cache申请空间，说明该映射下的内存块需求量大，我们就依次进行一个缓慢增加的策略，随着他申请次数的增多，一次分配的内存块数量也增多。那要如何保存他申请的次数呢？

直接封装在thread cache桶(FreeList)里面

class FreeList
{
public:void Push(void* obj){}void* Pop(){}bool Empty(){}size_t& Maxsize(){return _maxSize;}
private:void* _freeList=nullptr;size_t _maxSize = 1;
};

3.1.2  FreeList::PushRange()

static void*& NextObj(void* obj)
{return *(void**)obj;
}
class FreeList
{
public:void PushRange(void* start, void* end){NextObj(end) = _freeList;_freeList = start;}size_t& Maxsize(){return _maxSize;}
private:void* _freeList=nullptr;size_t _maxSize = 1;
};

3.1.3 函数实现

这里假设我们已经实现好了一个函数FetchRangeObj()，它可以直接返回申请空间的真实数量。

void* ThreadCache::FetchFromCentralCache(size_t index, size_t size)
{size_t batchNum = std::min(SizeClass::NumMoveSize(size),_freeLists[index].Maxsize());if (_freeLists[index].Maxsize() == batchNum) _freeLists[index].Maxsize() += 1;void* start = nullptr;void* end = nullptr;//batchNum只是理想的 申请的对象 但事实是可能只能申请一个 而申请不到期望值size_t actualNum = CentralCache::GetInstance()->FetchRangeObj(start, end, batchNum, size);if(actualNum == 1){assert(start == end);return start;}else{_freeLists[index].PushRange(NextObj(start), end);return start;}return nullptr;
} 

 3.2 CentralCache::FetchRangeObj()

经过上面的分析，ThreadCache::FetchFromCentralCache()中需要用到CentralCache::FetchRangeObj()，以拿到start end指针，以及确切的可以分配给thread cache的内存块数量。

首先，需要通过size找到哈希桶的下标，然后找一个非空的Span，用来申请空间。

其次，如果span->_freeList不为空就说明至少有一个内存块，所以actualNum初始值为1，所以end也只需向后走batchNum-1次，将[start,end]这部分拿走，使_freeList指向end后面的那个节点，即使那个节点是空也符合逻辑。

然后，依据batchNum获取内存块数，可能会不够，但至少有一个可以被拿走使用解决问题，所以处理的逻辑是，如果不能获取期望的内存块数，那就有几个拿几个。当end的后面为空时停止。

size_t CentralCache::FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size)
{//首先依据申请的对齐字节数判断在哪一个哈希桶内size_t index = SizeClass::Index(size);//上锁_spanLists[index]._mtx.lock();//要给别人分配内存 首先从自己这里找一个非空的Span 自己没有就从page cache找一个Span* span = GetOneSpan(_spanLists[index],size);assert(span);assert(span->_freeList);start = span->_freeList;end = start;int i = 0;size_t actualNum = 1;while (i < batchNum - 1 && NextObj(end) != nullptr){end = NextObj(end);i++;}span->_freeList = NextObj(end);NextObj(end) = nullptr;_spanLists[index]._mtx.unlock();return actualNum;
}

4.thread cache向central cache申请空间的流程图

流程图