1. 项目介绍

  当前项目旨在实现一个高并发内存池，其原型是 Google 的开源项目 TCMalloc, 全称 Thread-Caching Malloc，是 Google 开发的高性能内存分配器，用于替代 C/C++ 标准库中的 malloc。它专为多线程环境设计，旨在解决传统内存分配器（如 glibc 的 ptmalloc2 ）在高并发场景下的性能瓶颈和锁竞争问题。

  TCMalloc的核心设计理念是通过分层缓存与无锁化前端来彻底化解多线程内存分配中的锁竞争瓶颈，其架构自顶向下分为三层：前端采用线程或CPU局部的无锁缓存，使绝大多数malloc/free操作无需加锁便可迅速完成；中端通过中央空闲链表和传输缓存负责在前端与后端之间批量搬运内存对象，以降低跨线程交互的频率；后端则基于PageHeap直接与操作系统交互，按需申请或释放大块内存。在关键技术特性上，TCMalloc将小对象精细划分为约60至80个尺寸类别（size-classes），并让特定尺寸的对象占用专属的连续内存页（Span），从而极大降低内存碎片与内部簿记开销，同时对大对象操作辅以细粒度自旋锁来平衡性能与安全。此外，它还充分吸纳了现代操作系统特性，如在支持RSEQ（Restartable Sequences）的Linux内核上默认启用每CPU缓存模式以榨取极致性能，并在后端实现了对大页（Hugepage）的感知能力，以此兼顾高并发吞吐量与大规模服务器场景下的内存使用效率。

  下面是TCMalloc在Gitee上的源代码：

https://gitee.com/mirrors/tcmalloc

  这个项目的做法，是将 TCMalloc 最核心的框架简化剥离，然后模拟实现一个自己的高并发内存池。本质上，这和我们一路学习的方法如出一辙——通过模仿简化来深刻理解知识，就像学习STL的时候一样，我们习惯于最后自己手搓封装。不过，TCMalloc 的代码量和复杂度远超 STL 实现。

  另外需要留意的是，因为 TCMalloc 出身于 Google 这样的顶尖大厂，出自当时业界一流的 C++ 工程师之手，其知名度极高，被众多公司广泛使用，甚至 Go 语言也直接拿它作为自己的内存分配器。因此，很多面试官对这个项目相当熟悉。这既是利好，也是压力：好处是，如果你把项目理解扎实，面试时很容易获得认可；坏处是，正因为面试官可能也很懂，所以会问得深、问得细，稍有不实就容易露怯。

  最后，关于这个项目所需的知识储备和难度——将会用到 C/C++、数据结构（链表、哈希桶）、操作系统内存管理、单例模式、多线程及互斥锁等知识点。

2. 什么是内存池

2.1 池化技术

  池化技术（Pooling Technique）是一种以“空间换时间”为核心理念的资源管理策略。它颠覆了传统 “按需申请、用完即毁”的瞬态模式，改为“预申请、缓存、复用”的循环模式。简单来说，就是在系统启动或空闲时，提前向操作系统申请一大块资源保存在“池”中，当业务线程真正需要时，直接从池中“借用”而非“新建”，使用完毕后“归还”给池而非“销毁”。

  池化技术的核心机制通常包含以下四个环节：

  1. 初始化：根据预设容量（如最大连接数、内存池总大小），一次性向操作系统申请足量资源。

  2. 借用：当外部请求到来时，从池中分配一个空闲资源给请求者。若池中无空闲资源，视策略决定是阻塞等待还是动态扩容。

  3. 归还：请求者使用完毕后，将资源释放回池中，重置其内部状态以便下次复用，而不是真正释放给操作系统。

  4. 销毁：在系统关闭或池容量过剩时，统一清理池中的全部资源，归还给操作系统。

  然而，池化技术并非银弹，它的主要代价在于内存常驻（即使在空闲时也会占用资源）和调优复杂（池容量设置过大会造成浪费，设置过小则容易导致请求等待甚至饥饿）。因此，一个优秀的池化设计，往往需要结合业务负载的峰值与均值，精确调控池的上下水位，这在 TCMalloc 的后端 PageHeap 管理中体现得尤为明显——它甚至会根据释放的内存块大小动态决定是否真正归还给操作系统，以此平衡性能与内存占用。

2.2 内存池

  内存池（Memory Pool）是池化技术在内存管理领域的具体应用。它并非每次动态申请都直接与操作系统交互，而是在程序启动或运行期间，预先向操作系统申请一大块连续内存（称为“堆”或“池”），然后由内存池内部维护的数据结构（如链表、位图、哈希桶）负责将这块内存切割成大小不一的块，并管理这些块的分配与回收。

2.3 内存池主要解决的问题

  内存池主要致力于解决系统原生内存分配器比如 malloc / free 或者 new / delete 等，在实际工程中暴露出的三大痛点：

  1. 极致的性能开销：直接调用系统调用（如 brk 或 mmap）涉及用户态与内核态的切换，而多线程环境下默认的 malloc 实现往往还伴随全局锁的竞争。内存池通过批量预申请和无锁化的线程本地缓存（如 TCMalloc 的设计），使得大多数分配请求只需几次指针操作即可完成，大幅减少了 CPU 上下文切换和锁等待的时间。

  2. 内存碎片问题：频繁地按任意大小申请和释放内存，会导致堆空间中布满大量无法被有效利用的、细小的不连续空闲区域（即“外部碎片”）。内存池通常采用固定尺寸类别（如 8 字节、16 字节、32 字节……）或固定块的方式进行管理，将碎片控制在一定范围内，并通过针对性的回收合并策略，有效缓解了内存使用率的劣化。

  3. 分配的不确定性：原生内存分配器在运行时的耗时并不稳定（非确定性），当堆空间不足时，它会触发复杂的内存合并、回收甚至向 OS 重新申请的逻辑，这在实时系统或游戏引擎中容易导致致命的卡顿（延迟抖动）。内存池通过“预分配”机制，将内存的稀缺性风险前置到初始化阶段，从而保证了运行时分配的高效性与稳定性。

  总而言之，内存池通过以空间换时间和精细化管理，在提升多线程并发能力的同时，降低了内存碎片的危害，是构建高性能基础组件（如 TCMalloc、Jemalloc）的核心基石。

2.4 malloc

  C/C++ 中我们要动态申请内存都是通过 malloc 去申请内存，但是我们要知道，实际我们不是直接去堆获取内存的， 而 malloc 就是一个内存池。malloc () 相当于向操作系统 “批发” 了一块较大的内存空间，然后 “零售” 给程序用。当全部 “售完” 或程序有大量的内存需求时，再根据实际需求向操作系统 “进货”。 malloc 的实现方式有很多种，一般不同编译器平台用的都是不同的。比如 windows 的 vs 系列用的微软自己写的一套，linux gcc 用的 glibc 中的 ptmalloc。 

3. 设计一个定长内存池

  我们先利用设计一个定长内存池来熟悉一下简单内存池是如何控制的，第二他会作为我们后面内存池的一个基础组件。

  这里我们用 char* _memory 指向一次性从操作系统堆上申请的一整块连续大内存，这是整个内存池的底层物理存储空间。 定长内存池初始化时，只会调用一次 malloc 申请一大片连续内存，后续分配 / 释放都在这块内存内部操作，不再频繁调用系统 malloc。为什么强转成 (char*)？因为 char* 按字节移动，_memory += sizeof(T) 的时候，sizeof(T) 是多少个字节，指针就往后移动多少个字节，方便我们精确地按字节切内存。

  而 void* _freeList 指向空闲链表的头节点，由于整块 _memory 是一片连续内存，初始化时会切成 N 个大小完全相同的定长块。 程序运行时会发生两种操作：

• 分配：用户来申请一块内存，要快速拿一块空闲的返回；
• 释放：用户用完归还内存，要快速把这块回收备用。

  如果没有链表，每次分配都要遍历整块内存找空白块，效率极低； 用单向空闲链表把所有被释放后的空闲块串起来。

template<class T>
class ObjectPool
{
public:
	T* New()
	{
		if (_memory == nullptr)
		{
			_memory = (char*)malloc(128 * 1024);
			if (_memory == nullptr)
			{
				throw bad_alloc();
			}
		}
		T* obj = (T*)_memory;
		_memory += sizeof(T);

		return obj;
	}

private:
	char* _memory = nullptr;
	void* _freeList = nullptr;
};

  现在出现一个新的问题，当释放内存的时候，要实现一个空闲链表（_freeList）用来串起所有被释放回来的内存块。这个链表需要：

• 每个节点里存一个“指向下一个节点的指针”。

• 这个指针在 32 位系统下占 4 字节，64 位系统下占 8 字节。

  但问题的关键点是：你完全不知道用户用这个内存池来实例化什么类型 T。如果 T 是 int，sizeof(T) 是 4，链表指针占 4 个字节，刚好塞得下；如果 T 是一个只有 1 字节的 char，sizeof(T) 只有 1，那这 1 个字节连个指针都存不下，强行写 4 个字节就会越界踩坏别人的内存；如果 T 是一个巨大的结构体，那 sizeof(T) 倒是够用，但你也无法依赖这个大小。

  所以矛盾就是：链表节点需要固定大小的指针，但对象类型大小是未知且不可控的。

  我们初步的思路是这样的：

void Delete(T* obj)
{
    if (_freeList == nullptr)
    {
        _freeList = obj;
        *(int*)obj = 0;
    }
}

  这里做了一个非常巧妙且“霸道”的假设：当一个对象被释放时，这块内存已经不再被程序逻辑使用。既然不再用了，这块内存里原本存的是什么根本不重要——它现在就是一块纯粹的、空闲的、可以随意改写的内存。

  那么，我们可以把这块内存本身当作链表节点来使用。具体来说就是把 obj 这个指针强行解释为链表的头节点指针，存入 _freeList。然后，把 obj 所指向的那块内存的前 4 个字节（在 32 位系统下），写入下一个节点的地址。因此  *(int*)obj = nullptr; 这行就是：

  1. 把 obj 强转成 int*，意思是把这块内存当作一个 4 字节的指针槽位来看待。

  2. 然后往这个槽位里写入 nullptr，表示这个节点后面没有下一个空闲块了，同时obj成为了链表节点。

  之所以强转成 int* ，是因为你不能直接让 *obj = nullptr ， obj 是 T* 类型，它指向的内容类型是 T，而 T 可能是任何类型，未必支持赋值 nullptr（比如 T 是 int 时，*obj = nullptr 就报错了）。而 (int*)obj 的意思是：“我不管你这块内存原来存的是啥，我现在就把它当做一个存放指针的 4 字节空间来用”。这样就避开了类型安全约束，实现“内存的复用”。

  但是这段代码其实是在 32 位系统 下写的，所以指针大小是 4 字节，用 int* 来强转刚刚好。但如果你在 64 位系统下编译，指针大小是 8 字节，那么 *(int*)obj = nullptr; 只写了前 4 个字节，后 4 个字节仍残留了原 obj 的地址高位（因为 obj 本身是 64 位指针），这就出大事了——链表指针被截断，_freeList 跳转的时候就会访问到非法地址。

  所以正确的通用写法应该是利用二级指针：

	void* Delete(T* obj)
	{
		if (_freeList == nullptr)
		{
			_freeList = obj;
			//*(int*)obj = 0;
			*(void**)obj = nullptr;   
		}
	}

  对于二级指针，解引用后在 32 位下是 4 字节，64 位下是 8 字节，就会实现跟着平台自适应的效果，这里只是演示使用了 void** ，使用 int**，char** 等都是可以的。

  最后展示一下代码全貌：

#include <iostream>
#include <time.h>
#include <vector>
using std::cout;
using std::endl;


//定长内存池
//template<size_t N>
//class ObjectPool
//{};

template<class T>
class ObjectPool
{
public:
	T* New()//用于对象来申请空间
	{
		T* obj = nullptr;
		//先看看回收内存块的链表有没有可以使用的
		if (_freeList)
		{
			//头节点的下一个节点地址由 _freeList 的前四个字节存储
			void* next = *((void**)_freeList);
			obj = (T*)_freeList;
			_freeList = next;
			return obj;
		}
		else
		{
			//判断剩余内存是否足够下一个对象使用
			if (_remainBytes < sizeof(T))
			{
				_memory = (char*)malloc(128 * 1024);
				if (_memory == nullptr)
				{
					throw std::bad_alloc();
				}

                _remainBytes = 128 * 1024;
			}
			obj = (T*)_memory;  //对象接收内存
			
			size_t objSize = sizeof(T) < sizeof(void*) ? sizeof(void*) : sizeof(T);
			_memory += objSize;  //_memeory位置向后移动，可用内存减少
			
			_remainBytes -= objSize;
		}

		//定位 new ,显示调用T的构造函数初始化
		new(obj)T;
		return obj;
	}

	void Delete(T* obj)
	{
		//显示调用析构
		obj -> ~T();
		//头插节点
		*(void**)obj = _freeList;
		_freeList = obj;

	}

private:
	char* _memory = nullptr;  //指向内存的指针
	void* _freeList = nullptr;  //回收内存块形成的链表的头指针
	size_t _remainBytes = 0;  //内存的剩余字节数
};

4. 高并发内存池整体框架设计

  在现代多核多线程的开发环境下，内存分配面临的核心痛点就是锁竞争。虽然 malloc 本身已经足够优秀，但在高并发场景下，众多线程争抢同一把内存管理锁，会严重拖累性能。这正是 TCMalloc 大显身手的地方——它专为高并发而生。因此，我们要实现的内存池，必须重点攻克三个难题：极致性能、多线程锁竞争以及内存碎片。

  我们设计该并发内存池  ConcurrentMemoryPool  采用三层架构，各司其职：

1. Thread Cache（线程缓存）
   这一层是每个线程私有的"小金库"，负责分配小于 256KB 的内存。因为它是线程独占的，所以线程从这里申请或释放内存时不需要加锁，这是整个池子能如此高效的根本原因。每个线程都握着属于自己的缓存，互不干扰。

2. Central Cache（中心缓存）
   这是所有线程共享的"中央仓库"。Thread Cache 会按需从 Central Cache 批量"进货"（获取内存对象）；同时，Central Cache 也会在合适的时机，把一些线程中过于富余的内存"回收"回来，避免出现某个线程占用太多内存而其他线程却不够用的失衡局面。由于是共享资源，操作 Central Cache 时必须加锁，但这里采用了细粒度的桶锁，即哈希桶搭构的锁，且只有在线程的私有缓存完全耗尽时才会向这里申请，所以锁竞争并不激烈。

3. Page Cache（页缓存）
   这是最底层的"大块内存供货商"，管理着以页为单位的大块内存。当 Central Cache 的存货不足时，它会向 Page Cache 申请一定数量的页，然后将这些大页切割成小块，再分配给 Central Cache。反过来，当 Central Cache 中某个由多页组成的"Span"（跨度）的所有对象都被释放后，Page Cache 会将其回收，并尝试将相邻的空闲页合并成更大的连续页。这个合并机制有效缓解了内存碎片问题，使得大块内存的分配更加顺畅。

4.1 Thread Cache

4.1.1 哈希桶映射对齐规则

  thread cache 是哈希桶结构，每个桶是一个按桶位置映射大小的内存块对象的自由链表。每个线程都会有一个 thread cache 对象，这样每个线程在这里获取对象和释放对象时是无锁的。

   为什么采用哈希桶这个结构，我们得先从原生 malloc 的一个本质痛点说起。你仔细想想，如果让你设计一个内存分配器，面对的程序会申请 1 字节、16 字节、100 字节、2KB、100KB 这种五花八门的大小，你怎么管理？

  最朴素的想法是用一个全局自由链表，把所有释放的内存块串起来。申请时不管多大，都从链表头取一块；释放时直接头插回去。但这么做有两个致命问题：

1. 内存碎片：一块 100 字节的释放块，下次申请 200 字节时用不了，只能继续往后面找更大的块。堆空间很快就会被切割成无数细碎且大小不一的小块，明明总空闲内存很多，但任何一个请求都找不到足够大的连续空间。

2. 效率降低：为了能找到合适大小的块，你得遍历链表查找，时间复杂度 O(N)。更别说多线程环境下还得加锁，高并发时直接锁死。

  这就像你有一个巨大的仓库，但所有货物不分大小、不分类型全堆在一起。每次来找货，都得从头翻到尾，效率极低，而且时间长了仓库里全是零零碎碎的边角料，大件货物根本放不下。

  所以这里采用的映射规则是：使用一个数组哈希桶，数组里每一个下标（橙色方框）对应固定尺寸区间的内存块，当申请 N 字节内存，通过映射算法算出对应桶下标，只去该桶的自由链表取块。

  以 _freeList[0]举例 ， _freeList[0] 仅仅是一个指针变量（在 64 位下占 8 字节），它存储在 Thread Cache 的数组里。它的作用就像一根绳子的头，用来牵引后面所有的内存块，挂在 _freeList[0] 这个桶上的每一块内存碎片，大小都固定是 8 字节。

_freeList[0]   →  链表节点大小 = 8 字节    ← 这是第 1 个桶
_freeList[1]   →  链表节点大小 = 16 字节   ← 这是第 2 个桶
_freeList[2]   →  链表节点大小 = 24 字节   ← 这是第 3 个桶
...
_freeList[15]  →  链表节点大小 = 128 字节   ← 这是第 16 个桶
_freeList[16]  →  链表节点大小 = 144 字节   ← 这是第 17 个桶 (128+16)
...

  这其实是一个经过设计考量的算法，能控制内存浪费的原理是这样的：当你申请 129 字节时，Thread Cache 会给你 144 字节（16 对齐），那多出来的 15 字节就是单次分配的“对齐浪费”。控制手段就是我们在 SizeClass 里看到的那张区间表。它采用了“非均匀粒度”的对齐策略：

   1. 小尺寸（1~128）：用 8 字节对齐。最坏情况浪费 7 字节，占比 7/128 ≈ 5.4%。

   2. 中尺寸（129~1024）：用 16 字节对齐。最坏浪费 15 字节，占比 15/1024 ≈ 1.4%。

   3. 大尺寸（8KB~256KB）：用 8KB 对齐。最坏浪费 8KB，占比 8KB/256KB ≈ 3.1%。

  而这里对齐的最低标准是 8 字节，是因为自由链表要存储下一个节点的指针，指针大小在 64 位环境下是 8 字节，在 32 位环境下是 4 字节，为了确保能正常使用，所以最小也要 8 字节。

  下面我们看一下详细代码：

//计算对象大小的对齐映射规则
class SizeClass
{
public:
	// 整体控制在最多10%左右的内碎片浪费
    // [1, 128]                8byte对齐        freelist下标区间 [0, 16)
    // [129, 1024]             16byte对齐       freelist下标区间 [16, 72)
    // [1025, 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)


	static inline size_t _RoundUp(size_t size, size_t alignNum) //aligNum就是对齐数，8，16，128....
	{
		size_t alignSize; // 对齐后的数字
		if (size % alignNum != 0)  //假设要申请 9 个字节
		{
			alignSize = (size / alignNum + 1) * alignNum;
			//对齐后数字 = （9 / 8 + 1）* 8 = （1+1）* 8 = 16
		}
		else
		{
			alignSize = size;
		}
		return alignSize;
	}

	//static inline size_t _RoundUp(size_t bytes, size_t alignNum)
	//{
	//	return ((bytes + 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;
		}
	}

	static inline size_t _Index(size_t bytes, size_t alignNum) //bytes是申请的大小，alignNum是8，16，128......
	{
		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, 8);
		}
		
		else if (bytes <= 1024) 
		{
			return _Index(bytes - 128, 16) + group_array[0];
		}
		else if (bytes <= 8 * 1024) 
		{
			return _Index(bytes - 1024, 128) + group_array[1] + group_array[0];
		}
		else if (bytes <= 64 * 1024) 
		{
			return _Index(bytes - 8 * 1024, 1024) + group_array[2] +
				group_array[1] + group_array[0];
		}
		else if (bytes <= 256 * 1024) 
		{
			return _Index(bytes - 64 * 1024, 8*1024) + group_array[3] +
				group_array[2] + group_array[1] + group_array[0];
		}
		else 
		{
			assert(false);
		}
		return -1;
	}
};

  我们创建了一个 SizeClass 类，这个类做的所有事情，说到底就是两件事：用户申请 N 字节内存时，帮我们算出实际该给多少字节（对齐后的大小），以及算出这个大小对应哈希桶数组的哪个下标。这两件事是后续所有分配操作的基础，对齐不对，内存就乱；下标不对，桶就找错。

  我们先看对齐部分，_RoundUp(size, alignNum) 函数，它的逻辑非常朴实：如果 size 能被 alignNum整除，那就直接返回 size 本身；如果不能整除，就用（size / alignNum + 1）* alignNum 把它向上抬到 alignNum 的倍数。举个例子，申请 9 字节，对齐数是 8，9 除以 8 得 1 余 1，有余数，所以 (1+1)×8=16，9 字节就被抬到了 16 字节。而RoundUp（size） 就是对外暴露的入口函数，它根据用户申请的大小落在哪个区间，来决定用多大的对齐数，超过 256KB 就直接断言报错，因为 Thread Cache 不管这么大的内存。这样一来，当你在 ThreadCache::Allocate 里调用 SizeClass::RoundUp(size) 时，它就能返回一个对齐后的实际分配大小，比如申请 17 字节返回 24，申请 1025 字节返回 1152（因为 1025 在第三个区间，按 128 对齐，向上取到 1152）。

   接下来是下标计算的部分，_Index(bytes, alignNum) 是底层的映射工具，它的逻辑是算出 bytes 在当前对齐粒度下属于第几个槽位。而 Index(size) 是对外暴露的入口函数，它负责把用户申请的原始大小映射到全局唯一的桶下标。这里最关键的设计是 group_array[4] = {16, 56, 56, 56}，它记录了前四个区间各自占了多少个桶。所以在 Index 函数里，当 size 落在第一个区间时，直接调用 _Index(size, 8) 返回相对下标就行；当落在第二个区间时，先算 _Index(size - 128, 16) 得到在该区间内的相对位置，然后加上第一个区间的 16 个桶作为偏移量，就能得到全局下标；落在第三、第四、第五区间时也是同样的道理。

  所以整个逻辑串联起来就是：在 ThreadCache::Allocate 里，你先用 RoundUp(size) 拿到对齐后的实际分配大小，这个大小决定了你要从内存池里切出多大的内存块；然后用 Index(size) 拿到对应的桶下标，注意这里传入的是原始 size 而不是对齐后的 alignSize，因为 _Index 内部已经通过有余数除和无余数减一的逻辑，把原始大小正确映射到了对齐后的桶位置。拿到下标之后，你就去 _freeList[index] 这个桶里取链表头节点，如果链表非空就直接摘下来返回，如果链表为空就触发慢速路径向 Central Cache 批量申请。这就是 Thread Cache 分配内存的完整前置流程。

  另外大家可以看到被注释掉的两部分内容：

	static inline size_t _RoundUp(size_t bytes, size_t alignNum)
	{
		return ((bytes + alignNum - 1) & ~(alignNum - 1));
	}

	static inline size_t _Index(size_t bytes, size_t align_shift)
	{
		return ((bytes + (1 << align_shift) - 1) >> align_shift) - 1;
	}

  这其实是TCMalloc设计中用到的技巧，使用了位运算的技巧，一般情况下很难想得到，大家可以学习一下。

4.1.2 TLS无锁访问

  现在还有一个问题，我们来假设一个场景，比如现在有了两个线程——— t1 和 t2 ，它们分别去申请 6 字节和 7 字节。这里你要留意一个细节：6 字节和 7 字节经过 SizeClass 的对齐规则，都会被映射到 8 字节那个桶（也就是 _freeList[0] ）。所以这两个线程表面上是在申请不同的大小，实际上在底层，它们争抢的是同一个规格的桶——8 字节桶，这就是多线程同时向内存池申请小块内存的情况。

  这就引出了一个特别关键的问题：如果把 ThreadCache 设计成一个全局单例对象，那么t1 和 t2就会同时去操作这个全局对象的  _freeList[0]  链表头。而链表头插和头删的操作，本质上是对同一个指针变量（ _freeList[0] ）的读写。两个线程同时修改同一个内存地址，如果不加锁，就会发生数据竞争——t1 刚把链表头改成自己的节点，t2 又把它改成了另一个，导致链表指针错乱，程序崩溃。但如果你在操作链表头的时候加锁，那 Thread Cache 最核心的优势——“无锁分配”——就彻底没了，性能直接打回原形，跟 malloc 没啥区别了。

  而在 TCMalloc 中引入了 TLS（线程局部存储）的概念。TLS 的核心作用是让每个线程拥有自己独立的 ThreadCache 对象。对于线程 t1 来说，它访问的是自己专属的那份哈希桶数组；对于线程 t2 来说，它访问的是完全独立的另一份哈希桶数组。这样当两个线程向各自的哈希桶申请空间时，两个线程操作的是物理上完全不同的内存地址——t1 改的是自己地址空间里的链表头，t2 改的是另一个地址空间里的链表头，两者互不干扰，完全不需要加锁。

  我们主要是通过这行代码来实现，这行代码是在 Windows 平台（MSVC 编译器） 下，定义了一个线程局部存储变量。 _declspec(thread)  是微软编译器提供的一个扩展关键字，它告诉编译器：这个变量不是普通的全局变量，每个线程都拥有自己独立的一份副本。

  ThreadCache* 说明这个变量是一个指针，指向 ThreadCache 类型的对象。pTLSThreadCache 是这个变量的名字，从命名来看，它就是 “指向 Thread Local Storage 中 ThreadCache 对象的指针”。 nullptr 是初始化值，表示这个指针一开始是空的。

  这就是我们刚刚说的场景，两个线程同时去申请内存：

  这段代码的逻辑就是，pTLSThreadCache在一开始相当于是一个无人认领的空间，当t1执行ConcurrentAlloc的时候，因为此时的pTLSThreadCache是无人认领的，所以if(pTLSThreadCache == nullptr)这个语句是正确的，于是现在pTLSThreadCache创建了一个新的ThreadCache对象，并且将这个对象的地址存储到 pTLSThreadCache 变量里，此时这个pTLSThreadCache变量也属于 t1 了。然后再执行Allocate函数。当下一次再进入ConcurrentAlloc函数的时候，就不用执行 if(pTLSThreadCache == nullptr),因为t1已经有了属于自己的 pTLSThreadCache空间了。

  大家可以更生动形象的去理解：

  每个线程（t1、t2、主线程）在诞生的一瞬间，系统都给它们各自发了一个名为 pTLSThreadCache 的“空盒子”（盒子里初始塞着一张纸条，上面写着 nullptr）。

  完整的时间线是这样的：

  t1 第一次执行：

    t1 低头看了一眼自己手里的“空盒子”（pTLSThreadCache），发现纸条上是 nullptr。

    于是 t1 跑去堆上 new 了一个专门属于自己的 ThreadCache 对象（一个实体的内存池）。

    t1 把这个新对象的地址写在纸条上，塞回自己的盒子里。

    此时，这个盒子就属于 t1 了（准确说，盒子本来就是 t1 的，现在是盒子里的内容被 t1 的 ThreadCache 填满了）。

    然后 t1 通过盒子里的地址，找到那个对象，执行 Allocate 申请空间。

  t1 第二次执行：

    t1 再次低头看自己的盒子，发现纸条上写着一个有效的地址（不是 nullptr）。

    于是直接跳过 if，拿着这个地址就去执行 Allocate。

  与此同时，t2 第一次执行：

    t2 低头看自己手里的盒子（注意，t2 的盒子和 t1 的盒子是物理上不同的两个盒子）。

    因为 t2 从来没往自己盒子里写过东西，所以它盒子里的纸条依然是系统默认给的 nullptr。

所以 t2 也会走进 if，去 new 一个属于 t2 自己的 ThreadCache 对象，存在自己的盒子里。

4.2 Central Cache

  首先我们要明白 Central Cache 在整个架构里的角色定位。Thread Cache 是每个线程私有的“快取层”，负责无锁分配；Page Cache 是最底层的“大块内存供货商”，负责向操作系统申请和释放大页内存。而 Central Cache 就是夹在中间的“调度中心”——它一方面接收 Thread Cache 批量申请内存块的请求，另一方面从 Page Cache 获取大块内存页并切割成小块，同时还负责在 Thread Cache 释放过多内存时回收回来。Central Cache 是所有线程共享的，所以它的操作需要加锁，但因为 Thread Cache 只有在本地缓存完全耗尽时才会来找它，而且一次会批量拿走一批，所以锁竞争并不激烈，这是 TCMalloc 高效的核心设计之一。

   Central Cache 的结构和 Thread Cache 是对应的——它也有一个 208 个桶的哈希数组 _spanLists[NFREELIST]，每个桶对应一种尺寸。但桶里挂的不是一个个单块内存，而是 Span 对象。Span 是 Page Cache 和 Central Cache 之间的桥梁，它管理着一块连续的内存页（比如一页或多页），并且内部维护着一个自由链表 _freeList，把这块大内存切成了一个个固定尺寸的小块挂上去。所以 Central Cache 的每个桶里，挂的是多个 Span，而每个 Span 内部又挂着一串同尺寸的内存块。

//这是ThreadCache.hpp文件
void* ThreadCache::FetchFromCentralCache(size_t index, size_t size)
{
	//慢开始反馈调节算法
	//最开始不会一次向 Central Cache 批量要太多，多了用不完
	//如果不要size大小的内存需求，那 batchNum 就会不断增长，直到上限
	size_t batchNum = std::min(_freeLists[index].MaxSize(), SizeClass::NumMoveSize(size));
	
	void* start = nullptr;
	void* end = nullptr;
	size_t actualNum = CentralCache::GetInstance()->FetchRangeObj(start,end,batchNum,size);

	assert(actualNum > 1);

	if (actualNum == 1)
	{
		assert(start == end);
		return start;
	}
	else
	{
		_freeLists[index].PushRange(NextObj(start), end);
		return start;
	}

	if (_freeLists[index].MaxSize() == batchNum)
	{
		_freeLists[index].MaxSize() += 1;
	}
	
	return nullptr;
}

  我们主要先来看一下 FetchRangeObj 这个函数的代码：

size_t  CentralCache::FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size)
{
	size_t index = SizeClass::Index(size);

	_spanLists[index]._mtx.lock();

	//从span中获取 batchNum 个对象
	//如果不够 batchNum 个，有多少给多少
	Span* span = GetOneSpan(_spanLists[index], size);  //??

	assert(span);
	assert(span->_freeList);

	end = start;

	size_t i = 0;
	size_t actualNum = 1;
	while (i < batchNum - 1 && NextObj(end) != nullptr)
	{
		end = NextObj(end);
		++i;
		++actualNum;
	}

	span->_freeList = NextObj(end);
	NextObj(end) = nullptr;

	_spanLists[index]._mtx.unlock();
	return actualNum;
}

  它在函数一开始先通过 SizeClass::Index(size) 算出对应的桶下标，然后给对应的 _spanLists[index] 加锁，因为 Central Cache 是全局共享的，操作它必须保证线程安全。接着调用 GetOneSpan 从该桶中获取一个可用的 Span——这行代码的作用就是“拿到一个有货的 Span 对象”，这个 Span 的 _freeList 指向了第一个空闲块。拿到 Span 之后，FetchRangeObj 就从 span->_freeList 开始，沿着内存块的前 8 个字节（就是链表指针 NextObj）一步一步往后走，一直走到取了 batchNum - 1 步或者走到链表尾部为止，然后用 start 指向第一个块，end 指向最后一个块，actualNum 记录实际取到了多少块。取完之后，它把 span->_freeList 更新为 NextObj(end)，也就是指向剩下还没取走的块的头部，再把 NextObj(end) 置成 nullptr，把这一串取出来的块从原来的链表上断下来。最后解锁，把 actualNum 返回给调用者。

  其中的 GetOneSpan 就是用来“拿到一个有货的 Span”的。如果当前桶的 Span 链表里第一个 Span 的 _freeList 不为空，那就直接返回这个 Span，说明它手里还有空闲块可以分。但如果第一个 Span 的 _freeList 是空的，说明这个 Span 里所有的内存块都已经被分配出去了，还没回收回来，那就得往后找，或者干脆向 Page Cache 要一个新的 Span。这部分的逻辑我们暂时还没写全，但它的本质就是：在当前桶的 Span 链表中找到一个有空闲块的 Span，如果都找不到了，就去 Page Cache 申请新的页并切分成 Span。

  因此，FetchRangeObj 函数，它的核心职责就是：从 Central Cache 的某个桶里，取出 batchNum 个内存块，用 start 和 end 两个指针把这一串块的头尾返回给 Thread Cache，相当于批量切出一串连续的空闲块。 

  下面是另外涉及到的函数代码：

	size_t& MaxSize()
	{
		return _maxSize;
	}


	//一次thread cache从中心缓存获取多少个
	static size_t NumMoveSize(size_t size)
	{
		assert(size > 0);

		if (size == 0)
			return 0;

		//[2,512]一次批量移动多少个对象的（慢启动）范围
		//小对象一次批量上限高
		//大对象一次批量上限低
		int num = MAX_BYTES / size;
		if (num < 2)
			num = 2;
		if (num > 512)
			num = 512;
		return num;
	}

	static CentralCache* GetInstance()  //CentralCache 这个类的唯一化身
	{
		return &_sInst;
	}




	void PushRange(void* start, void* end)
	{
		NextObj(end) = _freeList;
		_freeList = start;
	}

  现在再回过头来看 Thread Cache 里的 FetchFromCentralCache 函数。它拿到 FetchRangeObj 返回的 start、end 和 actualNum 之后，要决定怎么处理这一批内存块。actualNum 是 Central Cache 实际给你的块数，它可能少于我们申请的 batchNum，因为 Central Cache 不一定有那么多的存货。代码里断言 actualNum > 1，这表示假设每次申请至少能拿到两块。如果 actualNum == 1，说明 Central Cache 这次只给了一块，那也没办法，直接把 start 返回给用户就行，当前线程的桶里没有剩余缓存。但更常见的情况是 actualNum > 1，这时就要做两件事：第一，把拿到的这批内存块从第二块开始到最后一块，全部挂到当前线程的 _freeLists[index] 链表上，作为备用缓存；第二，把第一块 start 直接返回给用户。那为什么从第二块开始挂？因为 start 是要返回给用户去用的，不能放到缓存里；剩下的 NextObj(start) 到 end 这一串才是空闲的，留着下次分配时再用。代码里调用的 PushRange(NextObj(start), end) 就是干这个的——它把 end 的 next 指针指向当前链表的头，然后把链表头更新为 NextObj(start)，就把这一整串一次性挂到 Thread Cache 的桶里了。

  最后是那个 if (_freeLists[index].MaxSize() == batchNum) 对 MaxSize 的调整。这个操作我们放到了整个函数的最后，但实际逻辑上它应该在 FetchRangeObj 调用之后、return 之前执行。它的含义是：如果这一次实际申请的批量数等于 NumMoveSize 算出来的上限，说明 Thread Cache 的需求量很大，Central Cache 给足了量，那下一次就可以再多要一块，所以 _maxSize += 1。反之，如果 Central Cache 这次根本没给够，那 _maxSize 就不做调整，维持原样，避免盲目增加导致反复申请失败。这就是“慢启动反馈调节”——用量决定增长，缺量决定停止。

4.3 Page Cache

  对于申请内存来说，当 central cache 向 page cache 申请内存时，page cache 先检查对应位置有没有 span，如果没有则向更大页寻找一个 span，如果找到则分裂成两个。比如：申请的是 4 页 page，4 页 page 后面没有挂 span，则向后面寻找更大的 span，假设在 10 页 page 位置找到一个 span，则将 10 页 page span 分裂为一个 4 页 page span 和一个 6 页 page span。

  如果找到_spanList [128] 都没有合适的 span，则向系统使用 mmap、brk 或者是 VirtualAlloc 等方式申请 128 页 page span 挂在自由链表中，再重复 1 中的过程。

  需要注意的是 central cache 和 page cache 的核心结构都是 spanlist 的哈希桶，但是他们是有本质区别的，central cache 中哈希桶，是按跟 thread cache 一样的大小对齐关系映射的，他的 spanlist 中挂的 span 中的内存都被按映射关系切好链接成小块内存的自由链表。而 page cache 中的 spanlist 则是按下标桶号映射的，也就是说第 i 号桶中挂的 span 都是 i 页内存。

  我们接下来来看代码，我们现在的首要任务是要先明白 Central Cache 里的 GetOneSpan 在什么时机被调用。当 Thread Cache 的某个桶没货了，它会调用 Central Cache 的 FetchRangeObj，而 FetchRangeObj 内部第一件事就是调用 GetOneSpan 去获取一个有可用空闲块的 Span。所以 GetOneSpan 的核心职责是：在 Central Cache 的某个桶里找到一个非空的 Span 返回，如果找不到，就去 Page Cache 申请新的内存页并切分成 Span。

  下面来看代码：

Span* CentralCache::GetOneSpan(SpanList& list, size_t size)
{
	//遍历spanlist，查看是否还有空闲的span
	Span* it = list.Begin();
	while (it != list.End())
	{
		if (it->_freeList != nullptr)
		{
			return it;
		}
		else
		{
			it = it->_next;
		}
	}

	//先把Central Cache的桶锁解掉
	//这样如果其他线程释放内存对象回来，不会阻塞
	list._mtx.unlock();

	//走到这里说明没有空闲span，只能找Page Cache
	PageCache::GetInstance()->_pageMtx.lock();
	Span* span =  PageCache::GetInstance()->NewSpan(SizeClass::NumMovePage(size)); 
	PageCache::GetInstance()->_pageMtx.unlock();
	
	//下面的部分是对获取到的Span进行切分，不涉及竞争，故不用加锁
	//这里为什么要用 char* 类型而非 void* 类型
	char* start = (char*)(span->_pageId << PAGE_SHIFT); //找到内存页的起始地址
	size_t bytes = span->_n << PAGE_SHIFT; //该操作等于 * 8K，目的是计算该内存块的大小（字节数） 
	char* end = start + bytes;

	//将大块内存切分，挂到链表中去
	span->_freeList = start;
	start += size;  //??
	void* tail = span->_freeList;

	while (start < end)
	{
		NextObj(tail) = start;
		tail = NextObj(tail);
		start += size;
	}

    NextObj(tail) = nullptr;

	//切好span后，要把切好的span挂到桶里，防止竞争桶，故加锁
	list._mtx.lock();
	list.PushFront(span);
	//后面FetchRangeObj函数中再解锁

	return span;
}

  现在我们从头看 GetOneSpan 的代码。它一开始遍历当前桶的 Span 链表，检查每个 Span 的 _freeList 是否为空。如果不为空，说明这个 Span 手里还有空闲块，直接返回它，这是最快路径。但如果遍历完整个链表发现所有 Span 的 _freeList 都是空的，那就意味着当前桶里所有的 Span 都已经把内存块全部分配出去了，还没有回收回来，这时候就必须向 Page Cache 要新的内存。

  在去 Page Cache 之前，它做了一件非常重要的事情：list._mtx.unlock()。这里就涉及第一个问题了——为什么要解锁，而且为什么用的是 list._mtx 而不是 _pageMtx。你要明白，Central Cache 是全局共享的，每个桶都有自己的锁（桶锁），而 Page Cache 也有一把全局锁。当前函数进来时，Central Cache 的桶锁是处于加锁状态的（因为 FetchRangeObj 在调用 GetOneSpan 之前已经锁住了 _spanLists[index]）。如果直接拿着这把锁去调用 Page Cache 的 NewSpan，而 NewSpan 可能要向操作系统申请内存，这个过程可能会很慢，期间其他线程想往这个桶里释放内存块时就会被这把锁堵住，导致本该无阻塞的释放操作被卡死。所以这里先把 Central Cache 的桶锁解开，让其他线程可以正常往这个桶里归还内存块，然后再去拿 Page Cache 的全局锁，从 Page Cache 申请新的 Span。这体现了 TCMalloc 设计中的一个核心理念：锁的粒度要尽量细，持有锁的时间要尽量短，能不持有锁的时候绝不留着锁不放。而 list._mtx 和 _pageMtx 的区别就在于：前者是 Central Cache 每个桶自己的锁，保护的是单个桶的 Span 链表；后者是 Page Cache 的全局锁，保护的是 Page Cache 内部所有桶的 Span 链表以及页的分配和回收。两者作用域不同，一个负责 Central Cache 层面的并发安全，一个负责 Page Cache 层面的并发安全。

  拿到 Page Cache 的锁之后，调用：

//GetOneSpan(SpanList& list, size_t size) 函数

PageCache::GetInstance()->NewSpan(SizeClass::NumMovePage(size))

  这里传入的参数是 NumMovePage(size)，我们先看这个函数在干什么：

	//计算一次向系统获取几个页
	//单个对象 8 byte
	//...
	//单个对象 256 byte

    static const size_t PAGE_SHIFT = 13;

	static size_t NumMovePage(size_t size)
	{
		size_t num = NumMoveSize(size);
		size_t npage = num * size;

		npage >>= PAGE_SHIFT;
		if (npage == 0)
			npage = 1;

		return npage;
	}

  NumMovePage 的逻辑是：先调用 NumMoveSize(size) 拿到一次批量移动的块数，然后用 num * size 算出这批内存块总共占多少字节，再右移 PAGE_SHIFT（13 位）把这个字节数转换成页数。

  >>是右移运算符：把数字的二进制位向右移动 13 位，左边补 0（无符号数），我们用一个例子来说明：

  因此右移的操作 >> ，实际上相当于 10000 去除以 2 的 13 次方后取整，2 的13 次方是 8192 ，做除法运算后是 1.22... ，取整就是 1 。同样的如果是左移的操作 << ，相当于是乘以 2 的 13 次方。

   npage >>= PAGE_SHIFT 就是 npage = npage >> PAGE_SHIFT ，等价于 npage / 8192，这里将PAGE_SHIFT的值定为 13 ，也正是因为一个页的大小就是 8 * 1024 字节 = 8 K，如果算出来小于 1 页就至少给 1 页。这个函数的作用就是：根据对象的大小，计算出 Thread Cache 一次批量申请所对应的总字节数需要多少页来承载。比如申请 8 字节，NumMoveSize(8) 返回 512 块，512 * 8 = 4096 字节，右移 13 位得 0，所以至少返回 1 页。

  在这里大家也可能会疑惑：为什么求页数要用右移 13 位而不是直接去除以 8192？因为位运算比除法快得多，在内存池这种性能敏感的场景中，能用移位就不用除法。

 拿到 Page Cache 返回的 Span 之后，GetOneSpan 把 Page Cache 的锁解开。接下来的操作是把这块 Span 切分成一个个固定尺寸的小内存块，挂到 Span 的 _freeList 上。这就涉及第二个细节问题了——为什么用页的起始地址的类型用 char* 而不是 void*？因为 char* 在 C++ 中支持指针算术运算，start += size 可以按字节移动指针，而 void* 不支持直接做加法。

// GetOneSpan 函数

char* start = (char*)(span->_pageId << PAGE_SHIFT); //找到内存页的起始地址
size_t bytes = span->_n << PAGE_SHIFT; //该操作等于 * 8K，目的是计算该内存块的大小（字节数） 
char* end = start + bytes;

  span->_pageId 存储的是这个 Span 在虚拟内存空间里的“起始页编号”。要把页编号转成实际的内存地址，就得左移 13 位——因为每一页是 8192 字节，即 8 KB，所以第 0 页的地址是 0，第 1 页的地址是 8192，第 2 页的地址是 16384，以此类推。_pageId << 13 算出来的就是页编号对应的起始内存地址。

  接下来是整段代码里最核心的切分逻辑。span->_freeList = start 把链表的头指向得到的页的起始地址。然后 start += size 把 start 向后移动 size 个字节，指向第二个块。void* tail = span->_freeList 让 tail 指向当前链表的“尾部”，因为此时span->_freeLists这个自由链表中刚刚被插入了一个内存块，此时 tail 就是第一个块，所以既是头也是尾。然后进入 while 循环：NextObj(tail) = start 把前一个块的 next 指针指向当前块，tail = NextObj(tail) 把 tail 移动到当前块，start += size 把 start 移动到下一块。循环直到 start < end 不成立为止，也就是 start 越过了整块 Span 的末尾。这个逻辑的本质就是：从头到尾遍历这段连续的大内存，把每一块的前 8 个字节（64 位下）当作 next 指针，串成一个单向链表。最后一块的 next 需要显式置 nullptr，因为 span->_freeList 在初始化时是 nullptr，而内存块刚申请回来时里面是随机值，因为 malloc（或者说这里的 VirtualAlloc）返回的内存，操作系统不会帮你清零（除非你指定了特殊标志）。在 Windows 下，新申请的堆内存往往被填充为 0xCD（调试态下的“脏数据”），在 Linux 下则可能是全零或者残留的其他数据。如果你不把最后一块的 next 置成 nullptr，那么 tail 的 next 指针里存的就是一个随机地址，就有可能引发越界访问。

  切完块之后，要把这个切好的 Span 挂回 Central Cache 的桶里。因为此时 Central Cache 的桶锁是解开的，而马上要操作 list.PushFront(span)，这是对共享资源的修改，所以必须重新加锁。注意这里加锁用的是 list._mtx.lock()，和之前解锁的是同一把锁，这就回到了第一个问题——加锁和解锁是一对操作，作用域内保护的是同一个共享资源。加锁之后把 Span 头插到桶里，然后返回这个 Span。注意这里返回时没有解锁，因为锁会在调用者 FetchRangeObj 函数中解锁，它在那里面调用了 _spanLists[index]._mtx.unlock()，这就保证了锁的对称性。

  现在来看 NewSpan 函数，这是 Page Cache 最核心的逻辑。Page Cache 维护的也是哈希桶结构，但桶的数量是 NPAGES（129 个），下标代表页数（0~128），每个桶里挂的是页数相同的 Span。按照我们开头说的申请内存时分割大块 Span 的理念，NewSpan 的职责是：从 Page Cache 中分配一个 k 页的 Span 返回给 Central Cache。  它先检查第 k 个桶有没有空闲的 Span，如果有直接弹出来返回，这是最快路径。如果没有，就往后找更大的桶，从 i = k + 1 开始，直到 NPAGES - 1。为什么往后找？因为第 k 个桶没有 k 页的 Span，但更大的桶里可能有更大的 Span，先 nSpan = _spanLists[i].PopFront() 把这个大 Span 拿下来，然后 kSpan = new Span 创建一个新的 Span 对象，让 kSpan->_pageId = nSpan->_pageId 指向这个大 Span 的起始页，kSpan->_n = k 表示它只有 k 页。接着把 nSpan->_pageId += k，让 nSpan 的起始页后移 k 页，nSpan->_n -= k 让它的页数减少 k，然后把剩下的 nSpan 挂到 _spanLists[nSpan->_n] 这个桶里。这就完成了“切一刀”的操作——大的切成小的和剩下的，小的给 Central Cache，剩下的存回 Page Cache。

  对于NewSpan函数如果感觉一团雾水的话，可以先想象一下，在物理内存的视角里，nSpan 代表的是什么？它代表的是 一段连续的物理页区间。nSpan->_pageId 是这段区间的起始页编号，nSpan->_n 是这段区间总共有多少页。比如 nSpan 的 _pageId = 100，_n = 128，那它管理的就是物理内存中第 100 页到第 227 页这一段连续的 128 页。

  现在 Central Cache 想要从这段区间里拿走 前 k 页（比如 k=2）。那被拿走的这 2 页，就是第 100 页和第 101 页。剩下的那部分，就变成了第 102 页到第 227 页，一共 126 页。

  现在问题来了：原来的 nSpan 对象，它记录的仍然是 _pageId = 100，_n = 128。但你实际要用它来表示剩下的那段 126 页的连续内存时，它的起始页应该变成 100 + 2 = 102，页数应该变成 128 - 2 = 126。如果你不改这两个字段，那 nSpan 记录的起始页还是 100，页数还是 128，这就等于它声称自己管着第 100 到 227 页，但事实上第 100 和 101 页已经被切走给别人了——如果你不改它，下次有人想用这个 nSpan 的时候，就会错误地认为第 100 页和 101 页还是空闲的，就会发生同一块内存被两次分配出去的严重错误。

  所以那两个操作——nSpan->_pageId += k 和 nSpan->_n -= k——干的活就是：让 nSpan 这个对象的元数据，准确地反映它“被切掉一块之后，现在实际管理的是哪一段内存”。相当于你有一根 128 米长的绳子，从最左边剪掉 2 米给人家，那这根绳子剩下部分的起点就不再是原来的 0 米处了，得往右移 2 米，长度也从 128 米变成 126 米。你必须更新绳子的“起点”和“长度”这两个记录，否则以后谁拿着这个记录去用这根绳子，就会用错位置。

  那为什么还要创建 kSpan 这个新对象呢？因为被切走的 2 页也需要一个“管理者”——Central Cache 需要用一个 Span 对象来管理这 2 页的分配和回收状态，所以你得 new 一个全新的 kSpan 对象，把它的起始页设为原来的 nSpan->_pageId（就是第 100 页），页数设为 2，然后把这个 kSpan 返回给 Central Cache。这样一来，剩下的 126 页由原来的 nSpan 对象管理（但它的起始页和页数已经被更新过了），被切走的 2 页由新的 kSpan 对象管理。两个 Span 对象各自管着各自连续的内存段，互不重叠。再把剩下的 nSpan 挂到 _spanLists[nSpan->_n] 这个桶里。最后返回 kSpan。这个过程叫做“大 Span 切小 Span”，它保证了页的利用率，不会因为申请小页数就浪费大块连续内存。

  如果 Page Cache 里所有桶都为空，也就是没有任何空闲的 Span 了，那就直接向操作系统申请内存。注意这里申请的是 NPAGES - 1 页，也就是 128 页，约 1MB。

inline static void* SystemAlloc(size_t kpage)
{
#ifdef _WIN32
	void* ptr = VirtualAlloc(0, kpage << 13, MEM_COMMIT | MEM_RESERVE, PAGE_READWRITE);
#else
	// linux下brk、mmap等
#endif
	if (ptr == nullptr)
	{
		throw std::bad_alloc();
	}
	return ptr;
}

  SystemAlloc(NPAGES - 1) 通过 VirtualAlloc（Windows 下）向系统申请内存，返回的是内存的起始地址。bigSpan->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT 把地址右移 13 位转成页号，bigSpan->_n = NPAGES - 1，然后把这块巨大的 Span 挂到对应的桶里，再递归调用 NewSpan(k) 重新走一遍分配流程，此时第 k 个桶仍然没有，但后面的桶里有这个 128 页的大 Span 了，所以第二次调用时就会触发切分逻辑，把 128 页切成 k 页和 128-k 页，返回 k 页的 Span 给 Central Cache。这就是 Page Cache 的完整分配流程。

4.4 内存回收

4.4.1 Thread Cache内存回收

  我们首先要清楚，Thread Cache 里每个桶的 _freeList 不只存着空闲块，它还记录着当前这个链表上有多少块，我们在FreeList这个类当中再维护一个新的变量 _size。当我们将 Deallocate 里把一块内存 Push 回链表之后，_size 自增 1，这个设计本质上就是用空间换时间——多维护一个计数器，换来每次释放时 O(1) 的长度判断。然后判断：如果 _size >= _maxSize，说明这个桶里囤积的空闲块数量已经超过了“一次批量申请的上限”，意味着当前线程占用了过多本该可以被其他线程使用的内存，所以触发 ListTooLong，把 _maxSize 个空闲块一口气还给 Central Cache。

  这个阈值设计就是之前使用过的慢启动机制的反向运用——_maxSize 控制着“一次拿多少块”和“攒到多少块就还回去”的平衡。当 _size 刚超过 _maxSize 时触发回收，回收的就是 _maxSize 个块，这样每次回收的大小和每次申请的大小是对称的，逻辑一致，比较直接、简单。而对于真正的TCMalloc来说，它考虑的情况就很复杂，代码量也非常多，我们这里主要参考其最核心的部分。

//当一个对象不再被程序使用时，把它归还到当前线程自己的空闲链表里，等着下次分配时再次复用
//内存是释放给 Thread Cache 缓存
void* ThreadCache::Deallocate(void* ptr, size_t size) 
{
	assert(ptr);
	assert(size <= MAX_BYTES);
	
	size_t index = SizeClass::Index(size);
	_freeLists[index].Push(ptr);
	
	//当链表长度大于一次批量申请的内存时就开始释放，还一段list给Central Cache
	if (_freeLists[index].Size() >= _freeLists[index].MaxSize())
	{
		ListTooLong(_freeLists[index], size);
	}
}

void ThreadCache::ListTooLong(FreeList& list, size_t size)
{
	void* start = nullptr;
	void* end = nullptr;
	list.PopRange(start, end, list.MaxSize());


	//为什么这个函数不用设计传参给 end ？？
	CentralCache::GetInstance()->ReleaseListToSpans(start, size);
}

	void PopRange(void*& start, void*& end, size_t n)
	{
		assert(n >= _size);

		start = _freeList;
        end = start;

		for (size_t i = 0; i < n - 1; ++i)
		{
			end = NextObj(end);
		}

		_freeList = NextObj(end);
		NextObj(end) = nullptr;
		_size -= n;
	}

  这个 PopRange 的作用是从链表头部摘下来 n 个节点，用 start 指向第一个，end 指向最后一个，然后把链表头移动到剩余节点的位置。PopRange 的实现逻辑就是，先把 start = _freeList，然后通过循环让 end 沿着 NextObj 往后走 n-1 步，走到这一串的最后一个节点，然后 _freeList = NextObj(end) 让链表头跳过这一串，NextObj(end) = nullptr 把这一串的尾部切断，最后 _size -= n 更新计数器。

  另外大家思考一个问题，为什么 ReleaseListToSpans 这个函数只需要传 start 和 size，而不需要额外再传一个 end？因为 start 指向这一串内存块的头部，而这一串内存块是通过链表指针串联起来的——每个块的前 8 个字节（64 位下）存的是下一个块的地址。所以 Central Cache 拿着 start，完全可以沿着 NextObj 指针一路走下去，直到碰到 nullptr 就知道这一串到哪里结束了。它不需要你额外告诉它 end 在哪里，因为链表本身的指针结构已经把这个信息编码进去了。但这里有一个前提：PopRange 必须确保取出来的这一串的最后一个块的 next 被置为 nullptr。 ReleaseListToSpans 只需要 start 就能完整遍历这一整串链表。至于 size 参数，传的是这个内存块本身的字节大小（比如 8 字节、16 字节），它的作用是让 Central Cache 知道这一串内存块属于哪个桶，以便正确地归还到对应的 Span 里。也就是说，start 告诉 Central Cache“内存块的起始地址”，size 告诉它“这些内存块属于哪个尺寸类别”，两者配合就能完成归还操作。

  这段代码里面最核心的函数是 ReleaseListToSpans 函数，我们将在Central Cache的内容中完善它。

4.4.2 Central Cache内存回收

  前面提到，当 Thread Cache 的某个桶里积压的空闲块数量超过了 _maxSize，它会调用 ListTooLong，从链表中摘下一批节点，然后调用 CentralCache::ReleaseListToSpans(start, size)，把这一串内存块整体归还给 Central Cache。start 指向这一串的第一个块，size 是每个块的大小。Central Cache 拿到这一串之后，要把它们逐个归还到对应的 Span 中。

  然后大家思考一个问题，该怎么知道从Thread Cache回收上来的内存块，原来都属于Central Cache的哪个页？我们可以用这个方法：因为一个页的大小是 8 K，比如内存块 1 和 2 ，因为处于第 2001 个页当中，那么当我使用内存块1 的地址去整除以 8 K的时候，得到的数字一定是 2000 ，这就可以判定位置。

  我们现在NewSpan函数里面做了一些调整：

//建立id和span的映射，方便central cache回收小块内存时，查找对应的页
for (PAGE_ID i = 0; i < kSpan->_n; ++i)
{
	_idSpanMap[kSpan->_pageId + i] = kSpan;
}


//Page Cache 类
std::unordered_map<PAGE_ID, Span*> _idSpanMap;

  这个 for 循环的判断条件是 i < kSpan->_n，其中 kSpan->_n 是当前这个 Span 占用的总页数。kSpan->_pageId + i 算的是这个 Span 所管理的每一页的页号——kSpan->_pageId 是起始页号，加 i 表示第 i 页的页号。而 _idSpanMap[kSpan->_pageId + i] = kSpan 的意思是：把从起始页号开始的每一页，都在一个全局的哈希映射表里建立一条记录，键是页号，值是这个 kSpan 的指针。为什么要把每一页都映射到同一个 kSpan？因为一个 Span 管理的是连续的若干页，这些页物理上相邻且都属于同一个 Span。当 Thread Cache 把某个内存块归还回来时，你只能拿到内存块的地址，通过右移 13 位算出页号，即除以一个页的大小 8KB，然后去 _idSpanMap 里查询。无论这个内存块落在该 Span 管理的哪一页上，查出来的结果都应该是同一个 Span 指针。所以必须把该 Span 占用的每一页都插入映射表，且都指向同一个 Span 对象。

  下面来看代码：

void CentralCache::ReleaseListToSpans(void* start, size_t size)
{
	size_t index = SizeClass::Index(size);
	_spanLists[index]._mtx.lock();

	while (start)
	{
		void* next = NextObj(start);
	
		Span* span = PageCache::GetInstance()->MapObjectToSpan(start);
		//？？？
		NextObj(start) = span->_freeList;
		span->_freeList = start;

		span->_useCount--;
		//说明span切分出去的小块内存都回来了
		//就可以回收给page cache
		if (span->_useCount == 0)
		{
			_spanLists[index].Erase(span); 

			span->_freeList = nullptr;
			span->_next = nullptr;
			span->_prev = nullptr;

			_spanLists[index]._mtx.unlock();  //解除桶锁

			PageCache::GetInstance()->_pageMtx.lock();  //加上页锁
			PageCache::GetInstance()->ReleadseSpanToPageCache(span); //释放页
			PageCache::GetInstance()->_pageMtx.unlock();//解除页锁
			
			_spanLists[index]._mtx.lock();  //加上桶锁
		}
		start = next;
	}
	_spanLists[index]._mtx.unlock();
}

Span* PageCache::MapObjectToSpan(void* obj)
{
	PAGE_ID id = (PAGE_ID)obj >> PAGE_SHIFT;
	auto ret = _idSpanMap.find(id);
	if (ret != _idSpanMap.end())
	{
		return ret->second;  //???
	}
	else
	{
		assert(false);
		return nullptr;
	}
}

  上面的映射关系建立完成之后，后面的每一个内存块相当于都有了编号。因为 _idSpanMap 是一个 std::unordered_map<PAGE_ID, Span*>，它的 find 函数返回的是一个迭代器，指向一个 std::pair<const PAGE_ID, Span*> 类型的键值对，也就是相当于此时的ret指向的就是pair<const PAGE_ID, Span*>这个键值对。

  在这个 pair 里，first 是键（页号），second 是值（指向 Span 的指针）。所以 ret->second 就是取到那个 Span*，这就是为什么它能返回 Span*。你不需要额外存储 end，就是因为 MapObjectToSpan 能通过内存地址反查出它所属的 Span，而 size 只是用来定位 Central Cache 的桶而已，Span 里本身已经记录了它的页数和起始页号，有了这些信息，回收操作就能精准到位。

  因此，MapObjectToSpan就通过哈希表映射的关系，经过位运算，再通过find函数找到该内存块对应的正确的span，然后返回，在ReleaseListToSpans函数当中由Span* span这个变量去接收，再将从Threa Cache中回收来的内存块依次对应返还。

  我们再来关注一下这里的返还逻辑：

  这里大家一定不要搞混了，因为我当时在这里就犯错误，浪费了很多时间，NextObj(start) 不是“下一块内存”，而是 “当前这块内存的前 8 个字节（指针槽位）”。

  我们来看它的定义：

static void*& NextObj(void* obj)
{
    return *(void**)obj;
}

  因此，NextObj(start) 就是 start 这块内存里用来存“下一个地址”的那个变量本身。我们不是在移动 start，而是在修改 start 内部存着的那个指针值。

  而 NextObj(start) = span->_freeList 并不是把“下一块内存”放到 Span 链表上，而是让 start 这个节点，抛弃它原来的下一个节点（即 Thread Cache 传给它的那一串的后半截），转而指向 Span 空闲链表的头部。

  我们用画图的方式走一遍：

  假设 Thread Cache 传过来的这一串是：block1 -> block2 -> block3 -> nullptr。start 是该链表的指针，指向 block1，故 NextObj(start) 就相当于 block1 的 next 。假设 Span 原来的空闲链表是：spanFree1 -> spanFree2 -> nullptr，span->_freeList 指向 spanFree1。

  执行第一行：NextObj(start) = span->_freeList;
  等价于 block1 的 next 指针（即前 8 字节）被修改，改写成 spanFree1 的地址。
  此时 block1 的状态变成了：block1 -> spanFree1 -> spanFree2 -> nullptr。
  此时 block2 还好端端地存在于内存中，只是 block1 不再指向它了，但在进入循环之前，已经执行了 void* next = NextObj(start)，此时 next 变量里存的正是 block2 的地址。block1 指向谁变了，完全不影响 next 这个局部变量。但是span->_freeLists指向的还是 spanFree1。

  执行第二行：span->_freeList = start；因为刚刚start的指向还是没变，还是 block 1  。
  所以这个操作就把 Span 的链表头指向 block1。
  此时 Span 的链表变成了：block1 -> spanFree1 -> spanFree2 -> nullptr。此时span->_freeLists指向的是 block1。

  然后执行循环末尾：start = next;
  start 被更新为 block2，进入下一轮循环，去处理 block2 的回收。

  因为每一次回收的时候，span链表中维护的 _usecount 都会减少，当减少到 0 的时候，就触发将该 span 回收给Page Cache的操作，进入 if 条件语句，这部分内容的核心就是ReleadseSpanToPageCache(span)函数，我们在Page Cache内存回收的部分会讲解。另外要注意的是，因为回收Span到Page Cache涉及锁的切换，当前我们持有的是 Central Cache 的桶锁（_spanLists[index]._mtx），而操作 Page Cache 需要持有 Page Cache 的全局锁（_pageMtx）。如果你持有桶锁的同时去拿页锁，可能造成死锁——比如 Page Cache 在某个操作里反过来要拿同一个桶锁，就会互相等待。所以这里先解桶锁，再拿页锁，然后调用ReleaseSpanToPageCache把整个 Span 归还给 Page Cache，再解页锁，最后重新把桶锁锁上，继续处理下一个内存块。这一套锁的切换，体现的是内存池设计中最精细的锁粒度控制：能不持有锁的时候绝不持有，能缩小锁作用域就尽量缩小。

4.4.3 Page Cache内存回收

    对于释放内存来说，如果 central cache 释放回一个 span，则依次寻找 span 的前后 page id 的没有在使用的空闲 span，看是否可以合并，如果合并继续向前寻找。这样就可以将切小的内存合并收缩成大的 span，减少内存碎片。

  就像这样，我们直接展示代码：

void PageCache::ReleaseSpanToPageCache(Span* span)
{
	//对sapn前后页尝试进行合并，缓解内存碎片问题
	//先向前合并
	while (1)
	{
		//合并时，找当前页的前一页，看是否有空闲页可以合并
		PAGE_ID prevId = span->_pageId - 1;
		auto ret = _idSpanMap.find(prevId);
		//前面相邻页的Span没有了，不合并
		if (ret == _idSpanMap.end())
		{
			break;
		}
		//前面相邻页的Span正在使用，不合并
		Span* prevSpan = ret->second;
		if (prevSpan ->_isUse == true)
		{
			break;
		}
		//前面页的页数加上当前页的页数，大于Page Cache能承受的最大 128 页，也不合并
		if (prevSpan->_n + span->_n > NPAGES - 1)
		{
			break;
		}

        for (PAGE_ID i = 0; i < prevSpan->_n; ++i)
        {
            _idSpanMap.erase(prevSpan->_pageId + i);
        }

		span->_pageId = prevSpan->_pageId;
		span->_n += prevSpan->_n;

		_spanLists[prevSpan->_n].Erase(prevSpan);
		delete prevSpan;
	}

	//向后合并
	while (1)
	{
		PAGE_ID nextId = span->_pageId + span->_n;
		auto ret = _idSpanMap.find(nextId);

		if (ret == _idSpanMap.end())
		{
			break;
		}

		Span* nextSpan = ret->second;
		if (nextSpan->_isUse == true)
		{
			break;
		}

		if (nextSpan->_n + span->_n > NPAGES - 1)
		{
			break;
		}

        for (PAGE_ID i = 0; i < nextSpan->_n; ++i)
        {
            _idSpanMap.erase(nextSpan->_pageId + i);
        }

		span->_n += nextSpan->_n;

		_spanLists[nextSpan->_n].Erase(nextSpan);
		delete nextSpan;
	}

	//合并后要重新挂起

	_spanLists[span->_n].PushFront(span);

	span->_isUse = false;
	_idSpanMap[span->_pageId] = span;
	_idSpanMap[span->_pageId + span->_n - 1] = span;
}

  我们先看向前合并的 while 循环。它每次先计算 prevId = span->_pageId - 1，也就是当前 Span 起始页的上一页的页号。然后去 _idSpanMap 里查这个页号有没有记录——_idSpanMap 里记录的是每个页号所属的 Span 指针。如果查不到，说明当前 Span 前面根本没有内存页了（比如已经到了地址空间的边界），那就不合并，直接跳出循环。如果查到了，拿到 prevSpan，检查它的 _isUse 字段。_isUse 为 true 表示这个 Span 正在被 Central Cache 使用，里面的内存块还没有全部回收回来，绝对不能动它，所以也要跳出循环。再检查一个上限条件：prevSpan->_n + span->_n > NPAGES - 1。NPAGES - 1 是最大页数 128。如果两个 Span 合并后超过 128 页，那这个合并后的 Span 就超出了 Page Cache 能管理的最大页数，挂了也没有对应的桶可以放，所以也不合并，直接跳出循环。

  如果前面三个条件都通过了，说明 prevSpan 是一个空闲的、可以合并的邻居。此时执行 span->_pageId = prevSpan->_pageId，把当前 Span 的起始页号往前挪到 prevSpan 的起始页；然后 span->_n += prevSpan->_n，把当前 Span 的页数累加，等于合并了两个 Span 的总页数。然后从 _spanLists[prevSpan->_n] 中把 prevSpan 摘掉，delete 掉这个已经没有任何作用的 Span 对象。合并完成之后，循环继续，重复同样的逻辑——因为合并之后，新的 Span 的起始页又往前移了，它前面可能还有空闲页，所以继续尝试向前合并，直到无法合并为止。

  向前合并跑完之后，紧接着是向后合并的 while 循环。逻辑和向前合并完全对称。

  前后两个 while 循环都结束之后，当前 Span 已经完成了所有可能的合并，变成了一个尽可能大的连续空闲块。接下来要做的是把它挂到 Page Cache 对应的桶里。_spanLists[span->_n].PushFront(span) 根据合并后的页数，把它挂到下标等于页数的那一个桶里。然后把 span->_isUse = false，标记为空闲状态，因为这是 Page Cache 自己的空余内存，Central Cache 以后需要时可以再次申请使用。

//NewSpan函数

//存储nSpan首尾页号和nSpan映射
_idSpanMap[nSpan->_pageId] = nSpan;

_idSpanMap[nSpan->_pageId + nSpan->_n - 1] = nSpan;

  我们在NewSpan函数里面添加了这部分内容，这里只存了 起始页号 和 结束页号 两条映射，而不是像之前那样用 for 循环把每一页都存一遍。这是一种索引优化策略——用首尾两条记录代表整个连续区间，而不是每一页都存一条。

  这个方法的优化逻辑是：当从 Page Cache 申请一个 Span 时，NewSpan 会通过切分得到一个 kSpan（我们要的那 k 页）和一个 nSpan（剩下的 n-k 页）。对于 kSpan，确实用 for 循环建立了全量映射。这个 kSpan 会被交给 Central Cache，然后被切分成一个个小块分配给 Thread Cache。所以 Thread Cache 拿到的所有内存块，都落在 kSpan 的页范围里，而 kSpan 的每一页都在映射表里有记录。

  那 nSpan 呢？它被挂回 Page Cache 的桶里，作为空闲内存等待下一次分配。它从来不会被切分成小块交给 Thread Cache，所以它的中间页也从来不会有内存块被传入 MapObjectToSpan。只有当 nSpan 被重新分配出去变成新的 kSpan 时，才会用 for 循环建立全量映射。所以，"只存首尾页"的优化只针对 Page Cache 中空闲状态的 Span 使用，用于快速判断某个页是否属于一个空闲 Span。而真正会被 Thread Cache 使用的 Span，依然保持着全量映射。

  这段代码的好处体现在 ReleaseSpanToPageCache 的合并过程中。当合并相邻页时，需要判断 prevSpan 和 nextSpan 是否存在且空闲。如果每次都用 _idSpanMap.find(prevId) 去查，而 prevSpan 是一个巨大的空闲 Span，它的每一页都有映射的话，那无论你查的是起始页还是中间页，都能找到它。但如果只存首尾页，你查询 prevId 时拿到的就是 prevSpan，完全满足需求。节省了 for 循环遍历的空闲 Span 的映射表内存和建立时间。

  这里有一个细节要注意：在 ReleaseSpanToPageCache 的合并过程中，当把 prevSpan 合并进来之后，我们最终 delete 了 prevSpan，但在 delete 之前，我们还要把 prevSpan 的首尾页映射从 _idSpanMap 中清除。因为合并之后，如果不清除这份映射关系，_idSpanMap 里就残留着 prevSpan 的 _pageId 和 _pageId + _n - 1 这两条记录，它们现在指向的是已经被 delete 的 prevSpan 对象的地址，是野指针。虽然后续会用 _idSpanMap[span->_pageId] = span 和 _idSpanMap[span->_pageId + span->_n - 1] = span 覆盖掉新的首尾页，但如果合并过程中某个旧页号恰好落在新的 span 的中间位置，而它刚好又是旧 prevSpan 的首尾页之一，那这条残留记录会一直存在，指向一个已经被销毁的对象，后续访问就会崩溃。

  我们用举例的方法更清晰的让大家意识到这个问题：

页号	映射到的Span
100	kSpan
101	kSpan
102	kSpan
103	nSpan
104~8	（没有记录）
109	nSpan

  现在的 _idSpanMap 的映射关系是这样的：

  ReleaseSpanToPageCache(kSpan) 现在开始合并逻辑。

  kSpan 的情况：kSpan._pageId = 100

  kSpan._n = 3

  kSpan._isUse = false

  向前合并：检查页号 99，_idSpanMap.find(99) 找不到，不合并。

  向后合并：检查页号 100 + 3 = 103，_idSpanMap.find(103) 找到了，返回 nSpan。检查 nSpan._isUse == false，且合并后 3 + 7 = 10 <= 128，可以合并。

//这就是发生隐患的地方
span->_n += nextSpan->_n;   // span->_n 从 3 变成 10
_spanLists[nextSpan->_n].Erase(nextSpan);  // 从 _spanLists[7] 里摘掉 nSpan
delete nextSpan;   // 销毁 nSpan 对象

  合并结束后，span 现在的状态是：span._pageId = 100（没变）span._n = 10（合并后）

  页号范围：100~109   然后执行合并后的映射更新：

_idSpanMap[span->_pageId] = span;                  // _idSpanMap[100] = span
_idSpanMap[span->_pageId + span->_n - 1] = span;   // _idSpanMap[109] = span

  因此当前的 _idSpanMap 映射关系就变成了：

页号	映射到的Span	状态
100	span（合并后的）	有效
101	kSpan	野指针
102	kSpan	野指针
103	nSpan	野指针
109	span（合并后的）	有效

  

  现在，假设程序再次分配内存，Thread Cache 从 Central Cache 拿了一个新的内存块，这个块碰巧落在了页号 101 上（因为合并后的 span 又重新被切分使用）。过了一段时间，Thread Cache 释放这个内存块，调用 ReleaseListToSpans，执行到 MapObjectToSpan，它根据内存地址算出页号，发现是 101，然后去 _idSpanMap.find(101)。

  _idSpanMap[101] 里现在存的是什么？是 已经 delete 掉的 kSpan 对象的地址。程序拿到了这个野指针，把它当作 Span* 去访问，比如访问 span->_freeList 或 span->_useCount，但实际上这块内存已经被释放回操作系统或重新分配给其他对象了，里面的数据完全不可预测。

  结果就是：读取访问权限冲突，程序崩溃。

  而如果我们在合并Span之前，先加上：

    for (PAGE_ID i = 0; i < nextSpan->_n; ++i)
    {
        _idSpanMap.erase(nextSpan->_pageId + i);
    }

  这样不管是被合并的 Span 是当初用全量映射还是首尾映射插入的，都可以解决这个问题，虽然牺牲了一点性能，但保证了映射表的绝对干净。

  所以，在 delete prevSpan 之前，应该把 prevSpan 的映射从 _idSpanMap 中 erase 掉。同理，对 nextSpan 也要做同样的清理。这是一个容易被忽略的细节，但如果不处理，你的程序可能在某个看似正常的回收操作中突然崩溃，且极难调试。

5. 细节优化

5.1 大于 256KB 的内存申请

  我们前面提到了，Thread Cache 负责分配小于 256KB 的内存，当有内存需要申请的时候会走三级缓存，但如果某线程当前需要一次性申请大于 256KB 的内存呢？这个时候我们就直接向上级缓存去申请，跳过 Thread Cache 。并且因为一个页的大小是 8KB， 256KB相当于是 32 页，所以比较好的方式是直接向 Page Cache 申请。但如果一次性申请的内存大小超过 128 页，我们就直接向堆申请。

static void* ConcurrentAlloc(size_t size) //这里的size是字节数
{
//=======================修改的部分==========================
	if (size > MAX_BYTES)
	{
		size_t alignSize = SizeClass::RoundUp(size);
		size_t kpage = alignSize >> PAGE_SHIFT;
		
		PageCache::GetInstance()->_pageMtx.lock();
		Span* span = PageCache::GetInstance()->NewSpan(kpage);
		PageCache::GetInstance()->_pageMtx.unlock();
	
		void* ptr = (void*)(span->_pageId << PAGE_SHIFT);
		return ptr;
	}
//===========================================================
	else
	{
		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)
{
//=======================修改的部分==========================
	if (size > MAX_BYTES)
	{
		Span* span = PageCache::GetInstance()->MapObjectToSpan(ptr);
		
		PageCache::GetInstance()->_pageMtx.lock();
		PageCache::GetInstance()->ReleaseSpanToPageCache(span);
		PageCache::GetInstance()->_pageMtx.unlock();
	}
//===========================================================
	else
	{
		assert(pTLSThreadCache);
		pTLSThreadCache->Deallocate(ptr, size);
	}
}

	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
		{
			return _RoundUp(size, 1 << PAGE_SHIFT);
		}
//==========================================================
	}

Span* PageCache::NewSpan(size_t k)
{
	assert(k > 0);
//========================修改的部分=========================
	//大于 128 页的直接向堆申请
	if (k > NPAGES - 1)
	{
		void* ptr = SystemAlloc(k);
		Span* span = new Span;
		span->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT;
		span->_n = k;

		_idSpanMap[span->_pageId] = span;

		return span;
	}
//===========================================================

//......
}

void PageCache::ReleaseSpanToPageCache(Span* span)
{
//=====================修改的部分======================
	//大于 128 页的 page 直接还给堆
	if (span->_n > NPAGES - 1)
	{
		void* ptr = (void*)(span->_pageId << PAGE_SHIFT);
		SystemFree(ptr);
		delete span;

		return;
	}
//====================================================

//.......
}

//Common.h文件

inline static void SystemFree(void* ptr)
{
#ifdef _WIN32
	VirtualFree(ptr, 0, MEM_RELEASE);
#else
	//sbrk unmmap等等
#endif
}

  对于上述代码，我们用一段样例来跑通逻辑：

void Test()
{
	void* p2 = ConcurrentAlloc(129 * 1024 * 8);//129页
	ConcurrentFree(p2, 129 * 1024 * 8);
}

  这个用例要申请的字节数是129 * 1024 * 8 = 129 * 8192，也就是129页。这个大小刚好超过了NPAGES - 1 = 128，所以走的是超大内存分支。

  第一步：进入ConcurrentAlloc，判断size > MAX_BYTES成立。129 * 8192 = 1,056,768 字节，大约是1MB出头，远大于256KB，进入大内存分支。

  第二步：SizeClass::RoundUp(size)按页对齐，129 * 8192已经是8192的整数倍，所以alignSize = 129 * 8192，没有额外对齐损耗。

  第三步：算页数kpage = alignSize >> PAGE_SHIFT，129 * 8192 >> 13 = 129。正确。所以kpage = 129。

  第四步：进入NewSpan(129)，NewSpan开头判断if (k > NPAGES - 1)，即129 > 128成立，进入直接向系统申请的路径。调用 void* ptr = SystemAlloc(k)，向操作系统申请129页的连续内存。在Windows下，SystemAlloc调用VirtualAlloc，返回一个页对齐的地址。然后新建一个Span对象：span->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT，把地址转成页号；span->_n = k（即129）。只存起始页的映射：_idSpanMap[span->_pageId] = span。最后直接返回这个Span，完全不经过Page Cache的桶，也不参与后续的合并。

  第五步：拿到Span*后转成地址，void* ptr = (void*)(span->_pageId << PAGE_SHIFT)，返回给用户。

  第六步：ConcurrentFree(p2, 129 * 1024 * 8)释放，释放时传入的size大于256KB，走进大内存释放分支。先调用 MapObjectToSpan(ptr) 反查Span，通过ptr >> PAGE_SHIFT算出页号。因为释放时传进来的ptr就是当初分配时返回的起始地址，所以算出来的页号正好等于当初存的span->_pageId，_idSpanMap里能查到，返回这个Span指针。

  第七步：这也是最关键的一步。ReleaseSpanToPageCache判断if (span->_n > NPAGES - 1)，即129 > 128成立，进入直接归还系统的路径：void* ptr = (void*)(span->_pageId << PAGE_SHIFT)，把页号还原成地址。SystemFree(ptr)，在Windows下调用VirtualFree(ptr, 0, MEM_RELEASE)，把这块内存归还给操作系统。

  最后 delete span，销毁Span对象。

5.2 使用定长内存池配合脱离使用new

  我们要知道，TCMalloc设计的初衷就是为了能替代掉malloc，对于我们原先的写法当中，涉及到直接使用new、delete等等，比如 new Span ，它在C++中做两件事：1. 从堆上分配一块足够存放Span对象的内存（operator new，底层调用malloc）。2. 调用Span的构造函数，初始化_pageId、_n、_freeList、_useCount等成员。

  而 delete span 反过来做两件事：1. 调用Span的析构函数。2. 把这块内存归还给堆（operator delete，底层调用free）。

  问题在于，new和delete是通用内存分配器，每次分配都要走系统调用或至少走一次malloc的慢速路径。在Page Cache里，Span对象的创建和销毁非常频繁——切分大Span时要new，合并时又要delete，每分配一次大内存就要新建一个Span对象，每释放一次就要销毁一个。如果每次都走malloc，这部分开销会积累成一个不可忽视的性能损耗。

  而我们之前写的定长内存池就可以帮解决这个问题，我们在PageCache类中维护一个新的变量 ObjectPool<Span> _spanPool ，这是一个专门为Span对象定制的内存池。它的核心逻辑是：在New()里，优先从_freeList中取一个已经释放回来的、不再使用的Span对象内存块复用，如果_freeList为空，才从预先申请的大块内存中切出一块。在Delete()里，调用Span的析构函数后，把这块内存头插回_freeList，等待下次复用。

  所以，Span* span = _spanPool.New();做的事情和new Span完全一样——都返回一个指向已构造好的Span对象的指针。但区别在于，_spanPool.New()不依赖系统堆，它的分配路径比malloc短得多，而且因为每次分配的大小固定（sizeof(Span)），不存在内存碎片问题。

  当然，细心的同学会发现我们在 ObjectPool 的 New 函数里也使用了 malloc ，可能会有疑惑，这不是脱裤子放屁吗，我之前直接用new，底层最后是用malloc，你这也是用malloc，那有啥区别？

  实际上，直接用 new Span，意味着每次你要创建一个 Span 对象来管理内存页时，都会触发一次 malloc 调用。我们说了，在 Page Cache 的运行过程中，Span 的创建和销毁极其频繁——切分大块内存时 new，合并空闲页时 delete，分配大内存时 new，释放大内存时 delete。如果你的程序频繁申请和释放不同大小的内存块，系统可能要在短时间内调用成千上万次 malloc，每次调用都涉及堆管理器的查找、锁竞争、空闲链表遍历，这是巨大的性能损耗。

  而 ObjectPool 只在 “_freeList 为空且当前大块内存用完” 时，才会调用一次 malloc(128 * 1024)，一次性拿下一大块连续内存。然后这 128KB 会被切成几十甚至上百个 Span 对象大小的槽位，逐个分配给后续的 _spanPool.New() 调用。

  算一笔账：假设 sizeof(Span) 是 32 字节，128KB 能切出 4096 个 Span 对象。也就是说，你每调用 4096 次 _spanPool.New()，才会触发一次 malloc。而直接用 new Span，则是每调用 1 次就触发 1 次 malloc。前者把 malloc 的调用频率降低了整整 4096 倍。这是性能优化的核心所在。

5.3 释放对象时优化为不传对象大小

  大家在写释放内存的函数时，特别是在测试释放内存的函数时，都要再加上释放内存的大小，对于手动释放来说，我还得清楚的知道这个内存大小是多少才能释放，这非常的不方便，我们也需要对此进行修改。

  修改的方式也特别简单，因为我们的内存要么是从NewSpan函数来，要么是从GetOneSpan函数里来的，那我们只需要在Span这个类当中去维护一个新的变量 size_t _objsize; 用它来表示切分好的小内存块的大小，然后在这两个地方进行修改：

5.4 代码性能测试

 为了测试我们的高并发内存池，这里提供了一套测试代码，用于测试我们编写的内存池和malloc的性能有何差异：

#define _CRT_SECURE_NO_WARNINGS 1

#include "ConcurrentAlloc.h"
#include "Common.h"

//ntime  一轮申请和释放内存的次数
//round  轮次
//nworks 线程数
void BenchmarkMalloc(size_t ntimes, size_t nworks, size_t rounds)
{
    std::vector<std::thread> vthread(nworks);
    std::atomic<size_t> malloc_costtime = 0;
    std::atomic<size_t> free_costtime = 0;
    for (size_t k = 0; k < nworks; ++k)
    {
        vthread[k] = std::thread([&, k](){
            std::vector<void*> v;
            v.reserve(ntimes);
            for (size_t j = 0; j < rounds; ++j)
            {
                size_t begin1 = clock();
                for (size_t i = 0; i < ntimes; i++)
                {
                    v.push_back(malloc(16));
                    //v.push_back(malloc((16 + i) % 8192 + 1));

                }
                size_t end1 = clock();
                size_t begin2 = clock();
                for (size_t i = 0; i < ntimes; i++)
                {
                    free(v[i]);
                }
                size_t end2 = clock();
                v.clear();
                malloc_costtime += (end1 - begin1);
                free_costtime += (end2 - begin2);
            } });
    }

    for (auto& t : vthread)
    {
        t.join();
    }
    printf("%u个线程并发执行 %u轮次,每轮次malloc %u次 : 花费： %u ms\n ", nworks, rounds, ntimes, malloc_costtime.load());
    printf("%u个线程并发执行 %u轮次,每轮次free %u次 : 花费： %u ms\n ", nworks, rounds, ntimes, free_costtime.load());
    printf("%u个线程并发 malloc &free % u 次，总计花费： % u ms\n ", nworks, nworks * rounds * ntimes, (malloc_costtime.load() + free_costtime.load()));
}
//单轮次申请释放次数  线程数  轮次

void BenchmarkConcurrentMalloc(size_t ntimes, size_t nworks, size_t rounds)
{
    std::vector<std::thread> vthread(nworks);
    std::atomic<size_t> malloc_costtime = 0;
    std::atomic<size_t> free_costtime = 0;
    for (size_t k = 0; k < nworks; ++k)
    {
        vthread[k] = std::thread([&](){
            //printf("=====线程lambda开始执行了====\n"); //线程一进来第一行！
            std::vector<void*> v;
            v.reserve(ntimes);
            for (size_t j = 0; j < rounds; ++j)

            {
                //printf("thread enter j=%zu\n", j); //调试
                size_t begin1 = clock();
                for (size_t i = 0; i < ntimes; i++)
                {
                    //printf("i=%zu,开始调用ConcurrentAlloc\n", i);
                    //v.push_back(ConcurrentAlloc(16));
                     v.push_back(ConcurrentAlloc((16 + i) % 8192 + 1));
                }
                size_t end1 = clock();
                size_t begin2 = clock();
                for (size_t i = 0; i < ntimes; i++)
                {
                    ConcurrentFree(v[i]);
                }
                size_t end2 = clock();
                v.clear();
                malloc_costtime += (end1 - begin1);
                free_costtime += (end2 - begin2);
                //printf("thread exit j=%zu\n", j); //调试
            }
        });
    }
    for (auto& t : vthread)
    {
        t.join();
    }
    printf("%u个线程并发执行 %u轮次，每轮次concurrent alloc %u次 : 花费： %u ms\n ", nworks, rounds, ntimes, malloc_costtime.load());
    printf("%u个线程并发执行 %u轮次，每轮次concurrent dealloc %u次 : 花费： %u ms\n ", nworks, rounds, ntimes, free_costtime.load());
    printf("%u个线程并发concurrent alloc &dealloc % u 次，总计花费： % u ms\n ", nworks, nworks * rounds * ntimes, (malloc_costtime.load() + free_costtime.load()));
}
int main()
{
    size_t n = 10000;
    cout << "==========================================================" << endl;
    BenchmarkConcurrentMalloc(n, 4, 10);
    cout << endl << endl;
    //BenchmarkMalloc(n, 4, 10);

    cout << "==========================================================" << endl;
    return 0;
}

  下面是我的测试结果：

  大家会发现，分配路径确实赢了一点——1050ms 对 2663ms， Thread Cache 无锁分配确实发挥了作用。但释放路径 5869ms 几乎占了总耗时的 85%，这已经不是“有点慢”了，这是整个系统的瓶颈所在。

  我们的释放路径是：ConcurrentFree(ptr) → ThreadCache::Deallocate(ptr, size) → Push 到本地链表 → 检查 Size() >= MaxSize() → 触发 ListTooLong → PopRange 摘下一批 → CentralCache::ReleaseListToSpans(start, size)

  而问题出在 ReleaseListToSpans 里。

  我们进到这个函数内部，看它在 16 字节固定申请的场景下干了什么：锁住 Central Cache 的桶锁：_spanLists[index]._mtx.lock()。4个线程都在释放 16 字节，它们都在争抢 同一个桶（index = 0）。这把锁成了所有释放操作的“单行道”，一次只能一个线程通过。进入 while (start) 循环，逐个处理释放块：你从 Thread Cache 摘下来的可能只有 MaxSize() 个块（初始是 1，慢启动慢慢涨），每释放一个块都要：调用 PageCache::GetInstance()->MapObjectToSpan(start)，在 std::map 里做一次红黑树查找（O(log N)），这个操作本身不慢，但在 Debug 模式下 std::map 的迭代器会做大量安全检查，而且 40 万次释放意味着 40 万次树查找。

  NextObj(start) = span->_freeList; span->_freeList = start; 头插回 Span。span->_useCount--，然后检查 if (span->_useCount == 0)。频繁触发 Span 完全空闲，引发锁切换：16 字节的块，一页 8KB 能放 512 个。我的测试在反复分配和释放，_useCount 在 0 和 512 之间来回摆动。每当 _useCount 降到 0：_spanLists[index].Erase(span)（桶锁保护下操作），解锁桶锁，加锁 PageCache 锁，调用 ReleaseSpanToPageCache(span)（里面会尝试向前向后合并，这会去查 std::map，修改映射），解锁 PageCache 锁，重新加锁桶锁。然后 start = next，继续处理下一个块。这把锁切换开销是致命的。 在 4 个线程同时释放的情况下，一个线程在持桶锁时触发锁切换，另外三个线程就在桶锁上等着。释放路径的大部分时间都消耗在 “排队等锁” 和 “切锁” 上，而不是真正在做内存操作。

  而 malloc 的释放路径通常只需要把内存块挂回线程本地缓存（tcache）或空闲链表中，不需要查 std::map，不需要判断 Span 是否空闲，不需要切锁。它的锁粒度极细，甚至完全无锁（取决于实现）。我们的释放路径为了“回收给 Page Cache”这个功能，增加了一层 Central Cache 的逻辑，这层逻辑在反复分配释放的微基准测试中成了累赘。

5.5 使用基数树优化性能瓶颈

  我们先来看看用在TCMalloc里的基数树是怎么操作的：

#pragma once

#include "Common.h"

// Single‑level array
template <int BITS>
class TCMalloc_PageMap1 
{
private:
    static const int LENGTH = 1 << BITS;
    void** array_;

public:
    typedef uintptr_t Number;

    explicit TCMalloc_PageMap1(void* (*allocator)(size_t)) 
    {
        array_ = reinterpret_cast<void**>((*allocator)(sizeof(void*) << BITS));
        memset(array_, 0, sizeof(void*) << BITS);
    }

    // Return the current value for KEY. Returns NULL if not yet set,
    // or if k is out of range.
    void* get(Number k) const 
    {
        if (((k >> BITS) > 0)) 
        {
            return NULL;
        }
        return array_[k];
    }

    // REQUIRES "k" is in range "[0,2^BITS‑1]".
    // REQUIRES "k" has been ensured before.
    //
    // Sets the value 'v' for key 'k'.
    void set(Number k, void* v) 
    {
        array_[k] = v;
    }
};


// Two‑level radix tree
template <int BITS>
class TCMalloc_PageMap2 
{
private:
    // Put 32 entries in the root and (2^BITS)/32 entries in each leaf.
    static const int ROOT_BITS = 5;
    static const int ROOT_LENGTH = 1 << ROOT_BITS;

    static const int LEAF_BITS = BITS - ROOT_BITS;
    static const int LEAF_LENGTH = 1 << LEAF_BITS;

    // Leaf node
    struct Leaf 
    {
        void* values[LEAF_LENGTH];
    };

    Leaf* root_[ROOT_LENGTH];             // Pointers to 32 child nodes
    void* (*allocator_)(size_t);          // Memory allocator

public:
    typedef uintptr_t Number;

    explicit TCMalloc_PageMap2(void* (*allocator)(size_t)) 
    {
        allocator_ = allocator;
        memset(root_, 0, sizeof(root_));
    }

    void* get(Number k) const {
        const Number i1 = k >> LEAF_BITS;
        const Number i2 = k & (LEAF_LENGTH - 1);
        if ((k >> BITS) > 0 || root_[i1] == NULL) 
        {
            return NULL;
        }
        return root_[i1]->values[i2];
    }

    void set(Number k, void* v) {
        const Number i1 = k >> LEAF_BITS;
        const Number i2 = k & (LEAF_LENGTH - 1);
        ASSERT(i1 < ROOT_LENGTH);
        root_[i1]->values[i2] = v;
    }

    bool Ensure(Number start, size_t n) {
        for (Number key = start; key <= start + n - 1;) 
        {
            const Number i1 = key >> LEAF_BITS;
            // Check for overflow
            if (i1 >= ROOT_LENGTH)
                return false;

            // Make 2nd level node if necessary
            if (root_[i1] == NULL) 
            {
                Leaf* leaf = reinterpret_cast<Leaf*>
                    ((*allocator_)(sizeof(Leaf)));
                if (leaf == NULL) return false;
                memset(leaf, 0, sizeof(*leaf));
                root_[i1] = leaf;
            }

            // Advance key past whatever is covered by this leaf node
            key = ((key >> LEAF_BITS) + 1) << LEAF_BITS;
        }
        return true;
    }

    void PreallocateMoreMemory() 
    {
        // Allocate enough to keep track of all possible pages
        Ensure(0, 1 << BITS);
    }
};


// Three‑level radix tree
template <int BITS>
class TCMalloc_PageMap3 
{
private:
    // How many bits should we consume at each interior level
    static const int INTERIOR_BITS = (BITS + 2) / 3; // Round‑up
    static const int INTERIOR_LENGTH = 1 << INTERIOR_BITS;

    // How many bits should we consume at leaf level
    static const int LEAF_BITS = BITS - 2 * INTERIOR_BITS;
    static const int LEAF_LENGTH = 1 << LEAF_BITS;

    // Interior node
    struct Node
    {
        Node* ptrs[INTERIOR_LENGTH];
    };

    // Leaf node
    struct Leaf 
    {
        void* values[LEAF_LENGTH];
    };

    Node* root_;                          // Root of radix tree
    void* (*allocator_)(size_t);           // Memory allocator

    Node* NewNode() 
    {
        Node* result = reinterpret_cast<Node*>((*allocator_)(sizeof(Node)));
        if (result != NULL) 
        {
            memset(result, 0, sizeof(*result));
        }
        return result;
    }

public:
    typedef uintptr_t Number;

    explicit TCMalloc_PageMap3(void* (*allocator)(size_t)) 
    {
        allocator_ = allocator;
        root_ = NewNode();
    }

    void* get(Number k) const 
    {
        const Number i1 = k >> (LEAF_BITS + INTERIOR_BITS);
        const Number i2 = (k >> LEAF_BITS) & (INTERIOR_LENGTH - 1);
        const Number i3 = k & (LEAF_LENGTH - 1);
        if (((k >> BITS) > 0) ||root_->ptrs[i1] == NULL || root_->ptrs[i1]->ptrs[i2] == NULL) 
        {
            return NULL;
        }
        return reinterpret_cast<Leaf*>(root_->ptrs[i1]->ptrs[i2])->values[i3];
    }

    void set(Number k, void* v) 
    {
        ASSERT(k >> BITS == 0);
        const Number i1 = k >> (LEAF_BITS + INTERIOR_BITS);
        const Number i2 = (k >> LEAF_BITS) & (INTERIOR_LENGTH - 1);
        const Number i3 = k & (LEAF_LENGTH - 1);
        reinterpret_cast<Leaf*>(root_->ptrs[i1]->ptrs[i2])->values[i3] = v;
    }

    bool Ensure(Number start, size_t n) 
    {
        for (Number key = start; key <= start + n - 1;) 
        {
            const Number i1 = key >> (LEAF_BITS + INTERIOR_BITS);
            const Number i2 = (key >> LEAF_BITS) & (INTERIOR_LENGTH - 1);

            // Check for overflow
            if (i1 >= INTERIOR_LENGTH || i2 >= INTERIOR_LENGTH)
                return false;

            // Make 2nd level node if necessary
            if (root_->ptrs[i1] == NULL) 
            {
                Node* n = NewNode();
                if (n == NULL) return false;
                root_->ptrs[i1] = n;
            }

            // Make leaf node if necessary
            if (root_->ptrs[i1]->ptrs[i2] == NULL) 
            {
                Leaf* leaf = reinterpret_cast<Leaf*>((*allocator_)(sizeof(Leaf)));
                if (leaf == NULL) return false;
                memset(leaf, 0, sizeof(*leaf));
                root_->ptrs[i1]->ptrs[i2] = reinterpret_cast<Node*>(leaf);
            }

            // Advance key past whatever is covered by this leaf node
            key = ((key >> LEAF_BITS) + 1) << LEAF_BITS;
        }
        return true;
    }

    void PreallocateMoreMemory() 
    {
    }
};

  这里一共有三级基数树，其中第三级主要用于 64 位环境下的内存操作，我们目前主要聚焦于一级和二级基数树。

   在讲解代码之前，我们需要深刻理解位的知识，我们在代码里经常看到 PAGE_ID 这个类型，它本质上就是一个整数。在 32 位系统下，PAGE_ID 是一个 32 位的无符号整数。我们平时说的“某块内存的地址是 0x12345678”，这是十六进制。但计算机底层看到的是一串 32 位的二进制位，比如：

地址 0x12345678 = 二进制：0001 0010 0011 0100 0101 0110 0111 1000

对于第一层基数树，这个BITS设置的是在32位下，BITS=32-PAGE_SHIFT，在64位下，BITS=64-PAGE_SHIFT，因为进程地址空间是2^32,一个页的大小是2^13字节，除以一下就等于2^19，即需要2^19个位置去存储页号，BITS就用来表示算出来的这个19。 

   而代码里经常出现的 (PAGE_ID)ptr >> PAGE_SHIFT，就是把地址右移 13 位，得到页号。因为一页是 8KB = 2^13 字节，右移 13 位等价于“除以 8192”。所以页号实际上就是地址的高 19 位（因为 32 - 13 = 19）。比如某个地址是 0x12345678，右移 13 位后得到的页号大概是 0x91A2B（十九位）。这个十九位的二进制数，就是我们要拿来在基数树里做索引的 PAGE_ID。

  一级基数树的思想最朴素：既然页号的范围是固定的（2^19 个），那我就直接开一个长度为 2^19 的数组，数组下标就是页号，数组里存的就是 Span*。

template <int BITS>
class TCMalloc_PageMap1 
{
private:
    static const int LENGTH = 1 << BITS;  // 如果 BITS=19，LENGTH=524288
    void** array_;  // 指向这个大数组的指针

  BITS 就是页号占用的位数。在 32 位系统下，BITS = 32 - PAGE_SHIFT = 19。所以 LENGTH = 1 << 19 = 524288，意思是这个数组有 524288 个槽位。array_ 是一个二级指针，它指向一块连续的内存，这块内存的大小是 LENGTH * sizeof(void*)。在 32 位系统下，sizeof(void*) = 4，所以总大小是 524288 * 4 = 2MB。

  二级基数树的思想是：我不一次性把 2^19 个槽位全分配了，我只先分配一个小的根数组，然后只在真正用到某个页号时，才去分配对应的叶子数组。

  举个例子，假设 BITS = 19，页号是一个 19 位的二进制数。为了方便理解，我们用 19 位二进制表示一个页号：

页号（19位）： 1 0 1 1 0 0 1 1 1 0 1 0 0 1 1 0 1 0 1
                      ↑                           ↑
                    高5位                       低14位

  高 5 位是这个二进制数的前 5 位（从左边数），低 14 位是剩下的 14 位。在代码里，这两段是这样定义的：

static const int ROOT_BITS = 5;           // 高5位
static const int ROOT_LENGTH = 1 << 5;    // 根数组有32个槽位

static const int LEAF_BITS = BITS - ROOT_BITS;  // 低14位
static const int LEAF_LENGTH = 1 << LEAF_BITS;  // 每个叶子数组有16384个槽位

  所以：根数组（root_）：有 32 个槽位，下标是页号的高 5 位（范围 0~31）。

  叶子数组（Leaf）：有 16384 个槽位，下标是页号的低 14 位（范围 0~16383）。

  一个根数组槽位 + 一个叶子数组，组合起来能覆盖 32 * 16384 = 524288 个页号，正好是 2^19。

页号（19位） = [高5位][低14位]
                  ↓        ↓
            根数组索引   叶子数组索引
             (0~31)     (0~16383)

查找过程：
1. 用高5位去 root_[高5位] 找到对应的 Leaf 指针
2. 用低14位去 Leaf->values[低14位] 取出 Span*

  然后我们只需要在我们的高并发内存池当中进行下列修改，调用基数树中的函数：

  下面看一下优化后的高并发内存池的性能：

  大家会发现当前高并发内存池的速率大大提升，现在的性能之所以会有这么大的提升，我们首先要理解，在原来的 std::map 方案里，每一次 ConcurrentFree 释放一个内存块，都要走一遍完整的“加锁 → 红黑树查找 → 解锁”流程。在 4 线程 40 万次释放的测试中，这个流程被重复了 40 万次。而且释放路径上还有 ReleaseListToSpans 里的 while 循环，每循环一次就调用一次 MapObjectToSpan——也就是说，40 万次释放可能对应着 40 万次甚至更多次的 MapObjectToSpan 调用。

  因为 map 是红黑树。红黑树的查找从根节点开始，每层比较一次键值，决定向左还是向右走。你的页号映射可能涉及几千到几万个 Span，树的高度大概在 12~15 层。也就是说，一次 map::find 要经历 12~15 次指针跳转和键值比较。每次跳转都访问一个不同的内存地址，这些地址在堆上分散分布，CPU 的缓存预取机制很难发挥作用，大量时间消耗在等待内存数据从 RAM 加载到 Cache 上。

  而基数树的 get 函数只做两件事：

const Number i1 = k >> LEAF_BITS;              // 一次右移
const Number i2 = k & (LEAF_LENGTH - 1);       // 一次按位与
return root_[i1]->values[i2];                  // 两次数组访问

  这两次数组访问的地址是连续的——root_ 数组只有 32 个元素，values 数组有 16384 个元素，都是连续内存。CPU 可以把它们一次性加载到 Cache 里，后续的访问全部命中 Cache，几乎没有内存等待延迟。而且两次数组访问的操作，在 CPU 指令层面只需要几条指令，比红黑树的十几层循环要快得多。

  这是最关键的一点，原来的 std::map 方案里，MapObjectToSpan 必须加锁。因为 std::map 的迭代器在并发场景下不安全——如果一个线程正在 find 遍历树，另一个线程在 set 插入或删除节点，迭代器就可能失效，导致崩溃。所以必须用一把全局锁把所有读写操作都串行化。

  在 4 个线程同时释放内存时，它们都在调用 MapObjectToSpan，都在争抢这把锁。大部分时间不是花在查找上，而是花在等待锁被释放上。这就是你之前释放路径耗时较长的核心原因——锁竞争把并发的释放操作变成了串行的排队。

  而基数树的 get 操作是只读的，而且它不涉及任何指针遍历，只是读取数组里的值。多个线程同时读取同一个数组位置，不会产生任何数据竞争，因为读操作之间天然互不干扰。你不需要加锁，多个线程可以同时执行 get，互不等待。

  所以现在的情况是：40 万次 MapObjectToSpan 调用，每次都是一次无锁、O(1)、Cache 友好的数组访问。4 个线程可以同时查映射，没有任何一个线程需要等待另一个线程释放锁。

  大家可能还会有一个疑问：我们可以看到 get 是无锁的，但 set 和 Ensure 还是有锁的，它们会不会在并发分配时成为新瓶颈？

  实际上 Ensure 和 set 只在分配新 Span 时调用——也就是在 NewSpan 里，当 Page Cache 把一块内存页切分成 kSpan 或 nSpan 时，需要往基数树里写入新的映射。分配的频率远远低于释放的频率。在你的测试场景中，分配 10 万次内存，但 NewSpan 被调用的次数可能只有几十次或几百次（因为一次 NewSpan 会返回一个包含很多内存块的 Span，后续的分配都从同一个 Span 里取）。所以 set 和 Ensure 的调用次数极少，即使它们需要加锁，也远远构不成瓶颈。

  而且 Ensure 里用了我们的的 ObjectPool<Leaf> 来管理 Leaf 节点的内存，这又是定长内存池的优势——叶子节点的分配复用极快，不依赖 malloc。

  下面是对本项目的所有代码， 放在了我的个人Gitee仓库，有需要的可以自取：

https://gitee.com/chen-yukun-1030/Program_set.git

  本文到此结束，感谢各位读者的阅读，如果有讲解的不到位或者错误的地方，欢迎各位读者进行批评或指正。