当前项目是实现一个高并发的内存池，他的原型是google的一个开源项目tcmalloc，tcmalloc全称Thread-Caching Malloc，即线程缓存的malloc，实现了高效的多线程内存管理，用于替代系统的内存 分配相关的函数（malloc、free）。

        另一方面tcmalloc是全球大厂google开源的，可以认为当时顶尖的C++高手写出来的，他的知名度也是非常高的，不少公司都在用它，Go语言直接用它做了自己内存分配器。

        本文是把tcmalloc最核心的框架简化后拿出来，模拟实现出一个自己的高并发内存池，目的是学习tcamlloc的精华。

数据结构（链表、哈希桶）、操作系统内存管理、单例模式、多线程、互斥锁等。

C/C++中我们要动态申请内存都是通过malloc去申请内存，但是我们要知道，实际我们不是直接去堆获取内存的， 而malloc就是一个内存池。malloc() 相当于向操作系统“批发”了一块较大的内存空间，然后“零售”给程 序用。当全部“售完”或程序有大量的内存需求时，再根据实际需求向操作系统“进货”。

        现代很多的开发环境都是多核多线程，在申请内存的场景下，必然存在激烈的锁竞争问题。malloc本身其实已经很优秀，那么我们项目的原型tcmalloc就是在多线程高并发的场景下更胜一筹，所以这次我们实现的内存池需要考虑以下几方面的问题。

concurrent memory pool主要由3个部分构成：

1. thread cache：线程缓存是每个线程独有的，用于小于256KB的内存的分配，线程从这里申请内存不需要加锁，每个线程独享一个thread cache，这也就是这个并发线程池高效的地方。

2. central cache：中心缓存是所有线程所共享，thread cache是按需从central cache中获取的对象。central cache合适的时机回收thread cache中的对象，避免一个线程占用了太多的内存，而其他线程的内存吃紧，达到内存分配在多个线程中更均衡的按需调度的目的。central cache是存在竞争的，所以从这里取内存对象是需要加锁，首先这里用的是桶锁，其次只有thread cache的对象没有内存时才会找central cache，所以这里竞争不会很激烈。

3. page cache：页缓存是在central cache缓存上面的一层缓存，存储的内存是以页为单位存储及分配的，central cache没有内存对象时，从page cache分配出一定数量的page，并切割成定长大小的小块内存，分配给central cache。当一个span的几个跨度页的对象都回收以后，page cache 会回收central cache满足条件的span对象，并且合并相邻的页，组成更大的页，缓解内存碎片的问题。

申请内存：

1. 当内存申请 size<=256KB ，先获取到线程本地存储的 thread cache 对象，计算 size 映射的哈希桶自由链表下标i 。

2. 如果自由链表 _freeLists[i] 中有对象，则直接 Pop 一个内存对象返回。

3. 如果 _freeLists[i] 中没有对象时，则批量从 central cache 中获取一定数量的对象，插入到自由链表并返回一个对象。

释放内存：

central cache 也是一个哈希桶结构，他的哈希桶的映射关系跟 thread cache 是一样的。不同的是他的每个哈希桶位置挂是SpanList 链表结构，不过每个映射桶下面的 span 中的大内存块被按映射关系切成了一个个小内存块对象挂在span 的自由链表中。

申请内存：

1. 当thread cache中没有内存时，就会批量向central cache申请一些内存对象，这里的批量获取对象的数量使用了类似网络tcp协议拥塞控制的慢开始算法；central cache也有一个哈希映射的spanlist，spanlist中挂着span，从span中取出对象给thread cache，这个过程是需要加锁的，不过这里使用的是一个桶锁，尽可能提高效率。

2. central cache映射的spanlist中所有span的都没有内存以后，则需要向page cache申请一个新的span对象，拿到span以后将span管理的内存按大小切好作为自由链表链接到一起。然后从span中取对象给thread cache。

3. central cache的中挂的span中use_count记录分配了多少个对象出去，分配一个对象给thread cache，就++use_count。

释放内存：

1. 当thread_cache过长或者线程销毁，则会将内存释放回central cache中的，释放回来时 --use_count。当use_count减到0时则表示所有对象都回到了span，则将span释放回page cache，page cache中会对前后相邻的空闲页进行合并。

---

<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类定义

和Thread Cache不同的是，每个线程中都会有一个Thread Cache对象以防止各个线程互相抢占内存的问题。而Central Cache为所有线程所共享且全局只有一个Central Cache对象，故我们应该将其实现为单例模式(其中懒汉模式涉及线程安全问题，故我们将其实现为懒汉模式)。其主要作用是向下为Thread Cache提供内存，向上向Page Cache申请内存。回收内存也是同理。故我们需要定义一下几个接口，从Page Cache获取非空spanNode、为Thread Cache提供内存、将内存释放回Page Cahce三个函数。如下定义：

//单例模式（懒汉）
class CentralCache {
private:SpanList _SpanLists[NFREELIST];
private:CentralCache() {}CentralCache(const CentralCache&) = delete;static CentralCache _sInst; //类外初始化
public:static CentralCache* GetInStance() {return &_sInst;}// 获取一个非空的spanNodeSpanNode* GetOneSpan(SpanList& list, size_t size);// 从中心缓存获取一定数量的对象给thread cachesize_t FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size);// 将一定数量的对象释放到spanvoid 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. 当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。

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

3. 需要注意的是central cache和page cache 的核心结构都是spanlist的哈希桶，但是他们是有本质区别的，central cache中哈希桶，是按跟thread cache一样的大小对齐关系映射的，他的spanlist中挂的span中的内存都被按映射关系切好链接成小块内存的自由链表。而page cache 中的spanlist则是按下标桶号映射的，也就是说第i号桶中挂的span都是i页内存。

释放内存：

1. 如果 central cache 释放回一个 span ， 则依次寻找 span 的前后 page id 的没有在使用的空闲 span ， 看是否可以合并，如果合并继续向前寻找。这样就可以将切小的内存合并收缩成大的 span ，减少 内存碎片 。

---

<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页SpanNodeSpanNode* NewSpan(size_t n);//获取从对象到SpanNode的映射SpanNode* MapObjectToSpanNode(void* obj);//释放空闲SpanNode回到PageCache，并合并相邻的SpanNodevoid ReleaseSpanNodeToPageCache(SpanNode* SpanNode);
};

代码实现：

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

windows和Linux下如何直接向堆申请页为单位的大块内存：// 去堆上按页申请空间
inline static void* SystemAlloc(size_t kpage)
{
#ifdef _WIN32void* ptr = VirtualAlloc(0, kpage << 13, MEM_COMMIT | MEM_RESERVE, PAGE_READWRITE);
#else// linux下brk mmap等
#endifif (ptr == nullptr)throw std::bad_alloc();return ptr;
}inline static void SystemFree(void* ptr)
{
#ifdef _WIN32VirtualFree(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位相当于除8kspanNode->_n = n;_idSpanNodeMap[spanNode->_pageId] = spanNode; //建立映射方便释放内存return spanNode;}//检查对应桶中是否有sapnNodeif (!_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、性能分析

---

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

运行结果，由下图可见我们当前实现的内存池效率和malloc比起来差了许多。

template<class T>
class FiedMemoryPool {
private:char* _memory = nullptr;   //指向大块内存的指针void* _freeList = nullptr;//还内存链接自由链表头指针size_t _residueBytes = 0;  //剩余字节数
public:T* New() {T* obj = nullptr;//优先把还回来的对象重复利用if (_freeList) {void* next = *((void**)_freeList);obj = (T*)_freeList;_freeList = next;}else {//剩余内存不够一个对象大小时重新开辟内存if (_residueBytes < sizeof(T)) {_residueBytes = 128 * 1024;_memory = (char*)SystemAlloc(_residueBytes >> PAGE_SHIFT);if (_memory == nullptr) {throw std::bad_alloc();}}//给目标分配内存obj = (T*)_memory;size_t objSize = sizeof(T) < sizeof(void*) ? sizeof(void*) : sizeof(T); //至少开辟一个指针大小_memory += objSize;_residueBytes -= objSize;}//用定位new显示调用其构造函数new (obj)T;return obj;}void Delete(T* obj) {//显示调用其析构函数obj->~T();//头插进自由链表*(void**)obj = _freeList;_freeList = obj;}
};