thread cache模拟设计实现

1 介绍thread cacheThreadCache是每一个线程自己独占的内存缓冲区。一个线程对应一个 Thread‑Cache别的线程访问不到只负责分配小于 256KB 的小对象最大优势线程申请和释放内存全程不用加锁性能极高concurrent memory pool 主要由下图的3个部分构成其中thread cache就是每个线程独立拥有Central Cache 全局公共仓库所有线程共享PageCache 页缓存全局唯一为了方便后续链接三部分我这里简单介绍一下各部分后续具体写完三个部分我会出一期详细的文章将三者串起来。2.实现thread cache2.1相关框架的实现首先可以建立俩个文件threadCache.h用于声明再建立threadCache.cpp用于定义,然后这里为了代码的完整性我将整体项目普遍需要的代码放到comm.h的头文件下边。这里理解一下thread cache的内部结构就是分为若干个桶每个桶是一个桶位置映射大小的内存块对象的自由链表。下边的代码只是方便理解记忆不一定可以跑起来。//common.h //就是包含这个项目普遍需要的内容 #includeiostream #includeassert using std::cout; using std::endl; #includevector #includetime.h //为了增减代码的可读性 设立这个函数来封装*(void**)//这个是啥意思上个文章里边讲过 static void* nextobj(void* obj)//加的原因是返回值可以修改不只是可读 { return *(void**)obj; } //创建一个链表 class FreeList { public: //链表上边增加和删除 void push(void* obj)//就是相等于释放内存 { assert(obj); //头插 nextobj(obj) _freeList; _freeList obj; } void* pop()//就是相等于申请内存就是把需要的内存从threadCache中删掉 { //头删除 void* obj _freeList; _freeList nextobj(obj); return obj; } private: void* _freeList nullptr; };#threadCache.h #includecommon.h//就是大部分需要的文件头都放里边 class ThreadCache { public: //申请和释放内存 void* Allocate(size_t size);//申请内存其实就是从threadcache中删东西出去 void Deallocate(void* ptr, size_t size);//释放内存就是加东西到threadcache中 private: FreeList freeList[];//这里是threadcache内存的结构 开辟的大小是桶的个数 这个后边再详细讲 };//threadCache.cpp #include common.h #include threadCache.h //申请内存 void* ThreadCache::Allocate(size_t size)//size就是用户申请的内存字节数 { assert(size 256 * 1024);//因为只有申请内存小于等于256kb才可以走threadCache //... } void ThreadCache::Deallocate(void* ptr, size_t size) { assert(ptr); assert(size 256 * 1024); //... }注意1.我为了提高代码的可读性封装了*(void**)记得前边加static因为是全局函数有风险所以加 static 限制仅当前头文件可见2.threadCache.h文件中的FreeList freeList[]需要定义大小不然跑不通后续讲具体需要多少个桶的时候再完善2.2threadCache类函数的完善前文已经介绍 ThreadCache 内部结构接下来我们需要确定 ThreadCache 划分的桶数量并明确每个桶对应的内存规格与存取逻辑。下边我将讲俩个细节问题来感受thread cache内部的内存是如何分配的问题一如何确定自由链表节点的标准内存大小我们简单设想全部统一 8 字节对齐会暴露出一个严重问题最大 256KB如果全程 8 字节一档数组需要 32768 个桶占用大量内存。TCMalloc 采用分段梯度对齐方案在保证内存内碎片浪费不超过 10%的前提下小块内存细粒度对齐大块内存放大对齐步长大幅减少桶的总数量。就按照上图的方式分梯度对齐拿具体数字举个例子申请1030字节区间(1024,8KB]步长 128 向上对齐找到≥1030 最小的 128 倍数 1024 是 128×8下一档 128×9 1152 字节浪费 1152 - 1030 122 浪费比例122 / 1152 ≈ 10.59%大概就是10%左右所以减少了内碎片的浪费。问题二如何确定申请内存对应的桶下标如何利用对齐后的块大小换算得到自由链表数组对应的桶下标从而定位到对应的空闲内存桶。大概的思路就是我们提前写一个数组内容是每一个区间对应的有几个桶分别是16565656最后一个区间桶式总数用不上。然后区分 size 所属区间传入对应对齐移位值调用_Index子函数通过位运算快速完成下标计算叠加前面所有区间桶的总数量也就是提前写好的数组得到全局唯一数组下标。我们这里直接举例子申请1030字节对齐结果 1152 字节这样我们确定区间是(1024,8*1024],1030-1024/ 128 0所以对应的下标就是16 0 16。俩个问题讲清楚了接下来我们就是代码实现我们需要对这俩个进行封装为sizeClass放在common.h里。还有个点就是我们现在确定thread cache中桶的个数了所以单独定义为全局静态变量。//common.h //就是包含这个项目普遍需要的内容 #includeiostream #includeassert using std::cout; using std::endl; #includevector #includetime.h static const size_t MAX_BYTES 256 * 1024;//定义一个最大值 因为如果内存申请小于256kb就是先获取thread cache对象 static const size_t NFREELIST 208;//这里是threadCache中桶的个数 通过按照计算得知 //为了增减代码的可读性 设立这个函数来封装*(void**)//这个是啥意思上个文章里边讲过 static void* nextobj(void* obj)//加的原因是返回值可以修改不只是可读 { return *(void**)obj; } //创建一个链表 class FreeList { public: //链表上边增加和删除 void push(void* obj)//就是相等于释放内存 { assert(obj); //头插 nextobj(obj) _freeList; _freeList obj; } void* pop()//就是相等于申请内存就是把需要的内存从threadCache中删掉 { //头删除 void* obj _freeList; _freeList nextobj(obj); return obj; } private: void* _freeList nullptr; }; class sizeClass { public: //首先实现问题一 //static inline size_t _RoundUp(size_t size, size_t alignNum)//size代表着申请内存的大小 alignNum代表着对其基数 //{ // size_t alignSize; // if (size % alignNum ! 0) // alignSize (size / alignNum 1) * alignNum; // else // alignSize size;//就是刚好等于分界点 // return alignSize; //} //利用二进制方式实现 static inline size_t _RoundUp(size_t size, size_t alignNum) { return ((size 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 8 * 1024) { return _RoundUp(size, 128); } else if (size 64 * 1024) { return _RoundUp(size, 1024); } else if (size 256 * 1024) { return _RoundUp(size, 8 * 1024); } else { assert(false); return -1; } } //实现问题二 //计算映射的哪一个自由链表桶从零开始的一共208个 /*size_t _Index(size_t size, size_t alignNum) { if (size % alignNum 0) { return size / alignNum - 1; } else { return size / alignNum; } }*/ //利用二进制实现 static inline size_t _Index(size_t size, size_t align_shift)//aline_shift 对应的就是2的n次方 { return ((size (1 align_shift) - 1) align_shift) - 1;//这里的位移相当于乘除 } static inline size_t Index(size_t size) { // 每个区间有多少个链 static int group_array[4] { 16, 56, 56, 56 };//每一种分段对应的桶的个数 if (size 128) { return _Index(size, 3); } else if (size 1024) { return _Index(size - 128, 4) group_array[0]; } else if (size 8 * 1024) { return _Index(size - 1024, 7) group_array[1] group_array[0]; } else if (size 64 * 1024) { return _Index(size - 8 * 1024, 10) group_array[2] group_array[1] group_array[0]; } else if (size 256 * 1024) { return _Index(size - 64 * 1024, 13) group_array[3] group_array[2] group_array[1] group_array[0]; } else { assert(false); } return -1; } };当我们封装好上边这俩个函数的时候就可以来实现申请和释放内存了这里没什么理解的看代码就能明白直接上代码。//threadCache.cpp #include common.h #include threadCache.h void* threadCache::FetchFromCentralCache(size_t index, size_t size)//这里不是重点 只是为了让代码跑起来 { // ... return nullptr; } //申请内存 void* ThreadCache::Allocate(size_t size)//size就是用户申请的内存字节数 { assert(size MAX_BYTES);//因为只有申请内存小于等于256kb才可以走threadCache size_t alignSize sizeClass::GroudUp(size); size_t index sizeClass::Index(size); if(!_freeList[index].Empty()) // 如果这个桶的链表中还有空节点就可以直接申请 return _freeList[index].pop; else // 向central cache 里边申请 FetchFromCentralCache(index, size); } //释放内存 void ThreadCache::Deallocate(void* ptr, size_t size) { assert(ptr); assert(size MAX_BYTES); size_t index sizeClass::Index(size); _freeList[index].push(ptr); }上边实现申请和释放内存的时候新增了一个判断链表是否为空的函数还有到central cache里边申请内存的函数接下来完善这俩个的声明和定义#threadCache.h #includecommon.h//就是大部分需要的文件头都放里边 class ThreadCache { public: //申请和释放内存 void* Allocate(size_t size);//申请内存其实就是从threadcache中删东西出去 void Deallocate(void* ptr, size_t size);//释放内存就是加东西到threadcache中 // 从中心缓存获取对象 void* FetchFromCentralCache(size_t index, size_t size); private: FreeList freeList[];//这里是threadcache内存的结构 开辟的大小是桶的个数 这个后边再详细讲 };//common.h //就是包含这个项目普遍需要的内容 #includeiostream #includeassert using std::cout; using std::endl; #includevector #includetime.h static const size_t MAX_BYTES 256 * 1024;//定义一个最大值 因为如果内存申请小于256kb就是先获取thread cache对象 static const size_t NFREELIST 208;//这里是threadCache中桶的个数 通过按照计算得知 //为了增减代码的可读性 设立这个函数来封装*(void**)//这个是啥意思上个文章里边讲过 static void* nextobj(void* obj)//加的原因是返回值可以修改不只是可读 { return *(void**)obj; } //创建一个链表 class FreeList { public: //链表上边增加和删除 void push(void* obj)//就是相等于释放内存 { assert(obj); //头插 nextobj(obj) _freeList; _freeList obj; } void* pop()//就是相等于申请内存就是把需要的内存从threadCache中删掉 { //头删除 void* obj _freeList; _freeList nextobj(obj); return obj; } bool Empty() { return _freeList nullpty; } private: void* _freeList nullptr; }; class sizeClass { public: //首先实现问题一 //static inline size_t _RoundUp(size_t size, size_t alignNum)//size代表着申请内存的大小 alignNum代表着对其基数 //{ // size_t alignSize; // if (size % alignNum ! 0) // alignSize (size / alignNum 1) * alignNum; // else // alignSize size;//就是刚好等于分界点 // return alignSize; //} //利用二进制方式实现 static inline size_t _RoundUp(size_t size, size_t alignNum) { return ((size 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 8 * 1024) { return _RoundUp(size, 128); } else if (size 64 * 1024) { return _RoundUp(size, 1024); } else if (size 256 * 1024) { return _RoundUp(size, 8 * 1024); } else { assert(false); return -1; } } //实现问题二 //计算映射的哪一个自由链表桶从零开始的一共208个 /*size_t _Index(size_t size, size_t alignNum) { if (size % alignNum 0) { return size / alignNum - 1; } else { return size / alignNum; } }*/ //利用二进制实现 static inline size_t _Index(size_t size, size_t align_shift)//aline_shift 对应的就是2的n次方 { return ((size (1 align_shift) - 1) align_shift) - 1;//这里的位移相当于乘除 } static inline size_t Index(size_t size) { // 每个区间有多少个链 static int group_array[4] { 16, 56, 56, 56 };//每一种分段对应的桶的个数 if (size 128) { return _Index(size, 3); } else if (size 1024) { return _Index(size - 128, 4) group_array[0]; } else if (size 8 * 1024) { return _Index(size - 1024, 7) group_array[1] group_array[0]; } else if (size 64 * 1024) { return _Index(size - 8 * 1024, 10) group_array[2] group_array[1] group_array[0]; } else if (size 256 * 1024) { return _Index(size - 64 * 1024, 13) group_array[3] group_array[2] group_array[1] group_array[0]; } else { assert(false); } return -1; } };2.3实现TLS无锁访问使用__declspec(thread)TLS 线程本地存储让每个线程拥有独立的 ThreadCache 指针线程只操作自己专属的 ThreadCache 对象 天然不存在多线程竞争ThreadCache 分配 / 释放全程不需要加锁这也是 TCMalloc 高性能的核心关键点之一。2.3.1定义TLS变量//threadCache.h // TLS Thread Local Storage 线程本地存储 static __declspec(thread) ThreadCache* pTLSThreadCache nullptr;__declspec(thread)修饰的变量特点 进程内每一个线程都会独立拥有一份该变量副本。 线程 A 修改pTLSThreadCache不会影响线程 B 的pTLSThreadCache线程之间完全隔离。所以为了实现TLS并且还是申请内存我们继续做一个上层并发分配接口封装创建一个文件concurrentAlloc.h在这里进行封装//ConcurrentAlloc.h #includecommon.h #includethreadCache.h void* concurrentAlloc(size_T size) { if(pTLSThreadCache nullptr) pTLSThreadCache new ThreadCache; return pTLSThreadCache-Allocate(size); } void concurrentFree(void* ptr, size_t size) { assert(ptr); pTLSThreadCache-Deallocate(ptr, size); }到这我们就基本写好了下边我来写一个测试来测试一下我们上边写的代码//unitTest.cpp void Alloc1()//线程1 { for (size_t i 0; i 5; i) { void* ptr concurrentAlloc(6); } } void Alloc2()//线程2 { for (size_t i 0; i 5; i) { void* ptr concurrentAlloc(7); } } void TLSTest() { //串行执行两个线程 std::thread t1(Alloc1); std::thread t2(Alloc2); t1.join(); t2.join(); }线程 1 第一次执行concurrentAllocpTLSThreadCache nullptrnew 一个专属ThreadCache后续复用。线程 2 拥有独立副本的pTLSThreadCache不受线程 1 影响同样新建属于自己的ThreadCache。并行的执行俩个线程这里需要包含头文件thread最后在main函数中调用执行就欧克。3.全部代码展示3.1common.h//common.h //就是包含这个项目普遍需要的内容 #includeiostream #includeassert using std::cout; using std::endl; #includevector #includetime.h #includethread static const size_t MAX_BYTES 256 * 1024;//定义一个最大值 因为如果内存申请小于256kb就是先获取thread cache对象 static const size_t NFREELIST 208;//这里是threadCache中桶的个数 通过按照计算得知 //为了增减代码的可读性 设立这个函数来封装*(void**)//这个是啥意思上个文章里边讲过 static void* nextobj(void* obj)//加的原因是返回值可以修改不只是可读 { return *(void**)obj; } //创建一个链表 class FreeList { public: //链表上边增加和删除 void push(void* obj)//就是相等于释放内存 { assert(obj); //头插 nextobj(obj) _freeList; _freeList obj; } void* pop()//就是相等于申请内存就是把需要的内存从threadCache中删掉 { //头删除 void* obj _freeList; _freeList nextobj(obj); return obj; } bool Empty() { return _freeList nullpty; } private: void* _freeList nullptr; }; class sizeClass { public: //首先实现问题一 //static inline size_t _RoundUp(size_t size, size_t alignNum)//size代表着申请内存的大小 alignNum代表着对其基数 //{ // size_t alignSize; // if (size % alignNum ! 0) // alignSize (size / alignNum 1) * alignNum; // else // alignSize size;//就是刚好等于分界点 // return alignSize; //} //利用二进制方式实现 static inline size_t _RoundUp(size_t size, size_t alignNum) { return ((size 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 8 * 1024) { return _RoundUp(size, 128); } else if (size 64 * 1024) { return _RoundUp(size, 1024); } else if (size 256 * 1024) { return _RoundUp(size, 8 * 1024); } else { assert(false); return -1; } } //实现问题二 //计算映射的哪一个自由链表桶从零开始的一共208个 /*size_t _Index(size_t size, size_t alignNum) { if (size % alignNum 0) { return size / alignNum - 1; } else { return size / alignNum; } }*/ //利用二进制实现 static inline size_t _Index(size_t size, size_t align_shift)//aline_shift 对应的就是2的n次方 { return ((size (1 align_shift) - 1) align_shift) - 1;//这里的位移相当于乘除 } static inline size_t Index(size_t size) { // 每个区间有多少个链 static int group_array[4] { 16, 56, 56, 56 };//每一种分段对应的桶的个数 if (size 128) { return _Index(size, 3); } else if (size 1024) { return _Index(size - 128, 4) group_array[0]; } else if (size 8 * 1024) { return _Index(size - 1024, 7) group_array[1] group_array[0]; } else if (size 64 * 1024) { return _Index(size - 8 * 1024, 10) group_array[2] group_array[1] group_array[0]; } else if (size 256 * 1024) { return _Index(size - 64 * 1024, 13) group_array[3] group_array[2] group_array[1] group_array[0]; } else { assert(false); } return -1; } };3.2threadCache.h#threadCache.h #includecommon.h//就是大部分需要的文件头都放里边 class ThreadCache { public: //申请和释放内存 void* Allocate(size_t size);//申请内存其实就是从threadcache中删东西出去 void Deallocate(void* ptr, size_t size);//释放内存就是加东西到threadcache中 // 从中心缓存获取对象 void* FetchFromCentralCache(size_t index, size_t size); private: FreeList freeList[NFREELIST];//这里是threadcache内存的结构 开辟的大小是桶的个数 这个后边再详细讲 };3.3threadCache.cpp//threadCache.cpp #include common.h #include threadCache.h void* threadCache::FetchFromCentralCache(size_t index, size_t size)//这里不是重点 只是为了让代码跑起来 { // ... return nullptr; } //申请内存 void* ThreadCache::Allocate(size_t size)//size就是用户申请的内存字节数 { assert(size MAX_BYTES);//因为只有申请内存小于等于256kb才可以走threadCache size_t alignSize sizeClass::GroudUp(size); size_t index sizeClass::Index(size); if(!_freeList[index].Empty()) // 如果这个桶的链表中还有空节点就可以直接申请 return _freeList[index].pop; else // 向central cache 里边申请 FetchFromCentralCache(index, size); } //释放内存 void ThreadCache::Deallocate(void* ptr, size_t size) { assert(ptr); assert(size MAX_BYTES); size_t index sizeClass::Index(size); _freeList[index].push(ptr); }3.4ConcurrentAlloc.h//ConcurrentAlloc.h #includecommon.h #includethreadCache.h // TLS thread local storage 线程本地存储注释 static __declspec(thread) ThreadCache* pTLSThreadCache nullptr; void* concurrentAlloc(size_T size) { if(pTLSThreadCache nullptr) pTLSThreadCache new ThreadCache; return pTLSThreadCache-Allocate(size); } void concurrentFree(void* ptr, size_t size) { assert(ptr); pTLSThreadCache-Deallocate(ptr, size); }3.5unitTest.cpp#includeObjectpool.h #includecommon.h //测试threadCache需要的函数 #includeConcurrentAlloc.h void Alloc1()//线程1 { for (size_t i 0; i 5; i) { void* ptr concurrentAlloc(6); } } void Alloc2()//线程2 { for (size_t i 0; i 5; i) { void* ptr concurrentAlloc(7); } } void TLSTest() { //并行执行 std::thread t1(Alloc1); std::thread t2(Alloc2); t1.join(); t2.join(); } int main() { TLSTest(); }