C++高并发内存池ConcurrenMemoPool

一、介绍高并发内存池

本项目的原型是Google的开源项目tcmalloc，即线程缓存的malloc，相较于系统的内存分配函数malloc，free，本项目能达到高效的多线程内存管理
旨在学习其核心框架，借鉴其实现方式来模拟实现出一个我们自己的高并发内存池，主要是学习其精华。
其实我们在C++中常用的malloc就是一种池化技术，当然了new也是（其本质就是对malloc的封装嘛）
在最后我们会对比我们自己模拟出来的高并发内存池和malloc，我们拭目以待

池化技术

谈起池化技术，对于我们学计算机的人来说，再熟悉不过了，就是程序先向系统申请一定量的资源，交由自己管理，这样不需要频繁的去系统申请资源（系统也是很忙的，需要处理的事情是很多的），所以这就大大的提高了程序运行的效率！！在计算机中，有很多地方都使用了池化技术，比如内存池，线程池，连接池，对象池等等

内存池

内存池是程序先预先从操作系统申请一块足够大的内存，当程序需要一定的内存时，先去内存池里申请，如果不够的情况下再向系统申请，当还回内存时也就直接还给内存池，这样轻易都不会去系统申请内存，一定程度上减轻了系统的压力，从而使系统能够有时间去处理其他操作

解决的问题

我们做事，最重要的就是效率，所以首当其冲一定是效率问题！
1 效率问题
2 内存碎片问题
比如我们向系统申请了几块连续的内存，但是零零散散的还回了一部分，我们又需要申请时，就可能会遇到这些零散的碎片化的内存加起来是足够的，但是由于他们并不连续，就没办法继续完成申请，这样的空间就是被浪费掉了，所以这个问题也待解决。这种情况就是外碎片内存，还有一种叫做内碎片问题，内碎片是指，我们要申请一段段小内存时，内存会有一定的规则来对齐，方便使用，所以在对齐时会使一定量的内存没有被使用，这就是内碎片，学过C/C++的都应该知道内存对齐原则，这种对齐原则虽然浪费了一定的内存，但是解决的两个问题，平台移植的问题和效率的问题，实质上也是一种空间换时间的方法，所以保留了这种内存对齐规则，优化了外碎片问题

3 多线程环境下，锁竞争的问题

要求

C/C++、数据结构、操作系统的内存管理、单例模式（饿汉模式，懒汉模式）、多线程、锁，基本上要熟练掌握

主体结构框架

主要分为三层 一 thread cache
二 central cache
三 page cache

第一层：thread cache 线程缓存是每个线程独有的，用于小于256KB内存的分配（这里的内存大小可以自行根据需求进行更改），线程从这里申请内存不需要加锁，每个线程独享一个cache，这样也减少了锁的竞争
第二层：central cache “中心缓存” ，它是所有线程所共享的，thread cache需要从central cache进行申请，合适的时候也会对thread cache中的内存进行回收，避免一个线程占用了过多的内存而其他线程内存不够用的情况，这里需要加锁，但是会用到一种比较巧妙地加锁方式
第三层：page cache 类似于第二层，具体实现方式会在下面细讲

核心流程框架

申请内存过程（看不懂没关系，先了解一个大概，后面会细细讲解）

释放内存过程

二、定长内存池

在C/C++里面，申请内存使用的是malloc，但是我们也知道，malloc是一个通用的接口，也就意味着它是一个性能不会很高的接口，设计一个高并发内存池相对比较复杂，我们这里先由浅入深，从一个简单的定长内存池入手
为了方便我们后续的使用，我们把他设计成一个类模板（ObjectPool）
有两个功能，申请+构造（New）释放+析构（Delete）类似于C++
分为两部分，用于内存申请的一个定长内存，以及申请的内存换回来之后要去的地方（自由链表）
第一部分是一个定长内存，这个很简单，当调用New时，如果自由自由链表里有对象的话，直接取走一个，如果没有的话，调用相应的系统函数去申请一块内存
第二部分 释放，直接将不用的内存挂到自由链表上即可（至于怎么挂，头插，尾插都可以，当然头插的效率要比尾插高，因为我们采用的单链表，尾插遍历时，效率不高）
自由链表的结构：我前一块小内存的一个指针大小的空间存一个指针，指针指向后一个小块内存
这里可能有人会问，不同平台下，指针所占空间大小也不相同啊？再分类讨论多麻烦啊，这里我找到一个巧妙的方法，就是利用（void*）指针，不同平台之间指针大小是不同，但是我直接利用二级指针来开大小呢？*（void**）

系统的内存申请和释放

inline static void* SystemAlloc(size_t kpage)
{
#ifdef _WIN32void* ptr = VirtualAlloc(0, kpage << 13, MEM_COMMIT | MEM_RESERVE, PAGE_READWRITE);
#else
#endifif(ptr == nullptr)throw std::bad_alloc();return ptr;
}inline static void SystemFree(void* ptr)
{
#ifdef _WIN32VirtualFree(ptr, 0, MEM_RELEASE);
#else// sbrk unmmap等
#endif
}

申请内存

T* New()
{T* obj = nullptr;//如果自由链表里有对象，直接取走一个if (_freeList){void* next = *((void**)_freeList);obj = (T*)_freeList;_freeList = next;}else{if (_remainBytes < sizeof(T)){_remainBytes = 128 * 1024;_memory = (char*)SystemAlloc(_remainBytes >> 13);if (_memory == nullptr){throw std::bad_alloc();}}obj = (T*)_memory;size_t objSize = sizeof(T) < sizeof(void*) ? sizeof(void*) : sizeof(T);_memory += objSize;_remainBytes -= objSize;}new(obj)T;return obj;
}

释放内存，挂到自由链表

void Delete(T* obj)
{obj->~T();*(void**)obj = _freeList;_freeList = obj;
}

成员变量

char* _memory = nullptr; //指向大块内存
size_t _remainBytes = 0; //大块内存在被切分的过程中剩余字节数
void* _freeList = nullptr; //还回的自由链表头指针

三、thread cache

thread cache 是哈希桶的结构，每个桶是一个按桶位置映射大小的内存块对象的自由链表，每个线程都会有一个thread cache对象，每个线程需要获取和还回对象都是在自己的thread cache中进行操作，所以这里不需要上锁

这里我设计了四个接口
1 Allocate 申请内存
2 Deallocate 释放内存
3 FetchFromCentralCache 从 central cache中申请内存（当thread cache中不满足我的要求时）
4 ListTooLong 当达到一定的要求时（自由链表过长时），可以适当的还回一些内存给central cache

申请内存

当内存申请的大小size<=256KB时，先获取thread cache中的内存，计算size所映射的哈希桶，如果自由链表中能找到一块内存供我使用，直接删除（Pop）自由链表中的一块内存给我即可，如果没有，那么就向central cache中获取一定数量的内存，插入到自由链表中，并返回一个对象

释放内存

当释放内存小于256KB时，将内存释放回thread cache，计算相应的哈希桶，插入即可，但大于256KB时，回收一部分还给central cache

class ThreadCache
{
public://申请void* Allocate(size_t size);//释放void Deallocate(void* ptr, size_t size);//从central cache中获取对象void* FetchFromCentralCache(size_t index, size_t size);//释放对象时，链表过长的话，回收内存还回给central cachevoid ListTooLong(FreeList& list, size_t size);private:FreeList _freeLists[BarrelNum];
};

这里我们要用到TLS（线程局部存储），也就是一种允许每个线程拥有变量独立副本的机制

//TLS 线程局部存储 静态
static _declspec(thread) ThreadCache* pTLSThreadCache = nullptr;

对齐规则
整体控制在最多10%左右的内碎片浪费
[1,128] 8byte对齐 freelist[0,16)
[128+1,1024] 16byte对齐 freelist[16,72)
[1024+1,81024] 128byte对齐 freelist[72,128)
[81024+1,641024] 1024byte对齐 freelist[128,184)
[641024+1,2561024] 81024byte对齐 freelist[184,208)
总的来说，这种对齐规则，内碎片浪费不会超过百分之十，我们这里需要209个桶
这里在计算时，用到了一些位运算，这里主要是借鉴一些巧妙的方法，这种方法能想到的人还是少数，重要的不是方法怎么来的，重要的是我们要学着去使用它！随便带入一些数据我们就可以理解这种方法

static inline size_t _RoundUp(size_t bytes, size_t alignNum)
{return ((bytes + alignNum - 1) & ~(alignNum - 1));
}
static inline size_t RoundUp(size_t size)
{if (size <= 128){return _RoundUp(size, 8);}else if (size <= 1024){return _RoundUp(size, 16);}else if (size <= (long long)8 * 1024){return _RoundUp(size, 128);}else if (size <= (long long)64 * 1024){return _RoundUp(size, 1024);}else if (size <= (long long)256 * 1024){return _RoundUp(size, (long long)8 * 1024);}else{assert(false);return -1;}
}

static const size_t BarrelNum = 208; //桶数

我们也可以把上面定长内存池中的自由链表中的指针进行一下封装

static void*& NextObj(void* obj)
{return *(void**)obj;//因为在不同平台上指针所占空间不同，void**解引用后是一个指针所占的大小，就不用再考虑他们的区别了
}

自由链表也需要进行统一管理

class FreeList
{
public://头插void Push(void* obj){assert(obj);NextObj(obj) = _freeList;_freeList = obj;++_size;}void PushRange(void* start, void* end,size_t n){NextObj(end) = _freeList;_freeList = start;_size += n;}//头删void* Pop(){assert(_freeList);void* obj = _freeList;_freeList = NextObj(obj);--_size;return obj;}void 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);NextObj(end) = nullptr;_size -= n;}bool Empty(){return _freeList == nullptr;}size_t& MaxSize(){return _maxSize;}size_t Size(){return _size;}
private:void* _freeList = nullptr;size_t _maxSize = 1;size_t _size = 0;
};

我们也需要计算其具体在哪个桶中

static inline size_t _Index(size_t bytes, size_t align_shift)
{return ((bytes + ((long long)1 << align_shift) - 1) >> align_shift) - 1;
}
static inline size_t Index(size_t bytes)
{assert(bytes < MAX_BYTES);static int group_array[4] = { 16,56,56,56 };if (bytes <= 128){return _Index(bytes, 3);//2^3=8}else if (bytes <= 1024) {return _Index(bytes - 128, 4) + group_array[0];}else if (bytes <= (long long)8 * 1024) {return _Index(bytes - 1024, 7) + group_array[1] + group_array[0];}else if (bytes <= (long long)64 * 1024){return _Index(bytes - (long long)8 * 1024, 10) + group_array[2] + group_array[1] + group_array[0];}else if (bytes <= (long long)256 * 1024){return _Index(bytes - (long long)64 * 1024, 13) + group_array[3] + group_array[2] + group_array[1] + group_array[0];}else {return _RoundUp(bytes, 1 << PAGE_SHIFT);}return -1;
}

申请内存：先计算好对齐之后的内存大小，计算其所在的桶，如果这个桶中有内存可以用，直接将该桶中的这一块内存Pop，如果没有的话，就去上一层中（central cache）中取

void* ThreadCache::Allocate(size_t size)
{assert(size <= MAX_BYTES);size_t alignSize = SizeClass::RoundUp(size);//对象对齐size_t index = SizeClass::Index(size);//桶个数if (!_freeLists[index].Empty()){return _freeLists[index].Pop();}else{return FetchFromCentralCache(index, alignSize);}
}

释放内存：计算其应该放回哪个桶，Push进去即可，顺便检查一下，当链表长度大于一次批量申请的内存时就开始还一段list给central cache

void ThreadCache::Deallocate(void* ptr, size_t size)
{assert(ptr);assert(size <= MAX_BYTES);size_t index = SizeClass::Index(size);_freeLists[index].Push(ptr);//当链表长度大于一次批量申请的内存时就开始还一段list给central cacheif (_freeLists[index].Size() >= _freeLists[index].MaxSize()){ListTooLong(_freeLists[index], size);}
}

FetchFromCentralCache 从 central cache中申请内存（当thread cache中不满足我的要求时），这里用一个慢开始调节算法，一开始不会一下就取很多，防止浪费，后面会慢慢增加，你取的次数越多就会随之增加，但也不会超过上限值，如果central cache中有，拿给我thread cache用，如果没有，central cache 再去向page cache 取

void* ThreadCache::FetchFromCentralCache(size_t index, size_t size)
{//慢开始反馈调节//一开始不会一下就取很多，防止浪费，后面会慢慢增加，你取的次数越多就会随之增加，但也不会超过上限值size_t batchNum = min(_freeLists[index].MaxSize(), SizeClass::NumMoveSize(size));if (_freeLists[index].MaxSize() == batchNum){_freeLists[index].MaxSize() += 1;}void* start = nullptr;void* end = nullptr;size_t actualNum = CentralCache::GetInstance()->FetchRangeObj(start, end, batchNum, size);assert(actualNum > 0);if (actualNum == 1){assert(start == end);return start;}else{_freeLists[index].PushRange(NextObj(start), end, actualNum-1);return start;}
}

当thread cache中的自由链表过长时，摘取下来给central cache ，至于怎么处理，交给central cache去处理

void ThreadCache::ListTooLong(FreeList& list, size_t size)
{void* start = nullptr;void* end = nullptr;list.PopRange(start, end, list.MaxSize());//从Thread cache中拿走CentralCache::GetInstance()->ReleaseListToSpans(start, size);//交给central cache
}

四、central cache

central cache也是一个哈希桶结构，映射关系和thread cache是一样的，不同的是，central cache 中每个桶中挂的是一个双向带头循环链表，由一个个span组成，而一个span上挂着一个个小块内存组成的自由链表

申请内存

当thread cache向central cache申请内存时，这个过程是要加锁的，但是我们这里采用的桶锁的方式，只有不同线程同时访问同一个桶时才会发生锁的竞争，这样既保证安全，又提高了效率，当central cache中也没有内存供给thread cache时，就可以向page cache中申请内存，central cache中设置变量记录分配了多少个对象出去，分配一个对象出去就让其++，这里主要是为了归还做铺垫

释放内存

当thread cache的自由链表过长时，会归还一部分给central cache，central cache中也有变量记录分配出去的内存，当分配出去的内存都回来后，可以把他们全都交给page cache ，交由page cache进行处理
central cache的设计采用了单例模式
三个接口
1 GetOneSpan 获取一个非空的span
2 FetchRangeObj 从central cache 中获取一定对象给 Thread cache
3 ReleaseListToSpans 将由thread cache还回的对象释放到span

//单例模式(饿汉模式)
class CentralCache
{
public:static CentralCache* GetInstance(){return &_sInst;}//获取一个非空的spanSpan* GetOneSpan(SpanList& list, size_t size);//从central cache 中获取一定对象给 Thread cachesize_t FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size);//将由thread cache还回的对象释放到spanvoid ReleaseListToSpans(void* start,size_t size);
private:SpanList _spanLists[BarrelNum]; //Span 桶
private:CentralCache(){}CentralCache(const CentralCache&) = delete;static CentralCache _sInst;
};

Span 的结构

#ifdef _WIN64
typedef unsigned long long PAGE_ID;
#elif _WIN32
typedef size_t PAGE_ID;
#endif
struct Span
{PAGE_ID _pageId = 0;  //大块内存起始页的页号size_t _n = 0;         //页的数量Span* _next = nullptr;Span* _prev = nullptr;size_t _useCount = 0;   //小块内存被分配出去的数量，每分出去一块++，还回来一块--void* _freeList = nullptr; //小快内存的自由链表bool _isUse = false; //是否被使用size_t _objSize = 0;  // 切好的小对象的大小};

span链表，采用双向带头循环链表

class SpanList
{
public:SpanList(){_head = new Span;_head->_next = _head;_head->_prev = _head;}Span* Begin(){return _head->_next;}Span* End(){return _head;}bool Empty(){return _head->_next == _head;}void PushFront(Span* span){Insert(Begin(), span);}Span* PopFront(){Span* front = _head->_next;Erase(front);return front;}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;}
private:Span* _head;
public:std::mutex _mtx; //这里用的是桶锁，只有在不同线程找到central cache的同一个桶时才会出现锁的竞争//而在不同的桶上时不会发生锁的竞争，从而也提高了效率，这里public主要是给其他使用
};

这些都是一些简单的关于链表的数据结构的知识
GetOneSpan 申请一个span
当thread cache的内存不够用时，就会去central cache中申请一个span，先去检查spanlist中是否有span可以给我呢？如果有直接给我，如果没有呢？那我就去page cache中找来一部分切分好挂在central cache的span上
这里在向page cache申请的时候可以顺便把central cache的桶锁解掉，这时候不会影响到其他归还的情况，也是一个小优化

Span* CentralCache::GetOneSpan(SpanList& list, size_t size)
{//检查当前的spanlist中是否有未被使用的span，如果有，直接给出，如果没有，去page cache去拿Span* it = list.Begin();while (it != list.End()){if (it->_freeList != nullptr){return it;}else{it = it->_next;}}// 可以先把central cache中的桶锁解除，在其他线程还回的时候不会阻塞list._mtx.unlock();//去page cache 中取PageCache::GetInstance()->_pageMtx.lock();Span* span = PageCache::GetInstance()->NewSpan(SizeClass::NumMovePage(size));span->_isUse = true;span->_objSize = size;PageCache::GetInstance()->_pageMtx.unlock();//取来之后对其进行切分，这里不需要加锁，因为这里只有当前线程能看到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;//先切下一块当做头，然后循环不断尾插int i = 1;while (start < end){++i;NextObj(tail) = start;tail = NextObj(tail);start += size;}NextObj(tail) = nullptr;list._mtx.lock();list.PushFront(span);return span;
}

从central cache中获取一定数量的对象给thread cache

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* span = GetOneSpan(_spanLists[index], size);assert(span);assert(span->_freeList);start = span->_freeList;end = start;size_t i = 0;size_t actualNum = 1;while (i < batchNum - 1 && NextObj(end) != nullptr){end = NextObj(end);++i;++actualNum;}span->_freeList = NextObj(end);NextObj(end) = nullptr;span->_useCount += actualNum;_spanLists[index]._mtx.unlock();//解锁return actualNum;}

useCount来记录的内存情况，当他为0时，说明所有申请出去的内存都已经回来了，就可以还回到page cache了，再由page cache进行后续的处理

void CentralCache::ReleaseListToSpans(void* start, size_t size)
{size_t index = SizeClass::Index(size);_spanLists[index]._mtx.lock();while (start){void* next = NextObj(start);Span* span = PageCache::GetInstance()->MapObjectToSpan(start);NextObj(start) = span->_freeList;span->_freeList = start;span->_useCount--;//span切分出去的所有小块内存都已经回来了// 这里就可以回收给page Cache，Page Cache可以再合并前后解决外内存碎片问题if (span->_useCount == 0){_spanLists[index].Erase(span);span->_freeList = nullptr;span->_prev = nullptr;span->_next = nullptr;//这时就可以把桶锁关闭，Page cache上锁_spanLists[index]._mtx.unlock();PageCache::GetInstance()->_pageMtx.lock();PageCache::GetInstance()->ReleaseSpanToPageCache(span);PageCache::GetInstance()->_pageMtx.unlock();_spanLists[index]._mtx.lock();}start = next;}_spanLists[index]._mtx.unlock();
}

五、page cache

page cache 的结构是根据页进行划分桶，每个同上挂着一个个span

申请内存

当central cache 向page cache申请内存时，page cache先检查对应的位置有没有span，如果有，直接分配一个span给central cache，如果没有，去顺着找更大的页的，找到的话就把其分成两部分，一部分是central cache需要的，另一部分则挂到符合要求的页桶上去，比如我需要一个5页的，找不到5页的，5页后面有个6页的，就把6页的分为一个5页的一个1页的，5页的拿走，1页的挂到1页的桶上去，如果一直都没有找到，就向系统进行申请一个128页的

释放内存

central cache 还回的span 看是否能和前后的桶上的span进行合并，如果能合并，就变成更大的页，这是解决外碎片问题的关键！！
同样，page cache也是使用单例模式

class PageCache
{
public:static PageCache* GetInstance(){return &_sInst;}//申请一个spanSpan* NewSpan(size_t k);//对象到span的映射Span* MapObjectToSpan(void* obj);//把空闲的span释放回Page cache 并在Page cache进行前后合并成更大的spanvoid ReleaseSpanToPageCache(Span* span);std::mutex _pageMtx;
private:SpanList _spanLists[NPAGES];ObjectPool<Span> _spanPool;TCMalloc_PageMap1<32 - PAGE_SHIFT>_idSpanMap;PageCache(){}PageCache(const PageCache&) = delete;//std::unordered_map<PAGE_ID, Span*> _idSpanMap;//大块内存起始页的页号 与 Span 相映射static PageCache _sInst;
};

三个接口
1 NewSpan 申请一个span
2 MapObjectToSpan 哈希表，页号和对应span的映射
3 ReleaseSpanToPageCache 把空闲的span释放回Page cache 并在Page cache进行前后合并成更大的span

申请一个span，首先看申请的这个span大小是否大于128页，如果大于128页直接由系统去堆栈上申请，如果不是，首先看对应的桶上是否有能用的，有的话直接返回即可，没有的话，顺着找更大的页，有的话，切分，没有的话系统申请一个128页的span，然后再重新执行上述操作，即可完成一次span的申请

Span* PageCache::NewSpan(size_t k)
{//大于128页时直接向堆申请assert(k > 0);if (k > NPAGES - 1){void* ptr = SystemAlloc(k);Span* span = _spanPool.New();span->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT;span->_n = k;_idSpanMap.set(span->_pageId, span);return span;}//assert(k > 0 && k < NPAGES);//先检查第k个桶里面是否有spanif (!_spanLists[k].Empty()){Span* kSpan = _spanLists[k].PopFront();for (PAGE_ID i = 0; i < kSpan->_n; ++i){_idSpanMap.set(kSpan->_pageId + i, kSpan);}return kSpan;}//没有span的话，去顺着往下找for (size_t i = k+1; i < NPAGES; ++i){if (!_spanLists[i].Empty()){//进行切割，切成两部分，一部分是我们需要的k页的span，另一部分挂到应该挂的地方Span* nSpan = _spanLists[i].PopFront();//Span* kSpan = new Span;Span* kSpan = _spanPool.New();kSpan->_pageId = nSpan->_pageId;kSpan->_n = k;nSpan->_pageId += k;nSpan->_n -= k;_spanLists[nSpan->_n].PushFront(nSpan);_idSpanMap.set(nSpan->_pageId, nSpan);_idSpanMap.set(nSpan->_pageId+nSpan->_n-1, nSpan);for (PAGE_ID i = 0; i < kSpan->_n; ++i){_idSpanMap.set(kSpan->_pageId + i, kSpan);}return kSpan;}}//到这没有return的话就是遍历之后并没有找到一个合适的span，这就需要向堆申请一个128页的span//Span* bigSpan = new Span;Span* bigSpan = _spanPool.New();void* ptr = SystemAlloc(NPAGES - 1);bigSpan->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT;//通过地址得出其起始页号bigSpan->_n = NPAGES - 1;_spanLists[bigSpan->_n].PushFront(bigSpan);return NewSpan(k);
}

哈希表（页号与span的映射）

Span* PageCache::MapObjectToSpan(void* obj)
{PAGE_ID id = ((PAGE_ID)obj >> PAGE_SHIFT);auto ret = _idSpanMap.find(id);if (ret != _idSpanMap.end()){return ret->second;}else{assert(false);return nullptr;}return ret;
}

将central cache还回的span进行前后的合并成一个更大页

void  PageCache::ReleaseSpanToPageCache(Span* span)
{if (span->_n > NPAGES - 1){void* ptr = (void*)(span->_pageId << PAGE_SHIFT);SystemFree(ptr);_spanPool.Delete(span);return;}//向前while (1){PAGE_ID prevId = span->_pageId - 1;auto ret = (Span*)_idSpanMap.get(prevId);if (ret == nullptr){break;}Span* prevSpan = ret;if (prevSpan->_isUse == true){break;}if (prevSpan->_n + span->_n > NPAGES - 1){break;}span->_pageId = prevSpan->_pageId;span->_n += prevSpan->_n;_spanLists[prevSpan->_n].Erase(prevSpan);_spanPool.Delete(prevSpan);}//向后while (1){PAGE_ID nextId = span->_pageId + span->_n;auto ret = (Span*)_idSpanMap.get(nextId);if (ret == nullptr){break;}Span* nextSpan = ret;if (nextSpan->_isUse == true){break;}if (nextSpan->_n + span->_n > NPAGES - 1){break;}span->_n += nextSpan->_n;_spanLists[nextSpan->_n].Erase(nextSpan);_spanPool.Delete(nextSpan);}//前后合并完之后，把他挂到该挂的地方，并标记未使用,修改对应的映射关系_spanLists[span->_n].PushFront(span);span->_isUse = false;_idSpanMap.set(span->_pageId, span);_idSpanMap.set(span->_pageId + span->_n - 1, span);
}

六、性能测试及优化

基本上大体的逻辑已经很清晰，这时候我们就可以试着测试一下他的性能如何，对比malloc，效率如何？
封装后的代码，方便我们直接调用

static void* ConcurrentAlloc(size_t size)
{//大于256k时if (size > MAX_BYTES){size_t alignSize = SizeClass::RoundUp(size);size_t kpage = alignSize >> PAGE_SHIFT;PageCache::GetInstance()->_pageMtx.lock();Span* span = PageCache::GetInstance()->NewSpan(kpage);span->_objSize = size;PageCache::GetInstance()->_pageMtx.unlock();void* ptr = (void*)(span->_pageId << PAGE_SHIFT);return ptr;}//TLS来使每个线程在无锁的情况下获取自己专属的ThreadCache对象else{if (pTLSThreadCache == nullptr){static ObjectPool<ThreadCache> tcPool;pTLSThreadCache = tcPool.New();}return pTLSThreadCache->Allocate(size);}
}
static void ConcurrentFree(void* ptr)
{Span* span = PageCache::GetInstance()->MapObjectToSpan(ptr);size_t size = span->_objSize;if (size > MAX_BYTES){PageCache::GetInstance()->_pageMtx.lock();PageCache::GetInstance()->ReleaseSpanToPageCache(span);PageCache::GetInstance()->_pageMtx.unlock();}else{assert(pTLSThreadCache);pTLSThreadCache->Deallocate(ptr, size);}
}

测试代码

void BenchmarkMalloc(size_t ntimes, size_t nworks, size_t rounds)
{std::vector<std::thread> vthread(nworks);std::atomic<size_t> malloc_costtime = 0;std::atomic<size_t> free_costtime = 0;for (size_t k = 0; k < nworks; ++k){vthread[k] = std::thread([&, k]() {std::vector<void*> 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轮次，每轮次malloc %u次: 花费：%u ms\n",nworks, rounds, ntimes, malloc_costtime.load());printf("%u个线程并发执行%u轮次，每轮次free %u次: 花费：%u ms\n",nworks, rounds, ntimes, free_costtime.load());printf("%u个线程并发malloc&free %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<std::thread> vthread(nworks);std::atomic<size_t> malloc_costtime = 0;std::atomic<size_t> free_costtime = 0;for (size_t k = 0; k < nworks; ++k){vthread[k] = std::thread([&]() {std::vector<void*> 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;cout << "==========================================================" << endl;BenchmarkConcurrentMalloc(n, 4, 10);cout << endl << endl;BenchmarkMalloc(n, 4, 10);cout << "==========================================================" << endl;return 0;
}

测试结果

这样看来，我们写的还是差点意思，分析一下，是哪里还可以优化呢？我们的花费主要在哪里？
仔细分析我们会发现，很多时间都花在了，哈希映射上，以及锁的竞争上
这里我就要引入一个新的东西，基数树，分为单层基数树和双层基数树，其实后面也有三层等等，与我们而言单双层足矣
单层的基数树最简单了，就是一个数组，严格的来说就是一个哈希表，一个用直接定址法来映射的哈希表，其中的 K-V 关系就是 页号-span*。
两层基数树，一共19位，前5位作为第一层数组进行哈希映射，在前五位确定后，再通过后14位进行哈希映射

//一层
template <int BITS>
class TCMalloc_PageMap1 {
private:static const int LENGTH = 1 << BITS;void** array_;public:typedef uintptr_t Number;//explicit TCMalloc_PageMap1(void* (*allocator)(size_t)) {explicit TCMalloc_PageMap1() {//array_ = reinterpret_cast<void**>((*allocator)(sizeof(void*) << BITS));size_t size = sizeof(void*) << BITS;size_t alignSize = SizeClass::_RoundUp(size, 1 << PAGE_SHIFT);array_ = (void**)SystemAlloc(alignSize >> PAGE_SHIFT);memset(array_, 0, sizeof(void*) << BITS);}// Return the current value for KEY.  Returns NULL if not yet set,// or if k is out of range.void* get(Number k) const {if ((k >> BITS) > 0) {return NULL;}return array_[k];}// REQUIRES "k" is in range "[0,2^BITS-1]".// REQUIRES "k" has been ensured before.//// Sets the value 'v' for key 'k'.void set(Number k, void* v) {array_[k] = v;}
};//两层
template <int BITS>
class TCMalloc_PageMap2 
{
public:typedef uintptr_t Number;explicit TCMalloc_PageMap2() {memset(root_, 0, sizeof(root_));PreallocateMoreMemory();}void* get(Number k) const {const Number i1 = k >> LEAF_BITS;const Number i2 = k & (LEAF_LENGTH - 1);if ((k >> BITS) > 0 || root_[i1] == NULL){return NULL;}return root_[i1]->values[i2];}void set(Number k, void* v) {const Number i1 = k >> LEAF_BITS;const Number i2 = k & (LEAF_LENGTH - 1);ASSERT(i1 < ROOT_LENGTH);root_[i1]->values[i2] = v;}bool Ensure(Number start, size_t n) {for (Number key = start; key <= start + n - 1;) {const Number i1 = key >> LEAF_BITS;if (i1 >= ROOT_LENGTH)return false;if (root_[i1] == NULL) {static ObjectPool<Leaf>	leafPool;Leaf* leaf = (Leaf*)leafPool.New();memset(leaf, 0, sizeof(*leaf));root_[i1] = leaf;}key = ((key >> LEAF_BITS) + 1) << LEAF_BITS;}return true;}void PreallocateMoreMemory() {Ensure(0, 1 << BITS);}
private:static const int ROOT_BITS = 5;static const int ROOT_LENGTH = 1 << ROOT_BITS;static const int LEAF_BITS = BITS - ROOT_BITS;static const int LEAF_LENGTH = 1 << LEAF_BITS;struct Leaf {void* values[LEAF_LENGTH];};Leaf* root_[ROOT_LENGTH];             // Pointers to 32 child nodesvoid* (*allocator_)(size_t);          // Memory allocator
};

只有在申请和释放两个函数中回去建立id和span的映射
基数树写之前会提前开好空间，写数据过程中不会动结构
读和写是分离的，一个线程对其读的时候，别的线程一定不会对其写


我们可以看到，优化后的效果是要比malloc好上不少的

七、总结

我们在做项目时，不是为了做一个多厉害的东西，重要的事我们在每一个细节中学到了什么，如果要比优劣，我们是一定比不过tcmalloc的，重要的是一件事的处理方法，站在巨人的肩膀上，我们才能看的更远，所以，学习应该是贯彻我们一生的一件事，不断的通过一些新鲜的事务来开阔自己的眼界，从而使自己成长！

八、源码

高并发内存池源码