从零实现一个高并发内存池 - 2

上一篇https://blog.csdn.net/Small_entreprene/article/details/147904650?fromshare=blogdetail&sharetype=blogdetail&sharerId=147904650&sharerefer=PC&sharesource=Small_entreprene&sharefrom=from_link

高并发内存池 - thread cache

一、基本结构与原理

Thread cache 基于哈希桶结构构建，每个哈希桶均是一个按桶位置映射大小的内存块对象自由链表。系统为每个线程配备独立的 thread cache 对象，使得各线程在该对象上进行内存的获取与释放操作时无需加锁，提升操作效率。

哈希桶是 thread cache 的基础结构，每个哈希桶代表特定大小内存块的集合，每个桶维护一个自由链表，链表中的节点是空闲的内存块对象。图片中的蓝色区域左侧部分展示了不同大小内存块对应的哈希桶，如 8Byte、16Byte、24Byte 等，每个桶通过链表（绿色方块及橙色箭头）连接空闲内存块。

我们实现thread cache主要的流程是：分段对齐---对齐后确定桶的位置，获取桶下的_freeList链表内容---没有就向central cache获取：

以下是细节解析：

对应映射规则

步骤	操作	说明
分段	将内存大小划分为不同的区间段：
	- 小于等于 128 字节
	- 大于 128 字节且小于等于 1024 字节
	- 大于 1024 字节且小于等于 8 KB（1024 × 8 字节）
	- 大于 8 KB 且小于等于 64 KB（1024 × 64 字节）
	- 大于 64 KB 且小于等于 256 KB（1024 × 256 字节）
	目的：适应不同大小的内存申请需求，提高内存分配的灵活性和效率；减少内碎片，通过为不同大小范围的内存块设置不同的对齐规则，使内存块的大小更加合理，避免因内存块大小不匹配而导致的浪费。
对齐	在每个分段区间内，根据内存块的大小确定对齐方式：
	- <=128字节：8字节对齐，确保内存块大小是 8 的整数倍。
	- 128<字节<=1024字节：16字节对齐，确保内存块大小是 16 的整数倍。
	- 1024<字节<=8×1024字节（即 8 KB）：128字节对齐。
	- 8×1024<字节<=64×1024字节（即 64 KB）：1024字节对齐。
	- 64×1024<字节<=256×1024字节（即 256 KB）：8×1024字节对齐（8 KB 对齐）。
	目的：内碎片控制，确保每个内存块的大小是其对齐单位的整数倍，减少内碎片，提高内存利用率；硬件兼容性，遵循现代计算机体系结构对内存访问的对齐要求，确保内存块的起始地址满足硬件的对齐规范，提高系统的稳定性和性能。
桶位置映射	通过特定的函数（如 SizeClass::Index）将对齐后的内存块大小映射到哈希桶的位置（即自由链表的索引）：
	1. 确定分段区间：根据申请的内存大小，确定其所属的分段区间。
	2. 计算对齐大小：根据分段区间对应的对齐规则，计算出对齐后的内存块大小。
	3. 映射桶位置：根据对齐后的内存块大小，使用特定的映射规则（如分组数组和位移操作）计算出其对应的哈希桶索引。
	映射的具体实现逻辑：
	- <128字节：对齐到 8 字节后，使用 _Index 函数（基于位移和减法操作）计算桶索引。
	- 128到1024字节：对齐到 16 字节后，同样使用 _Index 并结合分组数组偏移计算索引。
	- 1024到8192字节：对齐到 128 字节后，结合前面的分组和偏移确定索引。
	- 8192到65536字节：对齐到 1024 字节后，进行对应的索引计算。
	- 65536到262144字节：对齐到 8192 字节后，确定索引。
	目的：快速定位，通过将内存块大小映射到特定的哈希桶位置，可以快速定位到存储该大小空闲内存块的自由链表，提高内存分配和释放的效率；组织管理，将相同大小（或相近大小）的内存块组织在同一个哈希桶的自由链表中，便于管理和批量操作，减少内存分配和回收时的搜索复杂度。

以下是完整流程示例：

步骤	内存申请大小	分段区间	对齐大小	对齐规则	桶位置索引计算
示例	76 字节	128 字节	80 字节（_RoundUp(76, 16) = 80）	16 字节对齐	使用 _Index 函数结合分组数组偏移计算索引，假设得到索引为 i。

这种分段 - 对齐 - 桶位置映射的完整流程，可以实现对内存块的高效管理，优化内存分配和释放性能，同时控制内碎片和提高系统稳定性。

为什么需要分段对齐？

内碎片控制：通过将内存块按照不同的大小范围（如 1 - 128 字节、128 - 1024 字节等）进行分段对齐，可以将内碎片控制在合理的范围内。例如，对于小内存块（<=128 字节），采用 8 字节对齐；对于中等大小内存块（128 - 1024 字节），采用 16 字节对齐。这样可以确保每个内存块的大小都是对齐单位的整数倍，避免内存浪费。如果统一采用较大的对齐单位（如 128 字节），则小内存块的内碎片会显著增加，导致内存利用率下降。

减少外部碎片：分段对齐有助于更好地组织和管理内存块，使得不同大小的内存块能够更高效地被分配和回收，减少外部碎片。例如，当一个线程申请 100 字节的内存时，系统会分配一个 128 字节（经过 8 字节对齐后的大小）的内存块。这种对齐策略使得内存块的大小更加规则，便于后续的内存分配和回收操作。

快速匹配内存块：当线程申请内存时，系统可以根据申请的内存大小快速确定对应的对齐规则和哈希桶位置，从而快速找到合适的空闲内存块。这种分段对齐的方式使得内存分配算法更加高效，减少了在查找合适内存块时的计算复杂度。例如，对于 7 字节的内存申请，系统会将其对齐到 8 字节，并直接定位到对应的哈希桶，快速分配内存。

批量处理内存块：在从 central cache 向 thread cache 批量获取内存块时，分段对齐使得内存块的大小更加统一，便于批量操作和管理。例如，系统可以一次性获取多个 8 字节对齐的内存块，并将它们插入到对应的自由链表中，提高内存分配的效率。

我们内存对齐可以认为是缩小范围，减少空间浪费，没必要对应每一个n字节大小的都对应一个桶，一个_freeList，这样的话128KB就需要2万多个桶，也就是2万多个_freeList，通过分段对齐，我们总共只需要有208个桶位置/_freeList。

二、内存申请流程

1. 当申请的内存大小（size）不超过 256KB 时，线程首先获取本地存储的 thread cache 对象，依据特定规则计算出 size 对应的哈希桶自由链表下标 i。

2. 若自由链表 _freeLists[i] 中存在可用内存对象，则直接从中弹出（Pop）一个内存对象并将其返回给请求者，完成内存分配。

3. 一旦发现 _freeLists[i] 中无可用内存对象，系统将从 central cache 批量获取一定数量的内存对象，将这些对象插入到自由链表后，从中取出一个返回给请求者，以满足内存申请需求。

到了第三步，我们可能会有这样的问题：

我们知道，每一个线程都会有自己独享的thread cache，thread cache里面是一个哈希桶，每个桶里面挂的就是切好的小对象组织起来的自由链表，需要使用对应大小的内存就去找到对应映射位置的桶下的自由链表去取，有的话就直接使用，有的话效率是很高的，没有的话就去下一层去申请，如果是往下申请，那我们当前线程如何获取到的呢？

我们知道在一个进程里面，可能有多个线程，多个线程共享整个进程地址空间，每一个线程有自己独立的栈，寄存器等等，有些东西是共享的，比如说全局数据段，代码段等等，那么每一个线程都要有自己的thread cache，那么这个thread cache又是如何创建的呢？

我们不想要通过锁的方式来解决，那么我们该怎么办呢？其实实际当中真正要通过无锁，我们还需要补充一个知识：tls（不是网络当中的tls）

TLS - thraed local storage（线程本地存储）

TLS（Thread Local Storage，线程本地存储）是一种内存管理机制，用于为每个线程提供独立的变量副本，使得每个线程可以拥有自己的数据，而不会与其他线程的数据发生冲突。与其他线程共享的全局变量或静态变量不同，TLS 变量在每个线程中都是独立的。

linux gcc下 tls

windows下 tls

_declspec(thread) 用于声明线程本地存储变量，每个线程对该变量的访问是独立的，因此不需要锁机制来保护该变量的访问。线程本地存储变量的设计目的就是为了在多线程环境中，为每个线程提供独立的变量副本，从而避免线程之间的竞争和数据不一致问题。

pTLSThreadCache 是一个线程本地存储指针，每个线程都有自己的 pTLSThreadCache 副本，对这个指针的赋值和读取操作是在线程内部进行的，不会与其他线程的操作发生冲突，因此不需要额外的同步措施。

不过，在实际使用中，如果对 pTLSThreadCache 所指向的对象（即 ThreadCache 对象）进行复杂的操作，而这些操作可能涉及多个线程间共享的数据，那么在操作这些共享数据时可能需要使用锁或其他同步机制来保证线程安全。但对于 pTLSThreadCache 指针本身的简单赋值和读取操作，并不需要锁。

我们的详细细节都体现在了实现的代码当中：

Common.h:

#pragma once
#include<iostream>
#include<vector>
#include<ctime>
#include<windows.h>
#include<assert.h>
#include<thread>//方便，不使用using namespace std;是因为防止污染
using std::cout;
using std::endl;static const size_t MAX_BYTES = 256 * 1024;
static const size_t NFREELIST = 208;//总共的桶的数量static void*& NextObj(void* obj)
{return *(void**)obj;
}// 管理切分好的小对象的自由链表
class FreeList
{
public:void Push(void* obj){assert(obj);// 头插//*(void**)obj = _freeList;NextObj(obj) = _freeList;_freeList = obj;}void PushRange(void* start, void* end){NextObj(end) = _freeList;_freeList = start;}void* Pop(){assert(_freeList);// 头删void* obj = _freeList;_freeList = NextObj(obj);return obj;}bool Empty(){return _freeList == nullptr;}private:void* _freeList = nullptr;//这个要写，因为我们没有写构造
};// 计算对象大小的对其映射规则
class SizeClass
{
public:// 提供函数来计算给一个字节数，对应到正确的桶位置；// 规则如下：// 整体控制在最多10%左右的内碎片浪费！！！// ***************************************************************************************************// ***************************************************************************************************// [1,128]					            8byte对齐	         freelist[0,16)         这个没办法>^<// [128+1,1024]				        16byte对齐	         freelist[16,72)       15/（129+15）=0.10....// [1024+1,8*1024]			    128byte对齐	     freelist[72,128)     127/（1025+127）=0.11....// [8*1024+1,64*1024]		    1024byte对齐      freelist[128,184)   //......// [64*1024+1,256*1024]		8*1024byte对齐   freelist[184,208)  //......// ***************************************************************************************************// ***************************************************************************************************//相当于RoundUp的子函数：给定当前size大小和对应规则的对齐数AlignNum，用于处理//static inline size_t _RoundUp(size_t size, size_t alignNum)//{//	size_t alignSize;//	if (size % alignNum != 0)//	{//		alignSize = (size / alignNum + 1) * alignNum;//	}//	else//等于0就不需要处理了，已经对齐了//	{//		alignSize = alignNum;//	}//}//上面的是我们普通人玩出来的，下面我们来看看高手是怎么玩的！static inline size_t _RoundUp(size_t bytes, size_t alignNum){return ((bytes + alignNum - 1) & ~(alignNum - 1));}//你给我一个size，就需要算出对其以后是多少 --- 比如说：8->8  7->8static 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{//说明大于256KB了，那么就有点问题了assert(false);return -1;}}//一样的，我们来自己写写看，待会换成别人的刚好的方式//size_t _Index(size_t bytes, size_t alignNum)//{//	if (bytes % alignNum == 0)//	{//		return bytes / alignNum - 1;//	}//	else//	{//		return bytes / alignNum;//	}//}//高手的想法：static inline size_t _Index(size_t bytes, size_t align_shift){return ((bytes + (1 << align_shift) - 1) >> align_shift) - 1;}// 计算映射的哪一个自由链表桶static inline size_t Index(size_t bytes){assert(bytes <= MAX_BYTES);// 每个区间有多少个链static int group_array[4] = { 16, 56, 56, 56 };if (bytes <= 128) {return _Index(bytes, 3);}else if (bytes <= 1024) {return _Index(bytes - 128, 4) + group_array[0];}else if (bytes <= 8 * 1024) {return _Index(bytes - 1024, 7) + group_array[1] + group_array[0];}else if (bytes <= 64 * 1024) {return _Index(bytes - 8 * 1024, 10) + group_array[2] + group_array[1] + group_array[0];}else if (bytes <= 256 * 1024) {return _Index(bytes - 64 * 1024, 13) + group_array[3] + group_array[2] + group_array[1] + group_array[0];}else {assert(false);}return -1;}
private:};

ThreadCache.h:

#pragma once
#include"Common.h"class ThreadCache
{
public:// 申请和释放内存对象void* Allocate(size_t size);void Deallocate(void* ptr, size_t size);// 从中心缓存获取对象void* FetchFromCentralCache(size_t index, size_t size);
private:FreeList _freeLists[NFREELIST];//_freeLists 是一个数组，用于存储多个 FreeList 类型的对象。
};
//线程隔离 ：每个线程对 pTLSThreadCache 的访问都是独立的，一个线程对其的修改不会影响其他线程中的该变量值。
//避免锁竞争 ：由于每个线程都有自己的变量副本，避免了多线程环境下因共享变量而导致的锁竞争问题，提高了程序的并发性能。
//方便线程专用数据管理 ：适合存储线程专用的资源或状态信息，例如每个线程的缓存、配置等。
static _declspec(thread) ThreadCache* pTLSThreadCache = nullptr;
//实际中，不可能让每一个线程自己来获取ThreadCache*这个对象，我们还需要再封装一层——————》ConcurrentAlloc.h
//也就是说：一个线程起来了，并不是马上就有pTLSThreadCache了，而是需要去调用相关的函数

ThreadCache.cpp: 

#include"ThreadCache.h"//声明与实现分离void* ThreadCache::FetchFromCentralCache(size_t index, size_t size)
{//...return nullptr;
}void* ThreadCache::Allocate(size_t size)
{assert(size <= MAX_BYTES);//并不是所有要求申请的字节数都有完全对应的一个桶位，是1字节，7字节都是放在对应的8字节桶中，那么我们就需要实现一个对应的对其映射的规则：我们在Common.h中实现size_t alignSize = SizeClass::RoundUp(size);//对齐数size_t index = SizeClass::Index(alignSize);//计算出了对齐数，那么又是对应的哪一个桶呢？//通过TLS，每个线程无锁的获取自己专属的ThreadCache对象if (!_freeLists[index].Empty()){return  _freeLists->Pop();}else//这个桶下面的自由链表为空，为空只能向下一层去要了{return FetchFromCentralCache(index, alignSize);}
}void ThreadCache::Deallocate(void* ptr, size_t size)//free只需要传入对应要free空间的指针就可以了，但是为了知道放入到对应的正确的桶位置，需要size参数来定位桶的位置
{assert(ptr);assert(size <= MAX_BYTES);//找出对应的自由链表的桶，插入进去size_t index = SizeClass::Index(size);_freeLists[index].Push(ptr);
}

ConcurrentAlloc.h:

#pragma once
#include"Common.h"
#include"ThreadCache.h"//搞成全局静态，不然包在多个.cpp当中的话，静态的保持在当前文件可见，否则全局的不加上静态，其链接属性就会冲突(因为一个.h会形成一个obj，所包含的可能最终会导致冲突！)
static void* ConcurrentAlloc(size_t size)
{if (pTLSThreadCache == nullptr){pTLSThreadCache = new ThreadCache;}cout << std::this_thread::get_id() << ":" << pTLSThreadCache << endl;return pTLSThreadCache->Allocate(size);
}static void ConcurrentFree(void* ptr, size_t size)//我们传入size参数的问题更后面整体联系起来了再解决！！！
{assert(pTLSThreadCache);pTLSThreadCache->Deallocate(ptr, size);
}

注意：在创建线程时，线程缓存的初始化也是从 central cache 获取内存的，这是因为线程缓存本身不存储内存块，它需要从 central cache 获取内存来初始化自己的自由链表。当线程申请内存时，如果线程缓存的自由链表为空，则也会从 central cache 获取内存。这种机制确保了线程缓存中有足够的内存块来满足线程的内存申请需求，同时减少了从 central cache 获取内存的频率，提高了性能。（一点要理清楚哦）

三、内存释放流程

1. 对于释放的内存大小小于 256k 的情况，会将内存释放回 thread cache。此时，同样根据内存块的大小计算出对应的自由链表桶位置 i，将待释放的内存对象推入（Push）到 _freeLists[i] 中，完成内存释放。

2. 在内存释放过程中，若发现自由链表的长度超出设定阈值，即链表过长时，会触发内存回收机制，将部分内存对象从 thread cache 回收至 central cache，以优化内存的组织与管理，避免 thread cache 内内存对象的过度堆积。

都实现在了上面的代码当中！！！

接下来，我们就需要往下层挖掘，下一篇，我们精彩继续：central cache！