thread cache模拟设计实现-尧图网络科技

1 介绍thread cache

ThreadCache是每一个线程自己独占的内存缓冲区。

一个线程对应一个 Thread‑Cache，别的线程访问不到；
只负责分配小于 256KB 的小对象；
最大优势：线程申请和释放内存全程不用加锁，性能极高

concurrent memory pool 主要由下图的3个部分构成，其中thread cache就是每个线程独立拥有Central Cache 全局公共仓库，所有线程共享，PageCache 页缓存，全局唯一

为了方便后续链接三部分，我这里简单介绍一下各部分，后续具体写完三个部分我会出一期详细的文章将三者串起来。

2.实现thread cache

2.1相关框架的实现

首先可以建立俩个文件threadCache.h用于声明，再建立threadCache.cpp用于定义,然后这里为了代码的完整性，我将整体项目普遍需要的代码放到comm.h的头文件下边。

这里理解一下thread cache的内部结构，就是分为若干个桶，每个桶是一个桶位置映射大小的内存块对象的自由链表。

下边的代码只是方便理解记忆不一定可以跑起来。

//common.h //就是包含这个项目普遍需要的内容 #include<iostream> #include<assert> using std::cout; using std::endl; #include<vector> #include<time.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 #include"common.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%左右，所以减少了内碎片的浪费。

问题二：如何确定申请内存对应的桶下标？

如何利用对齐后的块大小，换算得到自由链表数组对应的桶下标，从而定位到对应的空闲内存桶。

大概的思路就是我们提前写一个数组，内容是每一个区间对应的有几个桶，分别是16，56，56，56，最后一个区间桶式总数用不上。然后区分 size 所属区间，传入对应对齐移位值，调用_Index子函数，通过位运算快速完成下标计算，叠加前面所有区间桶的总数量（也就是提前写好的数组），得到全局唯一数组下标。

我们这里直接举例子：申请1030字节，对齐结果 1152 字节，这样我们确定区间是(1024,8*1024],（1030-1024）/ 128 = 0，所以对应的下标就是16 + 0 = 16。

俩个问题讲清楚了接下来我们就是代码实现，我们需要对这俩个进行封装，为sizeClass，放在common.h里。还有个点就是我们现在确定thread cache中桶的个数了，所以单独定义为全局静态变量。

//common.h //就是包含这个项目普遍需要的内容 #include<iostream> #include<assert> using std::cout; using std::endl; #include<vector> #include<time.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 #include"common.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 //就是包含这个项目普遍需要的内容 #include<iostream> #include<assert> using std::cout; using std::endl; #include<vector> #include<time.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 #include"common.h" #include"threadCache.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 第一次执行concurrentAlloc，pTLSThreadCache == nullptr，new 一个专属ThreadCache；后续复用。

线程 2 拥有独立副本的pTLSThreadCache，不受线程 1 影响，同样新建属于自己的ThreadCache。

并行的执行俩个线程，这里需要包含头文件<thread>

最后在main函数中调用执行就欧克。

3.全部代码展示

3.1common.h

//common.h //就是包含这个项目普遍需要的内容 #include<iostream> #include<assert> using std::cout; using std::endl; #include<vector> #include<time.h> #include<thread> 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 #include"common.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 #include"common.h" #include"threadCache.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

#include"Objectpool.h" #include"common.h" //测试threadCache需要的函数 #include"ConcurrentAlloc.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(); }