高并发内存池 - thread cache

高并发内存池 - thread cache我们首先对高并发内存池中最底层的框架 thread cache 进行解析在定长内存池的实现中针对一种固定大小的变量在申请内存时效率非常好所以 thread cache 的设计思路大致延续了定长内存池的设计思路不过在此之上我们要解决定长内存池的痛点不同大小的类型申请空间对于这个问题先来分析一下在定长内存池中分配内存的关键角色就是自由链表通过自由链表可以快速地分配内存不用直接从申请的大块空间中切割但也正是自由链表导致定长内存池需要固定申请变量的大小所以解决方法就从这里入手增加对应不同字节数变量的自由链表如果申请 8 字节数有一个对应的链表申请 6 字节数的有一个链表那么这个问题就会得到解决但是解决一个问题往往会带来新的问题新的问题就是难道我们给 256KB 内的所有字节数都设计一个自由链表么那自由链表的数量将会来到可怕的256×1024262144256 \times 1024 262144256×1024262144所以这是不现实的对于这个问题我们采取折中方案将自由链表设计的有跨度将第一个自由链表对应的变量大小为 8 字节第二个为 16 字节 … 依次递增这样自由链表的数量就不会如此恐怖而且采用哈希桶的设计方法来实现索引效率也会非常高但无奈之举就是会增加内碎片可能有只需要 10 字节的变量分给了 16 字节这是这个解决方法无法避免的问题1.现阶段代码实现1.1 文件结构Comm.h 包含一些常用头文件、以及一些工具类ObjectPool.h 之前实现的定长内存池ThreadCache.xxx 实现 thread cache1.2 ThreadCache 初步剖析对于一个内存池的底层来说必须要具备两个接口申请空间归还空间void*Allocate(size_t size);//根据变量字节数大小申请空间voidDeallocate(void*ptr,size_t size);//归还空间由于会有不同的大小的变量申请空间所以传入变量的字节数大小是必要的从这里我们便可窥见malloc的设计中为什么也需传入变量大小对于ThreadCache来说实现内存的具体操作并不在这一层而是在自由链表中对于自由链表我们还要创建多个所以我们决定将其设计为一个类并放在Comm.h中在ThreadCache这层中还需对自由链表进行管理所以我们需要一个管理自由链表的数组FreeList _freeList[]这里共需多少个自由链表我们先按下不表后面会讲到1.3 自由链表初步剖析对于自由链表来说要有一个头结点要实现的接口目前有以下两个回收空间分配空间所以整个类就可以被我们如此设计//返回节点中指向下一节点地址的 _nextvoid*ObjNext(void*ptr){return*(void**)ptr;}//用于管理一个变量大小的自由链表classFreeList{public:voidPush(void*ptr){ObjNext(ptr)_freelist;_freelistptr;}void*Pop(){void*obj_freelist;_freelistObjNext(_freelist);returnobj;}private:void*_freelist;};这里讲述一下 *(void**) 几个思考要点对int*解引用会得到int变量所以对二级指针解引用会得到一级指针变量在ObjNext函数的返回值处一定要添加引用否则返回值属性为右值不能进行赋值即使这串地址变量确实指向一处空间1.3.1 自由链表对齐规则设计剖析首先我们要指定一种对齐规则首先排除 4 字节对齐因为在 64位 环境下指针变量大小为 8 字节所以我们采用 8 字节对齐方案其次如果全部采用 8 字节对齐方案256KB 会产生256×1024÷832768256 \times 1024 \div 8 32768256×1024÷832768个自由链表还是有些多了通过观察和简化tcmalloc的对齐我们采用以下方案字节大小在 [1, 128] 8bytes 对齐 freelist [0, 16)字节大小在 [128 1, 1024] 16bytes 对齐 freelist [16, 72)字节大小在 [1024 1, 8 * 1024] 128bytes 对齐 freelist [72, 128)字节大小在 [8 * 1024 1, 64 * 1024] 1024bytes 对齐 freelist [128, 184)字节大小在 [64 * 1024 1, 256 * 1024] 8 * 1024bytes 对齐 freelist [184, 208)通过这种方法我们将自由链表的个数压缩到了 208 个远远少于 32768其次空间浪费率尽量控制在 10%比如申请字节数为 129对齐到 144 字节浪费率为15÷14410.4%15 \div 144 10.4\%15÷14410.4%申请字节数为 64 * 1024 1 字节是对齐到 73728 字节浪费率为8191÷7372811.1%8191 \div 73728 11.1\%8191÷7372811.1%所以当申请大空间时浪费也会被控制。所以上层 ThreadCache 管理自由链表数组的个数就可以填上了现在在对齐策略制定后我们便可以开始着手代码设计了对齐功能应被设计成为一个独立的类SizeClass且最终被放入Comm.h头文件中classSizeClass{public:// 整体控制在最多10%左右的内碎⽚浪费// [1,128] 8 byte对⻬ freelist[0,16)// [1281,1024] 16 byte对⻬ freelist[16,72)// [10241,8*1024] 128 byte对⻬ freelist[72,128)// [8*10241,64*1024] 1024 byte对⻬ freelist[128,184)// [64*10241,256*1024] 8*1024 byte对⻬ freelist[184,208)staticinlinesize_t_RoundUp(size_t bytes,size_t align){return(((bytes)align-1)%~(align-1));}//根据字节大小返回对齐之后的结果staticinlinesize_tRoundUp(size_t bytes){if(bytes128){return_RoundUp(bytes,8);}elseif(bytes1024){return_RoundUp(bytes,16);}elseif(bytes8*1024){return_RoundUp(bytes,128);}elseif(bytes64*1024){return_RoundUp(bytes,1024);}elseif(bytes256*1024){return_RoundUp(bytes,8*1024);}else//超出 256KB 的情况{}return-1;//仅仅是为了所有路径均有返回值}由于是一个工具类将其设计为了 static inline 的模式这样就可以不用创建对象进行调用对于RoundUp函数的设计呢变量的字节数大小是一定要传入的要依此进行对齐操作在函数内部根据划分的区域进行对应的if else操作即可这没什么好说的有意思的是运算出对齐数的操作staticinlinesize_t_RoundUp(size_t bytes,size_t align){return(((bytes)align-1)%~(align-1));}这里将变量大小和对齐数都传入了其中的操作值得我们细细品味首先先对(bytes) align - 1先进行解剖加上一个和为对齐数 - 1的数这里作者认为是如若变量大小不能整除那么加上了这么一个数那么一定会经过一个比变量本身大能整除对齐数的数如果可以整除那么也不会跳到下一个对齐数其次进行了位操作 ~(align - 1):这里为了更能形象解释我们将align设为 8align - 1就为 7二进制为0111取反就成为了一个除了第三位其余位全为 1 的数然后在于我们第一步得到数进行按位与操作这里的目的很明显消除低三位将第三位全都变为 0 这里必须补充一个知识如若一个数的低 n 位为 0那么它一定能被2n2^n2n整除有了这个知识我们就知道得出来的数字最终一定能被对齐数整除而且一定大于变量大小因为加了对齐数 - 1且这个数一定是离变量大小最近的对齐数因为就加了一次所以这个算法非常巧妙。1.3.2 自由链表索引规则剖析在探明对齐规则后就应该对索引规则进行设计我们采用哈希桶的设计思路根据变量字节数的不同大小直接索引到对应的桶所以便可以开始着手代码设计代码设计在SizeClass中staticinlinesize_t_Index(size_t bytes,size_t align_shift){return(((bytes)(1align_shift)-1)align_shift)-1;}staticinlinesize_tIndex(size_t bytes){//断言申请的空间是否大于 256 KBassert(bytesMAX_BYTES);//每个对齐数自由链表的个数staticintgroup_array[4]{16,56,56,56};if(bytes128){return_Index(bytes,3);}elseif(bytes1024){return_Index(bytes-128,4)group_array[0];}elseif(bytes8*1024){return_Index(bytes-1024,7)group_array[0]group_array[1];}elseif(bytes64*1024){return_Index(bytes-8*1024,10)group_array[0]group_array[1]group_array[2];}elseif(bytes256*1024){return_Index(bytes-64*1024,13)group_array[0]group_array[1]group_array[2]group_array[3];}else{assert(false);}return-1;}对于 Index 函数设计思路是比较简单的必须传入的参数有变量大小根据不同的分区大小返回对应的自由链表下标但在这里我们可以做一些简化根据不同分区我们是知道分区对应的自由链表个数的所以创建了一个数组用来记录不同分区的freelist的个数用以简便计算static int group_array[4] {16, 56, 56, 56};所以这意味着我们只用计算出自由链表是在每个分区中的第几个然后在加上在自己前面分区的自由链表个数即可这也是传参并未传入变量的总体大小而是变量减去分区起始边界大小的原因简便了计算有意思的同样为我们如何计算出对应的下标以及为什么返回值要如此设计staticinlinesize_t_Index(size_t bytes,size_t align_shift){return(((bytes)(1align_shift))align_shift)-1;}这里的bytes和_RoundUp函数中的bytes就有所不同的传入的是减去分区开始大小后所剩余的字节数align_shift则是对齐数是 2 的几次幂(bytes) (1 align_shift) 的意义:通过位操作知晓是将bytes加上一个对齐数可以理解为将其跨过一个大于bytes且能整除对齐数的元素这么做是为了后面的位操作做铺垫(第一步结果 ailgn_shft) - 1 的意义将第一步得出得出的结果右移上align_shift本质就是除以align_shift对结果进行向下取整这里并不用担心向下取整操作会导致分配到不足变量大小的自由链表因为我们已经在第一步时加上了一个对齐数就导致一定会越过一个能整除变量大小且大于变量大小的对齐数这里其实就是在求这个对齐数是多少最后 -1是因为自由链表的下标从 0 开始这个算法同样非常巧妙使用了高效的位操作效率提高。1.3 完善分配空间函数将对齐策略和索引策略制定好后便可以开始着手完成ThreadCache的Allocate函数现阶段代码设计如下void*ThreadCache::Allocate(size_t size){assert(sizeMAX_BYTES);size_t alignsizeSizeClass::RoundUp(size);size_t indexSizeClass::Index(size);if(!_freeList[index].Empty()){return_freeList[index].Pop();}else{//从中心缓存获取数据returnFetchFromCentralCache(index,size);}}先对是否大于 256KB 进行断言确保不会申请过量的内存不过后期此逻辑仍需完善先暂且这么编写之后便要求得变量大小的对齐数和自由链表索引下标若自由链表不为空则直接返回分配空间若为空时我们则需向上层CentralCache进一步获取空间所以在ThreadCache中我们还需实现FetchFromCentralCache函数关于函数的参数为何如此设计我们先按下不表关于如何判断自由链表为空的逻辑也暂未实现一切的谜底都会在后文揭开