网站更改备案,平安区wap网站建设公司,wordpress子目录多站点设置,网站开发语言怎么识别目录
1️⃣项目介绍
#x1f359;项目概述
#x1f359;知识储备
2️⃣内存池介绍
#x1f359;池化技术
#x1f359;内存池
#x1f359;内存池主要解决的问题
#x1f365;内碎片
#x1f365;外碎片
#x1f359;malloc
3️⃣ 定长内存池设计
4️⃣ 项…目录
1️⃣项目介绍
项目概述
知识储备
2️⃣内存池介绍
池化技术
内存池
内存池主要解决的问题
内碎片
外碎片
malloc
3️⃣ 定长内存池设计
4️⃣ 项目整体框架实现
5️⃣Thread Cache设计
自由链表
对齐映射规则设计
对齐大小计算
映射桶号计算
ThreadCache类 申请内存
慢开始反馈调节算法
释放内存
TLS(thread local storage)无锁访问
6️⃣Central Cache设计
SpanList链表结构设计
Central Cache类
申请内存
释放内存
7️⃣Page Cache设计
Page Cache类
映射查找Span
申请内存
释放内存
8️⃣申请释放联调
申请内存联调
释放内存联调
9️⃣大于256Kb大块内存申请释放问题
大块内存申请问题
大块内存释放问题
性能对比及基数树优化
性能对比
性能瓶颈分析
基数树优化 1️⃣项目介绍
项目概述 本项目设计一个高并发内存池Concurrent Memory Pool基于Google开源项目tcmallocThread-Caching Malloc即线程缓存的malloc实现了高效的多线程内存管理可用于替代系统的内存分配函数malloc、freeGo语言还把tcmalloc做了自己的内存分配器。 本项目旨在把tcmalloc的核心精华框架部分简化后拿来自己模拟实现出一个学习版的高并发内存池。
知识储备 本项目会用到C/C 、数据结构链表、哈希桶、操作系统内存管理、单例模式、多线程、互斥锁等方面的知识。 2️⃣内存池介绍
池化技术 所谓 “池化技术” 就是 程序先向系统申请过量的资源然后自己管理以备不时之需 。之所以要申请过量的资源是因为每次申请该资源都有较大的开销不如提前申请好了这样使用时就会变得非常快捷大大提高程序运行效率。 在计算机中有很多使用“ 池 ” 这种技术的地方除了内存池还有连接池、线程池、对象池等。以服务器上的线程池为例它的主要思想是先启动若干数量的线程让它们处于睡眠状态当接收到客户端的请求时唤醒池中某个睡眠的线程让它来处理客户端的请求当处理完这个请求线程又进入睡眠状态。 内存池 内存池的研究重点不是向操作系统申请内存而是 对已申请到的内存的管理。 内存池是指程序预先从操作系统申请一块足够大内存此后当程序中需要申请内存的时候不是直接向操作系统申请而是直接从内存池中获取同理当程序释放内存的时候并不真正将内存返回给操作系统而是返回内存池。当程序退出(或者特定时间) 时内存池才将之前申请的内存真正释放。 内存池主要解决的问题 内存池首先主要解决效率的问题系统调用的性能开销是比较大的当程序对堆的操作比较频繁时这样做的结果会严重影响程序的性能所以可以实现一个内存池对内存进行管理而不是交给内核去进行系统调用。 其次分配内存时还要解决内存碎片的问题内存碎片分为内碎片和外碎片。
内碎片 内碎片的产生是因为申请内存空间时根据设计的对齐规则导致分配出去的空间有可能会有部分空间未被利用这些在已经分配出去但未被使用的内存空间就是内碎片。 外碎片 外碎片的产生是因为2段空间不连续碎片化即使有足够的内存空间也无法申请出来。 malloc C/C中我们要动态申请内存都是通过 malloc 去申请内存但是我们要知道实际我们不是直接使用系统调用去堆获取内存的而是通过内存池去进行管理的向系统获取一块大内存然后切开分配给程序当不够时再向系统申请大内存。malloc 就是一个内存池底层设计是ptmalloc。 参考博客 malloc的底层实现ptmalloc_z_ryan的博客-CSDN博客
3️⃣ 定长内存池设计
设计一个定长的内存池为了将申请和释放与malloc分开本项目要和malloc进行性能比较那么各处实现就不能调用malloc以及对应的freenew和delete是C的一个关键字其底层调用了malloc和free所以我们要避开使用C的关键字自己实现一个New和Delete。 定长内存池设计结构如下 //定长内存池
templateclass T
class ObjectPool {
public:T* New(){T* obj nullptr;//如果有还回的内存直接使用还回的内存块if (_freeList){obj (T*)_freeList;_freeList *(void**)obj;//内存块中首个指针大小头4/8字节存的是下一个还回内存块的地址}else{//如果内存块为空或者剩余的内存块不足以继续申请T对象if (_remainbytes sizeof(T)){_remainbytes 128 * 1024;//128kb_memory (char*)SystemAlloc(_remainbytesPAGE_SHIFT);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调用对象的构造函数创建对象不会自动分配内存new(obj)T;return obj;}void Delete(T* obj){//因为定位new不会管理内存释放必须显示调用对象的析构函数obj-~T();//头插到freeList*(void**)obj _freeList;_freeList obj;}
private:char* _memory nullptr;//指向内存块的指针void* _freeList nullptr;//管理还回内存的自由链表 int _remainbytes 0;//内存块中剩余的字节数
};
自由链表取到下一个内存块的地址设计在Thread Cache设计中自由链表模块有详细介绍定长内存池在Windows下使用系统调用VirtualAlloc从堆中申请内存在Linux下使用brk或mmap。
//堆上申请内存
inline static void* SystemAlloc(size_t kpage)
{
#ifdef _WIN32void* ptr VirtualAlloc(0, kpage * (1 PAGE_SHIFT),MEM_COMMIT | MEM_RESERVE, PAGE_READWRITE);
#else// brk mmap等
#endifif (ptr nullptr)throw std::bad_alloc();return ptr;
}
有了内存还需要创建对象这里采用定位new来调用对象的构造函数进行创建对象因为定位new不会管理内存释放所以我们在释放的时候要显示调用对象的析构函数对资源进行清理并且我们的释放实际上并不归还内存而只是释放资源然后将内存挂在自由链表进行管理。
4️⃣ 项目整体框架实现 高并发内存池整体框架——三级缓存 现代很多的开发环境都是多核多线程在申请内存的场景下必然存在激烈的锁竞争问题。malloc 本身其实已经很优秀但本项目的原型 tcmalloc在多线程高并发的场景下更胜一筹 所以本项目实现的内存池需要考虑以下几方面的问题 性能问题。 多线程环境下锁竞争问题。 内存碎片问题。 高并发内存池Concurrent Memory Pool三级缓存 ⭐线程缓存Thread Cache——无锁 ⭐中心缓存Central Cache——桶锁 ⭐页缓存Page Cache ——整体锁
设计Thread Cache分配对象最大256Kb根据定义的对齐映射规则计算出Thread Cache和Central Cache总桶数为208Page Cache桶数按页数设计为1290号桶不参与采取线性映射最大页数为128假设1页8Kb128*8Kb1Mb,完全够给最大字节256Kb分4个。
static const size_t MAX_BYTES 256 * 1024;//threadcache最大256kb
static const size_t NFREELISTS 208;//使用static const代替define208是根据定义的字节对齐算出的总共桶数
static const size_t NPAGES 129;//总共Page桶数128为最大页数假设1页有8Kb128*8Kb1Mb完全够给最大字节256Kb分4个
static const size_t PAGE_SHIFT 13;//2^13 页大小8k
5️⃣Thread Cache设计 Thread Cache线程缓存是每个线程独有的用于小于 256KB 的内存的分配 线程从这里申请内 存不需要加锁利用TLS无锁访问机制每个线程独享一个 cache 这也就是这个并发线程池高效的地方 。 Thread Cache是 哈希桶结构每个桶是一个按桶位置映射大小的内存块对象的自由链表 。每个线程都会有一个thread cache 对象这样每个线程在这里获取对象和释放对象时是无锁的。 自由链表 自由链表管理释放回来的小内存块和中心缓存中分配的未使用的小内存块结构如下 自由链表结构 因为自由链表是用来管理小内存块的所以其必须能够指向下一块小内存那么当对象的大小当前平台指针大小时需要按指针的大小进行划分。 关于不同平台的问题32位平台 指针大小4Byte64位平台 指针大小8Byte。那么如何设计获取指针大小呢
*(void**) 获取指针大小地址//获取结点obj存的下一个结点地址前4/8字节,加static仅当前文件可见防止重定义
static void* NextObj(void* obj)
{return *(void**)obj;
} * 解引用本质上是对地址区间进行获取类型大小的内容比如int*对int*进行 * 解引用实际上是获取int类型大小的地址内容也就是4Byte的内容。 void**是指针的指针*void**就是对获取void*类型大小的地址内容此时如果是32位平台就获得了4Byte大小内容如果是64位平台就获得了8Byte大小内容。 在Thread Cache中哈希桶每个桶就是一个自由链表自由链表中一定会有插入、删除、判空等操作并且我们还可以记录个数_size_maxSize这个桶最多能挂多少个那么这么多个自由链表就需要被管理我们设计一个管理自由链表的结构
//管理切好的小对象自由链表
class FreeList
{
public:void Push(void* obj){//头插NextObj(obj) _freeList;_freeList obj;_size;}void PushRange(void* start, void* end,size_t n){NextObj(end) _freeList;_freeList start;_size n;}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;}void* Pop( ){//头删void* obj _freeList;_freeList NextObj(obj);--_size;return obj;}bool Empty(){return _freeList nullptr;}size_t Size(){return _size;}size_t MaxSize(){return _maxSize;}
private:void* _freeList nullptr;size_t _maxSize 1;//自由链表最大个数size_t _size 0;
};
对齐映射规则设计
对齐大小计算
[1,128]8byte对齐freelist[0,16)[1281,1024]16byte对齐freelist[16,72)[10241,8*1024]128byte对齐freelist[72,128)[8*10241,64*1024]1024byte对齐freelist[128,184)[64*10241,256*1024]8*1024byte对齐freelist[184,208) 该设计规则除了第一个桶的内碎片浪费大保证其他桶内碎片浪费整体保证在10%左右。 内碎片浪费率浪费的字节/分配的字节比如现在有129字节就要分配144字节只使用第一个16byte对齐桶的1个字节浪费15字节但总共分配了12816144字节所以内碎片浪费率15/14410.4% 根据设计规则通过传入参数字节数进行简单逻辑判断跳转至子函数_RoundUp进行对齐后的字节数计算。
//对齐大小计算static inline size_t RoundUp(size_t bytes){assert(bytes MAX_BYTES);if (bytes 128){return _RoundUp(bytes, 8);}else if (bytes 1024){return _RoundUp(bytes, 16);}else if (bytes 8 * 1024){return _RoundUp(bytes, 128);}else if (bytes 64 * 1024){return _RoundUp(bytes, 1024);}else if (bytes 256 * 1024){return _RoundUp(bytes, 8*1024);}else{assert(false);}return -1;} 对齐后的字节数计算函数_RoundUp设计我们学习参考tcmalloc的实现采用位运算的方式进行该设计思路十分巧妙值得我们去学习使用。 //计算对齐后的bytes大小static inline size_t _RoundUp(size_t bytes, size_t alignNum){return (bytes alignNum - 1) ~(alignNum - 1);} 例子: bytes7 alignNum8 alignNum-17 0000 0111 ~(alignNum-1) 1111 1000 78-115 0000 1111 0000 1000 8 对齐后所占大小 bytes9 alignNum8 98-116 0001 0000 0001 0000 16 对齐后所占大小 映射桶号计算 首先根据上面设计的对齐映射规则我们可以计算得到对应桶号的区间利用数组将区间桶号保存再使用简单逻辑判断进入子函数_Index计算当前所在区间映射到的桶号最终对齐映射的桶号区间前的桶数当前区间桶号
//计算映射在哪一个桶static inline size_t Index(size_t bytes){assert(bytes MAX_BYTES);//每个字节对齐数区间的最大链数桶数static int group[4] { 16,56,56,56 };if (bytes 128) {return _Index(bytes, 3);}else if (bytes 1024) {return _Index(bytes - 128, 4) group[0];}else if (bytes 8 * 1024) {return _Index(bytes - 1024, 7) group[1] group[0];}else if (bytes 64 * 1024) {return _Index(bytes - 8 * 1024, 10) group[2] group[1] group[0];}else if (bytes 256 * 1024) {return _Index(bytes - 64 * 1024, 13) group[3] group[2] group[1] group[0];}else {assert(false);}return -1;} 同样的在这里学习参考tcmalloc的设计巧妙使用位运算进行当前区间桶号计算位运算比算术运算更加高效。
//计算当前对齐大小对应的所在桶号static inline size_t _Index(size_t bytes, size_t align_shift){return ((bytes (1 align_shift) - 1) align_shift) - 1;} 例子 [1,8] align_shift3 138 ((18-1)3)-10 0号桶 ... ((88-1)3)-10 0号桶 [9,16] align_shift3 138 ((98-1)3)-11 1号桶 ... ((168-1)3)-10 1号桶 bytes129 抛去bytes128前的桶只剩1bytes再分配16字节对齐的0号桶, 总桶号就是前128bytes桶号当前16bytes的桶号 ThreadCache类
class ThreadCache
{
public:// 申请和释放内存对象void* Allocate(size_t size);void Deallocate(void* ptr, size_t size);// 从中心缓存获取对象void* FetchFromCentralCache(size_t index, size_t size);//释放对象时链表过长时回收内存回到中心缓存void ListTooLong(FreeList list, size_t size);private:FreeList _freeLists[NFREELISTS];
};//TLS——无锁使变量在线程与线程之间独立
static __declspec(thread) ThreadCache* TLS_ThreadCache nullptr; 申请内存
//申请内存
void* ThreadCache::Allocate(size_t size)
{assert(size MAX_BYTES);size_t alignSize AMSize::RoundUp(size);size_t index AMSize::Index(size);if (!_freeLists[index].Empty()){return _freeLists[index].Pop();}else{//去中心缓存取return FetchFromCentralCache(index,alignSize);}
} ⭐ 当内存申请 size256KB 先获取到线程本地存储的 Thread Cache 对象计算 size 映射的哈希桶自由链表下标i 。 ⭐如果自由链表_freeLists[i] 中有对象则直接 Pop 一个内存对象返回。 Pop函数在上面的FreeList类中因为是从自由链表上取走一个去使用所以需要返回值void* //头删
void* Pop( ){void* obj _freeList;_freeList NextObj(obj);--_size;return obj;} ⭐如果_freeLists[i] 中没有对象时则批量从 Central Cache 中获取一定数量的对象头插入到自由链表并返回一个对象。 void* ThreadCache::FetchFromCentralCache(size_t index,size_t alignSize)
{size_t batchNum min(_freeLists[index].MaxSize(), AMSize::NumMoveSize(alignSize));if (_freeLists[index].MaxSize() batchNum){//想修改返回值所以使用引用作为MaxSize返回值_freeLists[index].MaxSize() 1;}void* start nullptr;void* end nullptr;size_t actualNum CentralCache::GetInstance()-FetchRangeObj(start, end, batchNum, alignSize);assert(actualNum 1);if (actualNum 1){assert(start end);return start;}else{//返回1个(start)剩下的(从start下一个开始)挂接到桶上_freeLists[index].PushRange(NextObj(start), end,actualNum-1);return start;}
} 对于需求不同字节大小从Central Cache获取的分配个数又需要考虑性能 对于分配8bytes可以多分配一些但要有上限对于256*1024bytes则少分配些但要有下限 采用慢开始反馈调节算法 1.最开始不会一次向Central Cache一次批量要太多因为要太多可能用不完 2.如果不要这个size大小内存需求那么betchNum就会不断增长直到上限。 3.size越大一次向Central Cache要的batchNum就越小 4.size越小一次向Central Cache要的batchNum就越大
慢开始反馈调节算法
// 一次从中心缓存获取多少个static size_t NumMoveSize(size_t size){assert(size 0);// [2, 512]一次批量移动多少个对象的(慢启动)上限值// 小对象一次批量上限高// 小对象一次批量上限低int num MAX_BYTES / size;if (num 2)num 2;if (num 512)num 512;return num;}
如果只需要8Byte大小从Central Cache获取批量数就是256*1024/8其结果大于512返回512个如果需要256Kb大小从Central Cache获取批量数就是256Kb/256Kb1其结果小于2返回2个。 这样设计批量在于确定上下限不会使得从中心缓存获取的小块内存过多或过少如果获取过多一直不使用达到一定数量时又会回收给Central Cache多此一举所以确定上下限。计算结果在上下限之间的就返回计算个数。
释放内存 ⭐当释放内存小于256Kb 时将内存释放回 Thread Cache 计算 size 映射自由链表桶位置 i 将对象 Push到_freeLists[i] 。 //释放内存
void ThreadCache::Deallocate(void* ptr, size_t size)
{assert(ptr);assert(size MAX_BYTES);//找到映射的自由链表桶对象插入进去size_t index AMSize::Index(size);_freeLists[index].Push(ptr);//当链表长度大于一次批量申请的内存时就开始还一段list给CentralCache if (_freeLists[index].Size() _freeLists[index].MaxSize()){ListTooLong(_freeLists[index], size);}
} ⭐当链表的长度过长则回收一部分内存对象到Central Cache 。 void ThreadCache::ListTooLong(FreeList list, size_t size)
{void* start nullptr;void* end nullptr;list.PopRange(start, end, list.MaxSize());CentralCache::GetInstance()-ReleaseListToSpans(start, size);
}
start和end在PopRange中是输出型参数进入PopRange中进行头删将待回收的链表内存对象拿出来返还给Central Cache 。
TLS(thread local storage)无锁访问 我们在设计中要求每一个线程都有一个独属于自己的ThreadCache类如果我们把他ThreadCache类实现为全局的那么必然每个线程共享这个类势必会发生竞争问题需要加锁。 频繁的控制锁的加锁和解锁会增加时间成本这显然和我们要的高性能不相符所以这里提出一个变量存储方法TLS线程局部存储TLS该方法下变量在当前线程下是全局可访问的在线程和线程之间是独立局部的这有效的实现了每个线程独属于自己的类避免加锁。
//TLS——无锁使变量在线程与线程之间独立
static __declspec(thread) ThreadCache* TLS_ThreadCache nullptr;
我们使用TLS机制创建一个ThreadCache类指针进行多线程下创建线程独立的类。该指针在申请和释放联调过程中调用。
6️⃣Central Cache设计 Central Cache也是一个哈希桶结构他的哈希桶的映射关系跟T hread Cache 是一样的。不同的是他的每个哈希桶位置挂是SpanList 链表结构不过每个映射桶下面的 span 中的大内存块被按映射关系切成了一个个小内存块对象挂在span 的自由链表中。 SpanList链表结构设计
Span管理以页为单位大小的大内存块
Span是以页为单位那么就涉及到一个问题页号在32位下最高2^32/2^132^192^19我们需要4字节大小来表示可以用size_t类型可以表示但如果是64位下页号最高2^64/2^82^51我们需要8字节大小来表示可以用unsigned long long类型。所以我们使用条件编译进行判断使用何种变量
#ifdef _WIN64typedef unsigned long long PAGE_ID;
#elif _WIN32typedef size_t PAGE_ID;
#else//linux
#endif
细节64位系统下包含了宏_WIN32和_WIN64如果把_WIN32放在最开始判断那么就无法识别出64位系统会一直识别为32位所以我们将_WIN64放在最开始判断64位系统
但实际上size_t在64位下是unsigned long long 或者unsigned _int64类型范围[0,2^64 -1]32位下是unsigned int类型。如果想要编写可移植的代码应该避免直接使用int或long类型而是要使用size_t类型。 所以你也可以简化为
#ifdef _WIN32typedef size_t PAGE_ID;//64位下也有宏_WIN32
#else//linux
#endif Span里存储页号、页数、前后指针、切分小块内存的大小用于释放的时候传参、切分好的小块内存的数目回收对象如果Span内切分出去的对象全部回收即_useCount0回收Span给PageCache进行页合并、切好小块内存的自由链表、该Span是否被使用用以合并Span判断
struct Span
{PAGE_ID _pageId 0;//大块内存起始页的页号size_t _n 0;//页的数量,本质和PageCache中的SpanList数组桶下标一致可以用来寻找挂接的桶位置Span* _next nullptr;Span* _prev nullptr;size_t _objSize 0;//切好的对象大小size_t _useCount 0;//切好小块内存,分配个thread_cache的计数void* _freeList nullptr;//切好小块内存的自由链表bool _isUse false;//判断是否被使用
};
SpanList带头双向循环链表其结构如下一个头结点以及桶锁
//带头双向循环链表
class SpanList
{
public:SpanList(){_head new Span;_head-_prev _head;_head-_next _head;}Span* Begin(){return _head-_next;}Span* End(){return _head;}void PushFront(Span* span){Insert(Begin(), span);}void Insert(Span* pos, Span* newSpan){assert(pos);Span* prev pos-_prev;prev-_next newSpan;newSpan-_prev prev;pos-_prev newSpan;newSpan-_next pos;}Span* PopFront( ){Span* span _head-_next;Erase(span);return span;}void Erase(Span* pos){assert(pos);assert(pos ! _head);//不能删带头结点Span* prev pos-_prev;Span* next pos-_next;prev-_next next;next-_prev prev;}bool Empty(){return _head-_next _head;}
private:Span* _head;
public:std::mutex _mtx;
};
Central Cache类 Central Cache中心缓存是所有线程所共享 Thread Cache 是 按需从 central cache 中获取 的对象。Central Cache 合适的时机回收 Thread Cache 中的对象避免一个线程占用了太多的内存而其他线程的内存吃紧达到内存分配在多个线程中更均衡的按需调度的目的 。 Central Cache是存在竞争的所以从这里取内存对象是需要加锁首先这里用的是桶锁其次只有 Thread Cache当 没有内存对象时才会找 Central Cache 所以这里竞争不会很激烈 。 Central Cache是所有线程共享的所以只设计1个并且当程序运行的时候我们就要创建出来所以我们用单例模式的饿汉模式。
#pragma once
#includeCommon.h//因为所有线程对象共用一个CentralCache
//所以设计成单例模式
class CentralCache
{
public:static CentralCache* GetInstance(){return _pInst;}//获取一个非空的spanSpan* GetOneSpan(SpanList list, size_t size);//从中心缓存获取一定数量的对象给Thread Cachesize_t FetchRangeObj(void* start, void* end, size_t batchNum, size_t size);// 将一定数量的对象释放到span跨度void ReleaseListToSpans(void* start, size_t size);private:SpanList _spanLists[NFREELISTS];
private://构造函数私有CentralCache(){}//禁止拷贝构造函数CentralCache(const CentralCache) delete;static CentralCache* _pInst;//声明
};
申请内存 ⭐当Thread Cache 中没有内存时就会批量向 Central Cache 申请一些内存对象这里的批量获取对象的数量使用了类似网络tcp协议拥塞控制的慢开始算法Central Cache也有一个哈希映射的 spanList spanList 中挂着 span 从 span中取出对象给Thread Cache这个过程是需要加锁的不 过这里使用的是一个桶锁尽可能提高效率。 从Central Cache中的span取对象那么一定是Thread Cache的桶中没有剩余的对象因为我们是从span中获取的那么一定是一端连续的内存我们 只需要首位地址就可以而且需要将首位地址返回设置为输出型参数 用来给Thread Cache头插挂接一段PushRange对象。 //从中心缓存获取一定数量的对象给Thread Cache
size_t CentralCache::FetchRangeObj(void* start, void* end, size_t batchNum, size_t size)
{size_t index AMSize::Index(size);_spanLists[index]._mtx.lock();//加桶锁Span* span GetOneSpan(_spanLists[index], size);assert(span);assert(span-_freeList);//从span中获取batchNum个对象如果不够batchNum个有多少拿多少end start span-_freeList;size_t i 0;size_t actualNum 1;while (i batchNum - 1 NextObj(end) ! nullptr){end NextObj(end);i;actualNum;}//span中内存的自由链表指向分出后的余下内存结点span-_freeList NextObj(end);//分出的最后个结点指向空NextObj(end) nullptr;span-_useCount actualNum;_spanLists[index]._mtx.unlock();return actualNum;
} 这里使用桶锁防止多个线程同时访问一个桶造成线程安全问题。 并且从Central Cache中的span切分在GetOne中切分batchNum对象给Thread Cache但是可能实际上span并没剩下那么多只能将剩下的分配给Thread Cache所以需要统计一个实际值actualNum_useCountactualNum更新span中切分出去的对象保证回收不会出错。 返回实际分配到的对象数目在Thread Cache中返回1个使用剩余的actualNum头插挂接到Central Cache对应的桶上。 ⭐Central Cache映射的spanList 中所有 span 的都没有内存以后则需要向 Page Cache 申请一个新的span对象拿到 span 以后将 span 管理的内存按大小切好作为自由链表链接到一起。然后从 span中取对象给Thread Cache。 //获取一个非空的span
Span* CentralCache::GetOneSpan(SpanList list, size_t size)
{//查看当前的spanlist这时是否有 未分配对象的spanSpan* it list.Begin();while (it ! list.End()){if (it-_freeList ! nullptr){return it;}else{it it-_next;}}//先把CentralCache的桶锁解掉这样如果其他线程释放内存对象回来不会阻塞list._mtx.unlock();//走到此说明没有空闲span了只能找PageCache要PageCache::GetInstance()-_pageMtx.lock();Span* spanPageCache::GetInstance()-NewSpan(AMSize::NumMovePage(size));span-_isUse true;span-_objSize size;//(小于256Kb)三级缓存中一定是从PageCache中去拿存储对象大小PageCache::GetInstance()-_pageMtx.unlock();//切分span并挂接到桶此时不需要加锁因为这会其他线程访问不到这个span//计算span的大块内存的起始地址和大块内存的大小字节数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;while (start end){NextObj(tail) start;tail NextObj(tail);start size;}//最后一个span内的小内存块指向空NextObj(tail) nullptr;//切好span后挂接到桶需要加锁list._mtx.lock();list.PushFront(span);return span;
} 如果Central Cache当前桶有剩余的span直接返回该span不需要去Page Cache申请span。 如果没有剩余span解开桶锁进入PageCache中获取span获取后记录使用情况和存储对象大小并且Page Cache实际上我们也只设计了1个所以他也需要加锁。 为什么要解开桶锁 CentralCache是桶锁PageCache是整个锁。 在CentralCache::GetOneSpan中获取一个span需要从Page获取Span时先把桶锁解掉如果此时线程1和2都执行GetOneSpan因为PageCache::NewSpan有整个锁产生阻塞也不会产生混乱。 也就是说CentralCache在此时解不解锁在获取Span时作用一样但是我可以线程1在这个桶拿Span并且线程2在这个桶释放Span为了提高效率所以我们解开桶锁。 页缓存获取span是按页来分配的所以接口NewSpan需要传递页数我们设计NumMovePage获取页数传递申请的对象对齐大小先进入NumMoveSize获取向Central Cache申请的span个数对齐大小*个数总Byte总Byte/页大小需要的页数不满足1页给1页。 static size_t NumMovePage(size_t size){size_t num NumMoveSize(size);size_t npage num * size; //算出需要的总Byte大小npage PAGE_SHIFT; //总Byte大小/页大小需要的页数if (npage 0)npage 1;return npage;} 从Page Cache中获取span后我们span中只存储了页信息但没有他的地址信息那我们怎么获得地址去管理连接内存对象呢 这里就要引入一个概念页的起始地址页号*页大小 页的尾地址起始地址页的数量*页的大小 页号页的起始地址/页大小 那么在相邻页之间地址其地址大小小于后面一页的起始地址➗页大小必定也能得到该页的页号。这在回收中有着重要作用。 从Page Cache中获取到span后我们通过上面的概念可以计算出该span的起始地址和尾地址我们再根据对象大小进行切分因为内存物理上其实是连续的而我们这里要在抽象的把他形成链式结构我们就需要通过尾插来保证地址的连续。切好后将该span挂在Central Cache的桶。 ⭐Central Cache的中挂的span 中_ useCount 记录分配了多少个对象出去分配一个对象给Thread Cache就 _useCount。 释放内存 ⭐当Thread Cache 过长或者线程销毁则会将内存释放回 Central Cache 中的释放回来时 _ useCount-- 。当 useCount 减到 0 时则表示所有对象都回到了 span 则将 span 释放回 Page Cache Page Cache 中会对前后相邻的空闲页进行合并。 // 将一定数量的对象释放到span跨度
void CentralCache::ReleaseListToSpans(void* start, size_t size)
{//找到在哪个桶上size_t index AMSize::Index(size);_spanLists[index]._mtx.lock();//加锁因为有桶锁防止多线程竞争//回收到spanwhile (start){void* next NextObj(start);//找到对应的span小内存自由链表头插Span* span PageCache::GetInstance()-MapObjToSpan(start);NextObj(start) span-_freeList;span-_freeList start;span-_useCount--;//如果为0说明span切分出的小块内存都回来了这个span可以再回收给PageCache再尝试去前后页合并if (span-_useCount 0){//从桶里拿掉这个span_spanLists[index].Erase(span);//知道span的页号就可以知道span的起始地址从而找到整块span不需要考虑小块内存链表_freeList了span-_freeList nullptr;span-_next nullptr;span-_prev nullptr;_spanLists[index]._mtx.unlock();//已经拿掉span了可以释放桶锁给别人PageCache::GetInstance()-_pageMtx.lock();PageCache::GetInstance()-ReleaseSpanToPageCache(span);PageCache::GetInstance()-_pageMtx.unlock();_spanLists[index]._mtx.lock();}start next;}_spanLists[index]._mtx.unlock();
} 头插回收一定数量对象到span如果全部回收即_useCount0则可以将该span拿给Page Cache进行页的合并。 那么如何通过地址获取对应的span呢我们就需要调用MapObjToSpan函数来获取这将在下面介绍。 7️⃣Page Cache设计 Page Cache 页缓存是在C entral Cache 缓存上面的一层缓存存储的内存是以页为单位存储及分 配的C entral Cache 没有内存对象时从P age Cache 分配出一定数量的 page 并切割成定长大小 的大块内存分配给 Central Cache 。 当一个 span 的几个跨度页的对象都回收以后P age Cache 会回收C entral Cache 满足条件的 span 对象并且合并相邻的页组成更大的页缓解内存碎片 的问题。 Page Cache类
Page Cache我们在设计中也是只有一个 所以设置成单例模式。 并且在Page Cache中我们桶的映射规则与上面2级缓存不同这里采用直接定址法i号桶挂i页内存。 桶的个数根据需求而定我们申请内存最大是256Kb页大小为8K也就是说我们要想申请一个256Kb的对象就必须要256/83232页的span那么我们可以多分配一些设置桶个数为128128页可以申请4个256Kb对象。实际上128页就是1Mb大小。 页缓存中主要对页进行操作所以我们有必要对页和span建立一个映射关系方便我们查找管理所以使用哈希表unordered_mapPAGE_ID,Span* 对页缓存的访问需求实际上很少所以我们使用一个整体锁来进行管理线程安全即可避免频繁调用锁消耗时间。 在创建Span中我们使用了最开始设计的定长内存池来申请和释放对象与new和delete分离。
#pragma once#includeCommon.h
#includeObjectPool.h
//单例模式
class PageCache
{
public:static PageCache* GetInstance(){return _pInst;}//获取从对象到span的映射Span* MapObjToSpan(void* obj);//获取一个K页SpanSpan* NewSpan(size_t k);// 释放空闲span回到Pagecache并合并相邻的spanvoid ReleaseSpanToPageCache(Span* span);std::mutex _pageMtx;//全局锁private:SpanList _spanLists[NPAGES];//页数作桶的映射下标std::unordered_mapPAGE_ID, Span*_idSpanMap;ObjectPoolSpan_spanPool;
private:PageCache(){}PageCache(const PageCache) delete;static PageCache* _pInst;};
映射查找Span 根据Central Cache申请内存部分引入的概念我们可以得知页的起始地址*页大小页号我们可以通过这个公式得到页号然后在哈希表中查找到对应的span。 这里我们使用RAII原则的unique_lock构造时加锁出作用域对象解锁防止程序异常退出导致死锁优化代码。
//通过页的起始地址找到页从而映射找到span
Span* PageCache::MapObjToSpan(void* obj)
{//算页号PAGE_ID id (PAGE_ID)obj PAGE_SHIFT;std::unique_lockstd::mutexlock(_pageMtx);//RAII思想构造时加锁出作用域对象销毁调用析构函数解锁查找auto ret _idSpanMap.find(id);if (ret ! _idSpanMap.end()){return ret-second;}else{assert(false);return nullptr;}
}
申请内存 ⭐当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 。 ⭐如果找到_spanList[128] 都没有合适的 span 则向系统使用 mmap 、 brk 或者是 VirtualAlloc 等方式申请128 页 page span 挂在自由链表中再重复 1 中的过程。 //获取一个k页的span
Span* PageCache::NewSpan(size_t k)
{assert(k 0);if (k NPAGES - 1){//页数大于128直接向堆申请void* ptr SystemAlloc(k);//Span* span new Span;Span* span_spanPool.New();//页号*页大小该页的起始地址span-_pageId (PAGE_ID)ptr PAGE_SHIFT;span-_n k;_idSpanMap[span-_pageId] span;//记录pageId和span映射关系,方便释放的时候通过页找到span//_idSpanMap.set(span-_pageId, span);//基数树优化return span;}//先检查第k个桶里面有没有spanif (!_spanLists[k].Empty()){Span* kSpan _spanLists[k].PopFront();//建立id和span的映射方便CentralCache回收小块内存时查找对应的spanfor (PAGE_ID i 0; i kSpan-_n; i){_idSpanMap[kSpan-_pageId i] kSpan;//_idSpanMap.set(kSpan-_pageId i, kSpan);//基数树优化}return kSpan;//kSpan页返回给CentralCache}//检查后面桶里有没有spanfor (size_t i k 1; i NPAGES; i){if (!_spanLists[i].Empty()){Span* nSpan _spanLists[i].PopFront();//Span* kSpan new Span;Span* kSpan _spanPool.New();//在nSpan的头部切下k页//k页span返回给CentralCachenSpan再挂接到对应映射的位置kSpan-_pageId nSpan-_pageId;kSpan-_n k;nSpan-_pageId k;//更新编号nSpan-_n - k;//既是剩余页数也是映射位置_spanLists[nSpan-_n].PushFront(nSpan);//挂接//存储nSpan的首尾页号跟nSpan映射方便PageCache回收内存时进行合并查找_idSpanMap[nSpan-_pageId] nSpan;_idSpanMap[nSpan-_pageId nSpan-_n - 1] nSpan;//nSpan最后一个页号//_idSpanMap.set(nSpan-_pageId, nSpan);//_idSpanMap.set(nSpan-_pageId nSpan-_n - 1, nSpan);//基数树优化//建立id和span的映射方便CentralCache回收小块内存时查找对应的spanfor (PAGE_ID i 0; i kSpan-_n; i){_idSpanMap[kSpan-_pageIdi] kSpan;//_idSpanMap.set(kSpan-_pageId i, kSpan);//基数树优化
; }return kSpan;//kSpan页返回给CentralCache}}//走到这说明后面没有大页的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);
} 如果申请页大于128页则需要向堆申请我们后续再说。如果该桶还有span则直接取出span给Central Cache并哈希表保存页号和span的映射。如果该桶没有则从后面的桶中取span并更新该span被切后的页号和页数再挂接到对应页号的桶上建立页号和span的映射关系方便后续回收。如果后续桶也没有span则向系统堆申请128页的span挂接到128号桶再递归调用切出要的页span。 释放内存 ⭐如果central cache 释放回一个 span 则依次寻找 span 的前后 page id 的没有在使用的空闲 span 看是否可以合并如果合并继续向前寻找。这样就可以将切小的内存合并收缩成大的 span 减少 内存碎片 。 void PageCache::ReleaseSpanToPageCache(Span* span)
{//大于128页直接还给堆if (span-_n NPAGES - 1){void* ptr (void*)(span-_pageId PAGE_SHIFT);SystemFree(ptr);//delete span;_spanPool.Delete(span);return;}//尝试span前后页合并缓解内存外碎片问题while (1){PAGE_ID prevId span-_pageId - 1;auto ret _idSpanMap.find(prevId);//如果没有前面的页号,不合并了if (ret _idSpanMap.end()){break;}//如果前面的相邻页span在使用不合并了Span* prevSpan ret-second;/*auto ret (Span*) _idSpanMap.get(prevId);if (ret nullptr){break;}Span* prevSpan ret;*///基数树优化if (prevSpan-_isUse true){break;}//如果合并超过128页的span没办法管理不合并了if (prevSpan-_n span-_n NPAGES - 1){break;}//合并span-_pageId prevSpan-_pageId;span-_n prevSpan-_n;//合并了要删除挂接在桶上的prevSpan_spanLists[prevSpan-_n].Erase(prevSpan);//delete prevSpan;_spanPool.Delete(prevSpan);}while (1){PAGE_ID nextId span-_pageId span-_n;auto ret _idSpanMap.find(nextId);if (ret _idSpanMap.end()){break;}Span* nextSpan ret-second;/*auto ret (Span*)_idSpanMap.get(nextId);if (ret nullptr){break;}Span* nextSpan ret;*///基数树优化if (nextSpan-_isUse true){break;}if (span-_n nextSpan-_n NPAGES - 1){break;}//合并span-_n nextSpan-_n;_spanLists[nextSpan-_n].Erase(nextSpan);//delete nextSpan;_spanPool.Delete(nextSpan);}//前后页合并后的span或者无法合并的span挂接到在PageCache对应桶_spanLists[span-_n].PushFront(span);span-_isUse false;_idSpanMap[span-_pageId] span;_idSpanMap[span-_pageId span-_n - 1] span;/*_idSpanMap.set(span-_pageId, span);_idSpanMap.set(span-_pageId span-_n - 1, span);*///基数树优化
} 如果归还页大于128页则直接还给堆同样我们下面再讲。首先向相邻前页合并再向相邻后页合并。如果相邻页没有就不合并跳出如果相邻页正在使用就不合并跳出这里为什么要使用_isUse而不使用_useCount0呢如果合并页超过128无法管理不合并跳出。走完前后页合并逻辑后将页挂接到Page Cache的桶并建立映射关系。 为什么要使用_isUse而不使用_useCount0来判断相邻页是否正在被使用呢 因为可能在给CentralCache划分span的时候_usercount还未此时还是0恰好有可能其他线程在PageCache判断此时划分给CentralCache的为0拿来合并这就造成了线程安全的问题。 解决方法span增加一个bool值判断是否被使用 8️⃣申请释放联调
申请内存联调 接口ConcurrentAlloc联调程序申请内存
static void* ConcurrentAlloc(size_t size)
{if (TLS_ThreadCache nullptr){//TLS_ThreadCache new ThreadCache;static ObjectPoolThreadCachetcPool;TLS_ThreadCache tcPool.New();}//cout std::this_thread::get_id() ; TLS_ThreadCache endl;return TLS_ThreadCache-Allocate(size);}
}
释放内存联调 接口ConcurrentFree联调程序释放内存
static void ConcurrentFree(void* ptr)
{Span* span PageCache::GetInstance()-MapObjToSpan(ptr);//通过映射关系找到spansize_t size span-_objSize;assert(TLS_ThreadCache);TLS_ThreadCache-Deallocate(ptr, size);
}
9️⃣大于256Kb大块内存申请释放问题
大块内存申请问题
我们三级缓存的设计主要考虑的是小于256Kb的对象那如果大于256Kb我们如何处理呢 在Page Cache中我曾提到256Kb需要32页但我们Page Cache设计的最大有128页。所以如果申请对象大于32页小于等于128页我们可以直接向Page Cache申请内存如果大于128页我们就需要向系统堆空间申请内存 修改联调程序大于256Kb我们就直接去Page Cache Page Cache中大于128页向堆申请内存小于等于则继续按逻辑获取页内存。 大块内存释放问题 大于128页直接向堆释放内存小于等于128页则继续走Page Cache逻辑页合并 修改释放联调程序在这里能看出MapObjToSpan的价值通过地址就可以映射找到span并且为了获取存储对象大小在span结构中增添_objSize。 Page Cache大于128页向堆释放内存。 并且在申请和释放的对象的过程中我们使用了定长内存池创建释放对象不使用new和delete使得可以和malloc进行性能比较。
性能对比及基数树优化
性能对比 对比多线程下设计的高并发内存池和malloc的性能分别对相同大小内存和不同大小内存进行申请和释放。
#includeConcurrentAlloc.hvoid BenchmarkMalloc(size_t ntimes, size_t nworks, size_t rounds)
{std::vectorstd::thread vthread(nworks);std::atomicsize_t malloc_costtime 0;std::atomicsize_t free_costtime 0;for (size_t k 0; k nworks; k){vthread[k] std::thread([, k]() {std::vectorvoid* 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, (unsigned int)malloc_costtime);printf(%u个线程并发执行%u轮次每轮次free %u次: 花费%u ms\n,nworks, rounds, ntimes, (unsigned int)free_costtime);printf(%u个线程并发mallocfree %u次总计花费%u ms\n,nworks, nworks * rounds * ntimes, (unsigned int)(malloc_costtime free_costtime));
}void BenchmarkConcurrentMalloc(size_t ntimes, size_t nworks, size_t rounds)
{std::vectorstd::thread vthread(nworks);std::atomicsize_t malloc_costtime 0;std::atomicsize_t free_costtime 0;for (size_t k 0; k nworks; k){vthread[k] std::thread([]() {std::vectorvoid* 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, (unsigned int)malloc_costtime);printf(%u个线程并发执行%u轮次每轮次concurrent dealloc %u次: 花费%u ms\n,nworks, rounds, ntimes, (unsigned int)free_costtime);printf(%u个线程并发concurrent allocdealloc %u次总计花费%u ms\n,nworks, nworks * rounds * ntimes, (unsigned int)(malloc_costtime free_costtime));
}int main()
{size_t n 1000;cout endl;BenchmarkConcurrentMalloc(n, 4, 10);cout endl endl;BenchmarkMalloc(n, 4, 10);cout endl;return 0;
}ntimes单轮申请、释放次数nworks线程数rounds轮次数线程内部使用lambda表达式C11新特性用于定义匿名函数以值传递捕获k以引用传递捕获其他父作用域的变量使用原子变量atomicC11新特性不会导致多线程下数据竞争注意printf没法直接大于atomic类型对象需要强转。
测试结果性能有待优化 性能瓶颈分析
我们使用VS自带的性能探查器进行时间检测。 根据检测结果我们发现性能瓶颈点在MapObjToSpan的锁竞争上。 基数树优化 在tcmalloc中实际上在释放内存中对该处使用了基数树优化那我们也学习使用基数树对我们的程序进行优化。 单层基数树是直接地址映射法进行直接哈希也就是说页号与span直接对应。
// 一层基数树直接哈希
template int BITS
class TCMalloc_PageMap1 {
private:static const int LENGTH 1 BITS;//页数目BITS是存储页号需要多少位假设一页8K2^1332位下存储页号需要(32-13)19位void** array_;//指针数组
public:typedef uintptr_t Number;explicit TCMalloc_PageMap1( ) {size_t bytes sizeof(void*) BITS;//需要开辟的字节数size_t alignSize AMSize::_RoundUp(bytes, 1 PAGE_SHIFT);//bytes2^18256*1024按页大小对齐array_ (void**)SystemAlloc(alignSize PAGE_SHIFT);//按页分配内存memset(array_, 0, sizeof(void*) BITS);}//返回映射值void* get(Number k) const {if ((k BITS) 0) {//页号不在页数目范围return NULL;}return array_[k];}//建立映射void set(Number k, void* v) {array_[k] v;}
}; 非类型模板参数BITS表示存储页号最多需要比特位的个数32位下最大页号2^19次此时BITS就是19数组个数就是2^19每个存储1个指针所以数组总大小2^212M。 64位下最大页号2^51次此时BITS就是51数组个数就是2^19每个存储1个指针所以数字总大小2^542^24G这实在是太大了所以我们需要继续分层。 二层基数树实际上就是把BITS进行分层映射在32位下用前5比特位映射第一层得到2^5个后14位映射到第二层得到该页的span指针。总共占用大小2^5 * 2^14 * 4 2^212M。和一层基数树开辟的大小是一样的但是二层基数树最开始只需要开辟第一层当需要某一页号进行映射再开辟第二层而一层基数树一开始直接开辟全部。
//二层基数树分层哈希
template int BITS
class TCMalloc_PageMap2 {
private:static const int ROOT_BITS 5;//前5个比特位static const int ROOT_LENGTH 1 ROOT_BITS;//2^5第一层存储元素个数static const int LEAF_BITS BITS - ROOT_BITS;//19-514,剩下14个比特位static const int LEAF_LENGTH 1 LEAF_BITS;//2^14第二层存储元素个数// Leaf nodestruct Leaf {void* values[LEAF_LENGTH];};Leaf* root_[ROOT_LENGTH];typedef uintptr_t Number;explicit TCMalloc_PageMap2( ) { memset(root_, 0, sizeof(root_));//第一层空间清理PreallocateMoreMemory();}void* get(Number k) const {const Number i1 k LEAF_BITS;//k低19位存储页号合法的高位都是0右移14位获取19位中的前5位[18,14]确定第一层的下标const Number i2 k (LEAF_LENGTH - 1);//获取后13位与k与运算获得第二层的下标if ((k BITS) 0 || root_[i1] NULL)// 页号值超过范围或者页号映射的空间未开辟{return NULL;}return root_[i1]-values[i2];//返回映射的span指针}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)//第一层i1指向的空间未开辟 {static ObjectPoolLeafLeafPool;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);}
}; 设计Ensure函数进行需要页号时再开辟第二层空间并且全部开辟内存消耗也不多所以我们在构造的时候就全部开辟出来。 32位可以使用一层和二层基数树64位下需要使用三层基数树分析过程和二层实际一样省略。 本项目只在32位平台使用基数数优化我们使用单层基数树优化代码 当我们需要建立映射关系时就调用基数树函数set
_idSpanMap.set(span-_pageId, span);
当我们需要读取映射关系时就调用基数树函数get
auto ret (Span*)_idSpanMap.get(id);
MapObjToSpan函数此时无需加锁
//通过页的起始地址找到页从而映射找到span
Span* PageCache::MapObjToSpan(void* obj)
{//算页号PAGE_ID id (PAGE_ID)obj PAGE_SHIFT;auto ret (Span*)_idSpanMap.get(id);assert(ret ! nullptr);return ret;
}
为什么无需加锁 MapObjToSpan在进行读操作。 1.只有这两个函数中会去建立id和span的映射也就是说会去写操作 2.基数树写之前会提取开好空间写数据过程中不会动结构。 3.读写是分离的。线程1对这个位置读写操作时线程2不可能对这个位置进行读写操作。 我们不会同时对同一个页进行读取映射和建立映射的操作因为我们只有在释放对象时才需要读取映射但是在这个位置地方进行读操作也绝不会进行写操作因为我们在开始开辟这个位置的时候就已经写操作写好映射了而建立映射的写操作都是在page cache进行的页缓存中我们加了一把大锁更不可能出现写操作的竞争也不可能2个线程对同一个位置进行读操作因为读操作是在释放对象过程中这期间有桶锁所以也不可能产生竞争。 再次性能测试优化结果多线程场景下性能比malloc好。 本项目最终性能优化后只实现了在32位下运行如若64位下则不应使用基数树优化。
源码
https://gitee.com/hao-welcome/ConcurrentMemoryPool
