【项目日记(四)】thread cache 层

前言

前面我们对整个项目的框架进行了介绍，本期开始我们将进行第一层线程缓存层(thread cache)的详细介绍与实现。

目录

前言

一、thread cache 的整体设计

二、内存对齐规则和哈希映射关系

2.1 如何对齐？

2.2 这样设计对齐规则的好处？

2.3 内存对齐和哈希桶映射实现

三、TLS 无锁的访问

---

一、thread cache 的整体设计

thread cache 实际上是一个 哈希桶 的结构，每个桶都是按照桶位置进行映射的管理固定内存块的自由链表。每个线程都会有一个 thread cache  对象，因此线程 申请 和 释放 内存时无锁的！

因为每个线程最大可以申请 256 KB 的大小，也有可能申请 1bytes 的大小，如果我们都按一个字节进行划分内存对象话，那光管理这些内存块的自由链表就得 20 多万个，仅仅是存储自由链表的指针都是一个很大的开销，所以不能按字节划分！

我们可选择多一些平衡的牺牲，让这些字节数按照某种对齐规则进行向上对齐。例如：我们让这些都按照 8 字节对齐，那么申请[1-8]字节，都统一给8字节，申请[9-16]字节就给16字节。即thread cache 的结构就是下面这样的：

当前自由链表中挂的一些内存对象是经过 内存对齐 的也就是实际的空间是 大于等于 申请 的空间的，此时多余的那部分无法被利用。多余的那部分就成为内碎片

例如，按照 8 字节对齐，申请 1 字节，但是实际有 8 字节，多出的那7字节，就是内碎片。

也就是说，外碎片是空间总量够，但是开不出大块内存。内碎片是由于对齐而无法使用的那部分空间。

---

由于整个项目比较复杂所以我们先对自由链表进行封装，而当前的自由链表后面也会用到，所以我们放在 common.h 中

我们目前只有三个接口， 向自由链表添加内存对象 Push ，从自由链表中 拿出一个 Pop，判断自由链表是否为空 Empty 。后期的接口我们到时候再加。

// 获取 obj 对象的 头部的 4/8 字节
static inline void*& NextObj(void* obj)
{return *(void**)obj;
}class FreeList
{
public:void Push(void* obj){assert(obj);// 头插NextObj(obj) = _freelist;_freelist = obj;}void* Pop(){assert(_freelist);// 头删void* obj = _freelist;_freelist = NextObj(obj);return obj;}bool Empty(){return _freelist == nullptr;}private:void* _freelist = nullptr;
};

有了自由链表的封装和上面的介绍，我们就可以先将 thread cache 的整体框架搭建出来：

它里面的成员属性就一个 自由链表的数组（哈希桶）。至于这个数组有多大，得根据具体的对齐规则确定，暂时这用 N 代替

成员方法有三个：1、申请内存  2、释放内存 3、去 central cache 申请

class ThreadCache
{
public:// 申请和释放内存对象void* Allocate(size_t size);// size 代表 申请的大小 字节数void Deallocate(void* ptr, size_t size);// ptr 代表还回来内存的地址，size 表示大小// 从中心缓存获取对象void* FetchFromCentralCache(size_t index, size_t size); // index 表示 哈希桶的下标，size 表示申请的内存大小private:FreeList _freelists[N];// 哈希桶
};

当线程申请某一大小的内存时，会调用 Allocate 它内部根据某种对齐规则计算对齐后的大小，然后更具哈希桶的映射规则找到对应的自由链表，如果自由链表不为空，头删，返回一个内存对象即可，否则，调用FetchFromCentralCache 去central cache 中申请，然后返回。同理，释放的内存对象调用 Deallocate，它内部根据内存大小进行映射找到哈希桶中的对应的自由链表，然后进行头插即可！

我们本期主要实现申请和释放，从中心缓存申请我们要结合下一层实现，这个在下一层在实现！

而要实现这个两个接口，我们就得线确定是什么内存对齐规则了！

---

二、内存对齐规则和哈希映射关系

2.1 如何对齐？

上面说了，不能每个字节都对应一个自由链表，否则花销太大。因此我们需要指定一种合适的对齐规则。

首先，我们得保证每个内存块是可以链接到自由链表上的，因此，一开始 给 8 最合适，这样无论 32位 还是 64 位下都可以存的下一个指针。

在者我们也不能都按照 8 字节对齐，因为如果都按照 8 字节对齐，的话 我们需要建立 3万多个桶（256 * 1024 / 8 = 32768）这样也不行，其实我们可以分段，每一段按照不同的对其数进行对齐，具体如下：

2.2 这样设计对齐规则的好处？

这样设计，内碎片的浪费率可以控制在 10% 左右（浪费率 = 浪费的字节数 / 对齐后的字节数）！

首先第一组就不讨论了，因为你如果申请 1 字节，就算对齐数 是 2  浪费率也是 50 % 我们从第二组开始！

根据上面的公式，我们要得到某个区间的最大浪费率，就应该让分子取到最大，让分母取到最小。比如129~1024这个区间，该区域的对齐数是16，那么最大浪费的字节数就是15，而最小对齐后的字节数就是这个区间内的前16个数所对齐到的字节数，也就是144，那么该区间的最大浪费率也就是15 ÷ 144 ≈ 10.42 %  同样的道理，后面两个区间的最大浪费率分别是127 ÷ 1152 ≈ 11.02 % 和1023 ÷ 9216 ≈ 11.10 %

2.3 内存对齐和哈希桶映射实现

有了上述的对齐规则，我们需要实现两个函数，分别获取某一字节数进行对齐后的字节数，和该字节数对应哈希桶的下标，我们可以将他封装在一个类当中：

// 管理对齐规则和映射等关系
class SizeClass
{
public:// 获取向上对齐后的字节数static inline size_t RoundUp(size_t bytes){// ...}// 获取对应哈希桶的下标static inline size_t Index(size_t bytes){// ...}
};

因为这个类中没有成员，我们可以设为静态的否则调用时还要创建对象，另外我们这些函数会频繁的调用所以设置为内联！

---

当我们我去到申请的内存根据其字节时，可以判断它属于哪一个区间，然后将申请的大小和对其数给子函数让子函数去进行对齐工作：

// 根据 bytes 和 alignNUm 进行对齐
static inline size_t _RoundUp(size_t bytes, size_t alignNum)
{}// 获取向上对齐后的字节数
static inline size_t RoundUp(size_t bytes)
{// 整体控制在最多10%左右的内碎片浪费// [1,128]               8byte对齐        freelist[0,16)// [128+1,1024]          16byte对齐       freelist[16,72)// [1024+1,8*1024]       128byte对齐      freelist[72,128)// [8*1024+1,64*1024]    1024byte对齐     freelist[128,184)// [64*1024+1,256*1024]  8*1024byte对齐   freelist[184,208)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;}
}

下面追要工作就是实现这个子函数了！一般我们可能想到的做法就是：

static inline size_t _RoundUp(size_t bytes, size_t alignNum)
{size_t alignSize = 0;if (bytes % alignNum == 0){// 申请的字节数是对其数的整数倍，无需对其alignSize = bytes;}else{// 因为向上取整，所以让 bytes / alingNum 的商 + 1就想上取整了，然后 在 乘 alignNum 即可alignSize = (bytes / alignNum + 1) * alignNum;}return alignSize;
}

这也是最能想到的方法，但是除了这种方法之外，还有一种位运算的方法很精妙：

static inline size_t _RoundUp(size_t bytes, size_t alignNum)
{// 当前字节数 + 对其数 - 1 就永远到不了，对其数的下一个整数倍，但是 >= alignNum 的 // 当前的结果 是由 最大对其数的 位数 中的 其中几位 组合而成的// 再将 对其数 -1 取反，意味着最大对其数的 最高位不变，后面的低位全部取反// 在和 + 对其数 - 1 的结果 按位与，就只剩下 对其数了// 例如：[1-8] --> + 8-1 = 7 ===> [8-15], 7 取反 == 1000 & [8-15] == 8return (bytes + (alignNum - 1)) & ~(alignNum - 1);
}

---

在获取某一字节对应的下标时，我么也可以根据字节的大小判断他在哪一个区间，然后调用子函数进行处理：

// 获取对应哈希桶的下标
static inline size_t Index(size_t bytes)
{assert(bytes <= MAX_BYTES);// 每个区间有多少个 桶（链）static int grounp_array[4] = { 16, 56, 56, 56 };if (bytes <= 128){return _Index(bytes, 3); // 2^3}else if (bytes <= 1024){// 这里因为每个区间采用的对齐数不一样，说我我们 采用绝对映射// 最后为了映射的正确性，我们需要加上 前面的一段的 桶的个数return _Index(bytes - 128, 4) + grounp_array[0];// 2^4}else if (bytes <= 8 * 1024){return _Index(bytes - 1024, 7) + grounp_array[0] + grounp_array[1];// 2^7}else if (bytes <= 64 * 1024){return _Index(bytes - 8 * 1024, 10) + grounp_array[0] + grounp_array[1] + grounp_array[2];// 2^10}else if (bytes <= 256 * 1024){return _Index(bytes - 64 * 1024, 13) + grounp_array[0] + grounp_array[1] + grounp_array[2] + grounp_array[3];// 2^13}else{// 为了避免程序出错走到这里assert(false);return -1;}
}

子函数的实现，也有两种方式，第一种是我觉的我们可以想到的，就是使用 字节大小和对其数 做除法，根据是否整除判定：

// 根据字节数和对其数判断 映射的位置
static inline size_t _Index(size_t bytes, size_t alignNum)
{if (bytes % alignNum == 0){// 如果 bytes / alignNum 是整除的，说明当前字节的位置就是, 商 的 前一个// 例如：Bytes = 8 ,alignNum = 8 ==> bytes / alignNum == 1， 而 [1-8] 是 0 号桶return bytes / alignNum - 1;}else{// 否则，上就是 合适的位置// 1 / 8 == 0 return bytes / alignNum;}
}

除了这种方式之外也还有一种很精妙的位运算的方法：他是使用字节数和对齐数的倍数进行操作的：

// 根据字节数和对其数的二进制次方判断 映射的位置
static inline size_t _Index(size_t bytes, size_t align_shift)// align_shift 表好似对其数对应的次方数
{// 当前的 bytes + 1 << align_shift) - 1 本质就是 bytes + 对其数 - 1，也就是刚好永远到不了，下一个对其数的倍数处// 他因为对其数 是 2^x，所以再将 上述的结果 >> align_shift 就是，对应桶的下一个位置 ，在 -1 即可return ((bytes + (1 << align_shift) - 1)) >> align_shift - 1;
}

OK，了解了内存对齐规则后，我们就可以知道 在thread cache 中有 208 个桶

static const size_t MAX_BYTES = 256 * 1024; // thread 最大申请的内存大小为 256 KB
static const size_t N_FREE_LIST = 208; // thread cahce 中一共有 208 个桶

thread cache 申请和释放的逻辑也就很简单了：

// 申请和释放内存对象
void* ThreadCache::Allocate(size_t size)// 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);}
}

void  ThreadCache::Deallocate(void* ptr, size_t size)// ptr 代表还回来内存的地址，size 表示大小
{assert(ptr);assert(size <= MAX_BYTES);// 根据内存的大小，计算映射下标，插入到自由链表即可size_t index = SizeClass::Index(size);_freelists[index].Push(ptr);
}

三、TLS 无锁的访问

每一个线程都有一个自己的 thread cache ，但是如何创建他呢？首先不能将创建thread cache 方成全局的，因为全局是所有线程共享的，这样的话就又得使用锁来控制

要实现每个线程无锁的访问属于自己的thread cache，我们需要用到线程局部存储TLS(Thread Local Storage)，这是一种变量的存储方法，使用该存储方法的变量在它所在的线程是全局可访问的，但是不能被其他线程访问到，这样就保持了数据的线程独立性。

//TLS - Thread Local Storage
static _declspec(thread) ThreadCache* pTLSThreadCache = nullptr;

我们可以在提供两个接口，用于 TLS无锁访问的控制：

static void* ConcurrentAlloc(size_t size)
{// TLS 无锁访问// 当线程调用 alloc 时，才创建 thread cache 对象if (pTLSThreadCache == nullptr){pTLSThreadCache = new ThreadCache;}std::cout << std::this_thread::get_id() << ": " << pTLSThreadCache << std::endl;return pTLSThreadCache->Allocate(size);
}static void ConcurrentFree(void* ptr, size_t size)
{assert(pTLSThreadCache);pTLSThreadCache->Deallocate(ptr, size);
}

简单测试一下，无锁访问：

void Alloc1()
{for (size_t i = 0; i < 5; ++i){void* ptr = ConcurrentAlloc(6);}
}void Alloc2()
{for (size_t i = 0; i < 5; ++i){void* ptr = ConcurrentAlloc(7);}
}void testTLS()
{std::thread t1(Alloc1);std::thread t2(Alloc2);t1.join();t2.join();
}int main()
{//TestObjectPool();testTLS();return 0;
}

上面代码的意思是，两个线程分别执行 Alloc1 和 Alloc2  ，如果按照上面的 TLS 的机制，应该是连个线程的 pTLSThreadCache 的值分别都是唯一的：

OK，证明我们当前的逻辑是正确的！这就是 thread cache 层了，下一期我们来探索 central cache 层！