高并发内存池

#高并发内存池| 来源: 网络整理| 查看: 265

一、项目背景介绍

本项目的原型是Google开源的tcmalloc项目，tcmalloc全称为Thread Cache Malloc，意为线程缓存的malloc，用于替换系统的malloc和free。其替换原因在于malloc在对于多线程申请内存时的效率较低，使用该方法能够提高大概20%的效率。

二、项目整体架构

整个项目是一个三级缓存的结构，第一层为ThreadCache，第二层为CentralCache，第三层为PageCache，其代码实现也是从第一层，逐步走到第三层，请大家耐心观看。

三、理解内存池是什么

内存池是指程序预先向操作系统申请一块足够大的内存，此后，当程序中需要申请内存的时候，不是直接向操作系统申请，而是直接从内存池中获取；同理，当释放内存的时候，并不是真正将内存返回给操作系统，而是将内存返回给内存池。当程序退出时（或某个特定时间），内存池才将之前申请的内存真正释放。

四、开胃菜——定长内存池

定长内存池顾名思义，就是一个长度固定的内存块，其只能申请和释放固定大小的内存块。因此我们可以将它的性能发挥到机制。设计定长内存池的目的也是为后面打好基础。

4.1内存申请的管理

首先设计不考虑内存释放时的场景！

上述申请的过程,可能有以下疑虑：

为什么每次申请的大小都是T呢？这就是定长内存池的特点,每次申请都是固定大小的,这样方便管理！

只能一块一块的申请吗？是的，目前只能一块一块的申请，后续可以扩展

被切分出去的内存块是连续的吗？答案是：一定是连续的！

上述过程中，并没有考虑内存释放的问题！下面先让我们看看内存释放！最后再做总结。

template class ObjectPool { public: //构造函数很容易理解，刚开始没有内存块，并且剩余字节数为0 ObjectPool():_memory(nullptr),_freelist(nullptr),_remainBytes(0){} T* New() { T* obj = nullptr;//obj用于申请的内存块的起始地址 //每次只能申请一块大小为T的内存块 //如果T > 剩余内存的数量，就需要重新申请一块定长的内存块(很容易理解叭) if (sizeof(T) > _restnum) { _memory = (char*)malloc(128 * 1024); _remainBytes = 128 * 1024; } //计算objSize与一个指针的大小，这么做是为了后面释放内存的时候方便！可以暂时记住。 int objSize = sizeof(T) > sizeof(void*) ? sizeof(T) : sizeof(void*); //_memory是剩余内存块的头，将_memory给obj,_memory向后移objSize个单位 //因为_memory是char*类型，所以申请多少内存，就移多少个单位。 //如果还不理解的话，就看一下画的图。 obj = (T*)_memory; _memory += objSize; //此时剩余字节数减掉刚刚申请的内存块的数量。 _remainBytes -= objSize; new(obj)T;//定位new，显示调用T的构造函数的初始化 return obj; } private: char* _memory; //大块内存的起始地址 void* _freelist;//暂时不考虑它！ size_t _remainBytes; //剩余字节数 }; 4.2内存释放的管理

被释放掉的内存，我选择通过自由链表来进行管理。那么什么是自由链表呢？

自由链表：看起来就是一个链表，但其并没有显性设置链表的结点next，而是将next存在了当前内存块的前4位或8位地址。这样说比较抽象，不如来看看图解。

后面被释放的内存依旧按照这个流程！

void Delete(T* obj) { obj->~T();//显示调用obj的析构函数 *(void**)obj = _freelist; //强转为void**，再解引用，将_freelist的值放入obj的头4个或8个字节。 //为什么是4或者8字节呢？因为对应32位和64位系统，指针大小可能不同。 //再更新_freelist _freelist = obj; } 4.3 完善内存申请的管理

现在考虑到有释放的内存，那么将释放的内存进行回收，是可以进行再利用的！

因此现在内存申请的流程，应该是先看_freelist里面是否有回收的内存，如果有的话，先申请_freelist里面的内存。否则再到内存池中进行申请！

完整代码:

template class ObjectPool { public: ObjectPool() :_memory(nullptr), _freelist(nullptr), _remainBytes(0) {} T* New() { T* obj = nullptr; if (_freelist) { void* next = *(void**)_freelist;//这里是为了取到当前内存块保存的下一块内存的地址。 obj = _freelist; _freelist = next; } else { if (sizeof(T) > _remainBytes) { _memory = (char*)malloc(128 * 1024); _remainBytes = 128 * 1024; } } //计算objSize与一个指针的大小，这么做是为了后面释放内存的时候方便！可以暂时记住。 int objSize = sizeof(T) > sizeof(void*) ? sizeof(T) : sizeof(void*); obj = (T*)_memory; _memory += objSize; _remainBytes -= objSize; new(obj)T; return obj } void Delete(T* obj) { obj->~T(); *(void**)obj = _freelist; _freelist = obj; } private: char* _memory; void* _freelist; size_t _remainBytes; }; 五、ThreadCache的设计 5.1 ThreadCache整体框架

ThreadCache的结构是一个哈希桶结构，每一个哈希桶对应不同的内存大小。需要申请多少内存，就到对应的桶里面去取一块内存就可以了。并且每个线程都有自己独立的ThreadCache，不需要加锁处理，这也是能够提高效率的原因之一。

申请内存:

1. 当内存申请size_next = nullptr; _head->_prev = nullptr; } private: Span* _head = nullptr; public: std::mutex _mtx; 每一个Spanlist 都具有一个锁,因为CentralCache是只有一个的。会存在多个线程共同竞争同一个桶的情况，因此需要加锁 };

接口的实现：

补充ThreadCache中的:FetchFromCentralCache 从CentralCache中拿内存 void* ThreadCache::FetchFromCentralCache(size_t index, size_t size) { size是申请的内存大小,index是桶的编号 size_t batchNum = min(_freelist[index].MaxSize(), SizeClass::NumMoveSize(size)); if (batchNum == _freelist[index].MaxSize()) { _freelist[index].Maxsize++; } void* start = nullptr; void* end = nullptr; 实际申请到的内存块数量 int actualNum = CentralCache::GetInstance()->FetchRangeObj(start, end, batchNum, size); 再将申请到的内存块插入到对应的桶中为什么要判断这一步呢？因为如果 actualNum等于1的话,就不需要再插入到桶中了！ if (actualNum == 1) { assert(start == end); } else { //插入桶中的时候，一定是先将start给排除了，再插入的。 _freelist[index].PushRange(NextObj(start), end, actualNum - 1); } return start; } 在CentralCache中截取部分内存 size_t CentralCache::FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size) { int index = SizeClass::Index(size); 找到对应的桶号 _spanlist[index]._mtx.lock(); 因为要对桶进行操作,所以需要加锁要找到一个不为空的span Span* span = CentralCache::GetInstance()->GetOneSpan(_spanlist[index],size); assert(span); assert(span->_freelist); 开始截取 start = span->_freelist; void* cur = start; size_t actualNum = 0; while (cur && batchNum != 0) { end = cur; cur = NextObj(cur); actualNum++; batchNum--; } span->_freelist = cur; span->_useCount += actualNum; span中已经使用的内存块 _spanlist[index]._mtx.unlock();对桶的操作已经结束了，解锁！ return actualNum; } 获取一个非空的Span Span* CentralCache::GetOneSpan(Spanlist& spanlist, size_t size) { size_t index = SizeClass::Index(size); Span* start = spanlist.Begin(); while (start != spanlist._head) { if (start->_freelist != nullptr) return start; start = start->_next; } 此时这里需要解锁，因为现在并不对spanlist进行操作了，虽然现在spanlist是空的，但是可能会有内存的释放，回收问题。如果不解锁，会影响内存回收。 spanlist._mtx.unlock(); 走到这里说明Spanlist是空的需要向PageCache申请内存。 int k = SizeClass::NumMovePage(size); 向PageCache申请内存是需要加锁的。 PageCache::GetInstance()->_mtx.lock(); Span* span = PageCache::GetInstance()->NewSpan(k); PageCache::GetInstance()->_mtx.unlock(); 标记这个span正在使用 span->_IsUse = true; span->_objsize = size; 此时是切分了一个很大的span 不是CentralCache中所需要的span 我们是需要将这个span 进行切割的，放入到自由链表中。通过页号计算出 span的起始位置 char* start = (char*)(span->_pageId _n _freelist = start; start += size; 再将剩余部分进行尾插 void* tail = span->_freelist; while (start != end) { NextObj(tail) = start; tail = NextObj(tail); start += size; } NextObj(tail) = nullptr; spanlist._mtx.lock(); spanlist.PushFront(span); return span; }

图解：如何在CentralCache中申请内存:

（1）找到对应的桶并找到一个非空的Span

（2）如果在PageCache中申请一个新的span

新申请的newspan是需要处理后，再放入Central Cache的

（3）对该非空Span进行分割:

①：batchNum小于等于Span中内存块的数量

②：batchNum大于Span中内存块的数量

（4）将申请多的内存块挂到ThreadCache中

如何理解申请多的内存块？因为每一次向CentralCache申请内存的时候，不是一块一块的申请，而是一批一批的申请内存块，这样可以减少对CentralCache的访问。因为每次对CentralCache进行访问的时候，都需要加锁和解锁的操作。因此需要一次性申请多块内存。

七、PageCache的设计 7.1 PageCache的整体框架

PageCache是一个以页为单位的spanlist。

申请内存：

1.当central cache向page cache申请内存时，page cache先检查对应位置有没有span，如果没有则向更大页寻找一个span，如果找到则分裂成两个。比如：申请的是4页page，4页page后面没有挂span，则向后面寻找更大的span，假设在10页page位置找到一个span，则将10页pagespan分裂为一个4页page span和一个6页page span。

2.如果找到_spanList[128]都没有合适的span，则向系统使用mmap、brk或者是VirtualAlloc等方式申请128页page span挂在自由链表中，再重复1中的过程。

7.2 PageCache代码框架现只考虑申请内存的流程将PageCache设计为单例模式 class PageCache { public: static PageCache* GetInstance() { return &_sInit; } Span* NewSpan(size_t k); private: PageCache(){} PageCache(const PageCache&) = delete; private: Spanlist _pagelist[NPAGES]；用于管理每一页的自由链表。NPAGES =129；因为没有第0页,但数组下标是从0开始的。因此一共设置了129页，不用第0页。 public: static PageCache _sInit; std::mutex _mtx; PageCache只有一把整体的锁，是因为回收内存的时候，每个页之间会相互影响，因此得整体加锁。为什么不每个page都加锁呢？这样的话会降低整体的效率，访问每一页的时候都需要加锁解锁，性能不好。 }; PageCache PageCache::_sInit;

接口实现：

class Spanlist { void Erase(Span* pos) { assert(pos); assert(pos!=_head); Span* next = pos->_next; Span* prev = pos->_prev; next->_prev = prev; prev->_next = next; } Span* PopFront() 头删，这里的删只是把当前Span从Spanlist中删除，并不是将它delete! { Span* front = _head->_next; Erase(front); return front; } void PushFront(Span* span) 头插 { Span* next = _head->_next; _head->_next = span; span->_prev = _head; next->_prev = span; } } Span* PageCache::NewSpan(size_t k) { 如果一次性要的内存是大于128页的，那么直接向系统申请内存。 if (k > NPAGES - 1) { Span* span = new Span; void* ptr = SystemAlloc(k); span->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT; span->_n = k; return span; } if (_pagelist[k].Empty())如果这一页为空,则往后面找不是空白页的地方 { for (int i = k + 1; i < NPAGES; i++) { if (!_pagelist[i].Empty()) { Span* nspan = _pagelist->PopFront();将这一部分内容弹出来 Span* kspan = new Span; 设置页号，可能有的同学会好奇，这里的页号在哪呢？仔细慢慢看 kspan->_pageId = nspan->_pageId; 页数为k kspan->_n = k; nspan->_pageId += k; nspan->_n -= k; _pagelist[i - k].PushFront(nspan); return kspan; } } 到这里说明_pagelist[128]都为空，这时需要向系统申请空间 Span* bigSpan = new Span; void* ptr = SystemAlloc(NPAGES - 1);自己封装的向系统申请空间 bigSpan->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT;计算页号的公式 bigSpan->_n = NPAGES - 1;页号填充。 _pagelist[bigSpan->_n].PushFront(bigSpan); 申请完毕后，再递归一次，此时的_pagelist[128]不为空白页，可以进行切割了。 return NewSpan(k); } 代码走到这里说明当前页不为空页,可以直接弹回Span* else { return _pagelist[k].PopFront(); } }

图解：向PageCache中申请内存

以上就是申请内存的全部过程啦~还有一些小细节将在回收内存的时候进行补充

八、释放内存 ThreadCache回收内存的理论

释放内存依旧是从ThreadCache开始设计，再到CentralCache、PageCache，最后还回系统。

在ThreadCache中释放内存要满足以下条件：

1. 当释放内存小于256k时将内存释放回thread cache，计算size映射自由链表桶位置i，将对象Push

到_freeLists[i]。

2. 当链表的长度过长，则回收一部分内存对象到central cache。

ThreadCache回收内存代码框架 void Deallocate(void* ptr, size_t size); 释放内存接口，带size有点鸡肋，后面会修改。 void ListTooLong(FreeList& list, size_t size);链表过长的处理

接口实现

void ListTooLong(FreeList& list, size_t size) { void* start = nullptr; void* end = nullptr; 将这一部分内存块从自由链表中弹出来 list.PopRange(start, end, list.MaxSize()); 再还原到CentralCache中。 CentralCache::GetInstance()->ReleaseToSpan(start, size); } 回收内存 void ThreadCache::Deallocate(void* ptr, size_t size) { assert(size > 0); size_t index = SizeClass::Index(size); 将ptr传入到对应的桶中。 _freelist[index].Push(ptr); 如果当前_freelist的数量已经大于了一次批量申请的数量时就要还回CentralCache了 if (_freelist[index].MaxSize() < _freelist[index].Size()) { 还回CentralCache的数量就是一次批量申请的数量传入size是为了找到在CentralCache中的index. ListTooLong(_freelist[index], size); } }

图解：ThreadCache回收内存

CentralCache回收内存的理论

首先是从ThreadCache处回收内存，ThreadCache都会释放一批内存到CentralCache中，而每一块内存都具有相应的span，不能直接将这一批内存挂到同一个span上，因为这样不利于后面还回PageCache。

当一个Span满了，就直接还给PageCache。那如何定义Span满了呢？每一个Span都具有UseCount，用于记录用了多少块内存，那么当UseCount等于0的时候，就意味着这个Span装满了，就可以还回PageCache了。

CentralCache回收内存代码框架 void Freelist::PopRange(void*& start,void*& end,size_t n) { assert(n >= _size); start = _freelist; end = start; for (size_t i = 0; i < n - 1; i++) { end = NextObj(end); } _freelist = NextObj(end); _size -= n; NextObj(end) = nullptr; } 这个容器是用来记录页号与Span之间的对应关系。那这个容器是在什么时候用的呢？答案是：在接口NewSpan的时候。就已经用上了，不过之前实现的时候，为了容易理解，没有加上。现在大家理解这个申请内存的过程了，我们现在可以将这一步给加上了。正好带大家回忆一下NewSpan的过程。 Span* PageCache::NewSpan(size_t k) { 如果一次性要的内存是大于128页的，那么直接向系统申请内存。 if (k > NPAGES - 1) { Span* span = new Span; void* ptr = SystemAlloc(k); span->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT; span->_n = k; _idSpanmap[span->pageId] =span; return span; } if (_pagelist[k].Empty())如果这一页为空,则往后面找不是空白页的地方 { for (int i = k + 1; i < NPAGES; i++) { if (!_pagelist[i].Empty()) { Span* nspan = _pagelist->PopFront();将这一部分内容弹出来 Span* kspan = new Span; 设置页号，可能有的同学会好奇，这里的页号在哪呢？仔细慢慢看 kspan->_pageId = nspan->_pageId; 页数为k 新加入的代码：将nspan的头和尾插入到map中。 _idSpanmap[nspan->_pageId] = nspan; _idSpanmap[nspan->_pageId + nspan->_n - 1] = nspan; kspan->_n = k; nspan->_pageId += k; nspan->_n -= k; 遍历kspan,将kspan都加入到map中。 for (PAGE_ID i = 0; i < kspan->_n; i++) { _idSpanmap[kspan->_pageId + i] = kspan; } _pagelist[i - k].PushFront(nspan); return kspan; } } 到这里说明_pagelist[128]都为空，这时需要向系统申请空间 Span* bigSpan = new Span; void* ptr = SystemAlloc(NPAGES - 1);自己封装的向系统申请空间 bigSpan->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT;计算页号的公式 bigSpan->_n = NPAGES - 1;页号填充。 _pagelist[bigSpan->_n].PushFront(bigSpan); 申请完毕后，再递归一次，此时的_pagelist[128]不为空白页，可以进行切割了。 return NewSpan(k); } 代码走到这里说明当前页不为空页,可以直接弹回Span* else { Span* cur = _pagelist[k].PopFront(); for (int i = 0; i < cur->_n; i++) { _idSpanmap[cur->_pageId + i] = cur; } return cur; } } PageCache::unordered_map _idSpanmap; 通过地址找到对应的span。 Span* PageCache::MapObjectToSpan(void* obj) { 通过地址找到对应的页号 PAGE_ID id = ((PAGE_ID)obj >> PAGE_SHIFT); auto ret =_idSpanmap.find(id); if (ret != _idSpanmap.end()) { return _idSpanmap[id]; } return nullptr; } void CentralCache::ReleaseListToSpans(void* start, size_t size)；

接口实现：

将ThreadCache释放的内存，还到对应的span中 void CentralCache::ReleaseToSpan(void*& start,size_t size) { size_t index = SizeClass::Index(size); 对哪一个桶操作，就需要进行加锁 _spanlist[index]._mtx.lock(); while (start) { void* next = NextObj(start); 通过MaptoSpan寻找start所对应的span. Span* span = PageCache::GetInstance()->MaptoSpan(start); 再挂到对应的span上 NextObj(start) = span->_freelist; span->_freelist = start; span->_useCount--; 当_useCount为0的时候，就需要还回PageCache了 if (span->_useCount == 0) { span->_IsUse = false; 对PageCache进行访问的时候，需要加锁。PageCache是只有一把锁。 PageCache::GetInstance()->_mtx.lock(); PageCache::GetInstance()->ReleasetoPage(span); PageCache::GetInstance()->_mtx.unlock(); } } _spanlist[index]._mtx.unlock(); } void PageCache::ReleasetoPage(Span* span) { if (span->_n >= NPAGES) { void* ptr = (void*)(span->_pageId _pageId - 1; auto prev = (Span*)_idSpanmap.get(Prev_Id); if (prev == nullptr)//没有找到对应的Span,不能合并 { break; } if (prev->_IsUse ==true)//如果当前这个内存块被使用,则不能合并 { break; } if (prev->_n + span->_n > NPAGES) //合并起来一共的页数大于128也不能合并 { break; } span->_pageId = prev->_pageId; span->_n += prev->_n; _pagelist[prev->_n].Erase(prev); delete prev; } //向后合并 while (1) { PAGE_ID Next_Id = span->_pageId + span->_n; auto ret = (Span*)_idSpanmap.get(Next_Id); if (ret == nullptr) { break; } //Span* next = _idSpanmap[Next_Id]; Span* next = (Span*)_idSpanmap.get(Next_Id); if (next->_IsUse ==true) { break; } if (next->_n + span->_n > NPAGES) { break; } span->_n += next->_n; _pagelist[next->_n].Erase(next); delete next; } _pagelist[span->_n].PushFront(span); span->_IsUse = false; _idSpanmap.set(span->_pageId,span); _idSpanmap.set(span->_pageId + span->_n - 1, span); return; }

图解：CentralCache回收内存

PageCache回收内存理论

当CentralCache的Span满的时候，会将这个Span还给PageCache。恰好这个Span的总体大小就是X页的内存，因此只需要找到这个Span对应的页数并向前合并和向后合并！

6. PageCache回收内存代码框架 inline static void SystemFree(void* ptr) { #ifdef WIN32 VirtualFree(ptr, 0, MEM_RELEASE); #endif // WIN32 } void PageCache::ReleasetoPage(Span* span) { 页数大于NPAGES 直接还给系统 if (span->_n >= NPAGES) { void* ptr = (void*)(span->_pageId _pageId - 1; auto ret = _idSpanmap.find(Prev_Id); //没有找到对应的Span,不能合并 if (ret == _idSpanmap.end()) { break; } Span* prev = _idSpanmap[Prev_Id]; //如果当前这个内存块被使用,则不能合并 if (prev->_IsUse) { break; } //合并起来一共的页数大于128也不能合并 if (prev->_n + span->_n > NPAGES) { break; } span->_pageId = prev->_pageId - prev->_n+1; span->_n += prev->_n; _pagelist[prev->_n].Erase(prev); delete prev; } //向后合并 while (1) { PAGE_ID Next_Id = span->_pageId + span->_n; auto ret = _idSpanmap.find(Next_Id); if (ret == _idSpanmap.end()) { break; } Span* next = _idSpanmap[Next_Id]; if (next->_IsUse) { break; } if (next->_n + span->_n > NPAGES) { break; } span->_n += next->_n; _pagelist[next->_n].Erase(next); delete next; } 将最后合并好的span插入到对应的桶中。 _pagelist[span->_n].PushFront(span); 并将它存入map中。 _idSpanmap[span->_pageId] = span; _idSpanmap[span->_pageId + span->_n-1] = span; return; }

九、性能测试 void BenchmarkMalloc(size_t ntimes, size_t nworks, size_t rounds) { std::vector vthread(nworks); std::atomic malloc_costtime = 0; std::atomic free_costtime = 0; for (size_t k = 0; k < nworks; ++k) { vthread[k] = std::thread([&, k]() { std::vector v; v.reserve(ntimes); for (size_t j = 0; j < rounds; ++j) { size_t begin1 = clock(); for (size_t i = 0; i < ntimes; i++) { v.push_back(malloc(16)); //v.push_back(malloc((16 + i) % 8192 + 1)); } size_t end1 = clock(); size_t begin2 = clock(); for (size_t i = 0; i < ntimes; i++) { free(v[i]); } size_t end2 = clock(); v.clear(); malloc_costtime += (end1 - begin1); free_costtime += (end2 - begin2); } }); } for (auto& t : vthread) { t.join(); } printf("%u个线程并发执行%u轮次，每轮次concurrent alloc %u次: 花费：%u ms\n", nworks, rounds, ntimes, malloc_costtime.load()); printf("%u个线程并发执行%u轮次，每轮次concurrent dealloc %u次: 花费：%u ms\n", nworks, rounds, ntimes, free_costtime.load()); printf("%u个线程并发concurrent alloc&dealloc %u次，总计花费：%u ms\n", nworks, nworks * rounds * ntimes, malloc_costtime.load() + free_costtime.load()); } // 单轮次申请释放次数线程数轮次 void BenchmarkConcurrentMalloc(size_t ntimes, size_t nworks, size_t rounds) { std::vector vthread(nworks); std::atomic malloc_costtime = 0; std::atomic free_costtime = 0; for (size_t k = 0; k < nworks; ++k) { vthread[k] = std::thread([&]() { std::vector v; v.reserve(ntimes); for (size_t j = 0; j < rounds; ++j) { size_t begin1 = clock(); for (size_t i = 0; i < ntimes; i++) { v.push_back(ConcurrentAlloc(16)); //v.push_back(ConcurrentAlloc((16 + i) % 8192 + 1)); } size_t end1 = clock(); size_t begin2 = clock(); for (size_t i = 0; i < ntimes; i++) { ConcurrentFree(v[i]); } size_t end2 = clock(); v.clear(); malloc_costtime += (end1 - begin1); free_costtime += (end2 - begin2); } }); } for (auto& t : vthread) { t.join(); } printf("%u个线程并发执行%u轮次，每轮次concurrent alloc %u次: 花费：%u ms\n", nworks, rounds, ntimes, malloc_costtime.load()); printf("%u个线程并发执行%u轮次，每轮次concurrent dealloc %u次: 花费：%u ms\n", nworks, rounds, ntimes, free_costtime.load()); printf("%u个线程并发concurrent alloc&dealloc %u次，总计花费：%u ms\n", nworks, nworks * rounds * ntimes, malloc_costtime.load() + free_costtime.load()); } int main() { size_t n = 100000; std::cout > LEAF_BITS) + 1)

【本文地址】

高并发内存池

高并发内存池

今日新闻

推荐新闻