引言

为了解决系统的内存分配函数malloc / free调用系统调用的开销与锁竞争导致性能急剧下降，该项目基于tcmalloc（全称Thread-Caching Malloc）实现了一个高并发的内存池，实现高效的多线程内存管理，来替代malloc / free。

google的原型太复杂，我们是把其中核心的框架简化出来模拟实现一个高并发内存池，希望我们都能从中汲取知识。

一、内存池

1、是什么

这就跟家里母亲管钱是一个道理。我和父亲都要生活费，但我每个月拿一两千就够花了，我父亲却总是隔两天要几百买烟、几千买鱼竿，花销特别没规律。母亲每次都得费心费力地给他单独分配。后来母亲干脆给他一笔钱当‘小金库’，让他自己慢慢花。这下母亲省心了，以后只管给我打钱就行。其中母亲就是操作系统，我是一个堆区上的小数组，而父亲就是内存池。

2、作用

（1）提高申请内存空间效率

（2）解决内存碎片问题：

主要解决外碎片问题：申请内存必须是连续的，外碎片就是在一个内存空间中，由于申请释放内存的顺序差异，导致总空间足够用户需求，但由于不连续而申请不出来的问题。

引入该项目内存池能够通过合理的设计解决这一问题。

但内碎片无法解决：内碎片是由于一些内存对齐的需求导致分配出去的空间中一些空间始终无法被利用导致的内存浪费问题，这当然是不能通过架构解决的。

如下图所示：

二、高并发内存池框架设计：三层缓存

基本目的是设计框架解决高并发场景下的锁竞争问题，我们来看看是如何设计的。

1、thread cache：线程缓存模块，用于分配小于256kb的内存分配，每个线程对应一个thread cache，所以线程在此处申请内存时无需加锁，这是该内存池高效的关键。

2、central cache：中心缓存模块，所有线程共用一个，thread cache就是从此处申请内存块，当thread cache占据过多内存时central cache负责回收一个批量，当central cache某个span块的use_count 降为0时（表示现在是一块“完整且空闲”的连续内存），将该span块摘除归还给page cache，实现多线程均衡的内存分配。

3、page cache：页缓存模块，按照页为单位存储及分配内存，当central cache没有内存块时会从此处申请一定数量的page，并按照对应内存大小切割好再分配给central cache。当有span块归还时，page cache会并且合并相邻的页，直到无法合并为止，组成更大的页，缓解外碎片的问题。

哈希桶映射对齐规则：

1、Thread Cache模块

1、结构介绍

thread cache是一个哈希桶结构，每个桶是⼀个自由链表，并且按照映射规则对应内存块大小。

thread cache框架：

自由链表：利用空闲块头部存储下一个节点的指针，兼容32/64位环境使用void*作为头部大小。

class ThreadCache { public: //申请和释放对象 void* Allocate(size_t size); void Deallocate(void* ptr, size_t size); //从中心缓存申请对象 void* FetchFromCentralCache(size_t index, size_t size); //释放逻辑，当归还的内存块大于等于一个批量时就让central cache回收 void ListTooLong(FreeList& list, size_t size); private: FreeList _hash_bucket[MAX_NUMS]; }; //TLS thread local storage //将thread cache 设置为线程独立的变量 static _declspec(thread) ThreadCache* pTLSThreadCache = nullptr; //自由链表 class FreeList { public: void Push(void* obj) { assert(obj); //头插 NextObj(obj) = _freelist; _freelist = obj; _size++; } //插入一段节点 void PushRange(void* start, void* end, size_t n) { NextObj(end) = _freelist; _freelist = start; _size += n; } void* Pop() { assert(_freelist); //头删 void* obj = _freelist; _freelist = NextObj(obj); _size--; return obj; } void PopRange(void*& start, void*& end, size_t n) { assert(n >= _size); start = _freelist; end = start; for (size_t i = 1; i < n; i++) { end = NextObj(end); } _freelist = NextObj(end); NextObj(end) = nullptr; _size -= n; } bool Empty() { return _freelist == nullptr; } size_t& GetMaxSize() { return _maxsize; } size_t& GetSize() { return _size; } private: void* _freelist = nullptr; //自由链表头指针 size_t _size = 0; //节点个数 size_t _maxsize = 1; };

2、申请内存：
（1）当内存申请大小size <= 256KB，先获取到线程存储的thread cache对象，计算size映射的哈希桶自由链表下标 i 。

映射对齐：RoundUp() 按照对齐规则向上取整得到对齐字节数，如12Byte对齐8Byte，那么对齐字节数就是16Byte。

下标映射：Index（） 可以拿着对齐后的字节数找到哈希桶的下标，但最好与另一个接口解耦，使用申请字节数向下取整计算哈希桶下标。

计算对齐字节数(请大小 size, 对齐步长 align): // 检查是否整除 如果 (size % align == 0): // 情况1：正好是对齐数的整数倍 // 例如：align=8, size=16 -> 应该对应16Byte 返回 size 否则： // 情况2：有余数，说明在下一个区间 // 例如: align=/8,, size=15 -> 应该对应16Byte 返回 ((size / align) + 1) * align 计算桶下标(申请大小 size, 对齐步长 align): // 检查是否整除 如果 (size % align == 0): // 情况1：正好是对齐数的整数倍 // 例如：align=8, size=16 -> 应该落在第1个桶(下标1) 返回 (size / align) - 1 否则: // 情况2：有余数，说明在下一个区间 // 例如：align=8, size=9 -> 应该落在第1个桶(下标1) 返回 (size / align)

位运算优化：

RoundUp(bytes, align): // 先加 (align-1) 向上推，再按位与掩码截断低位 返回 (bytes + align - 1) & ~(align - 1) Index(bytes, align_shift): // 1. 算出步长 (如 align_shift=3，步长即为8) 步长 = 1 << align_shift // 2. 加上(步长-1)向上推，再右移算出商，最后减1转为0基下标 返回 ((bytes + 步长 - 1) >> align_shift) - 1

（2） 如果自由链表_freeLists [i] 中有对象，则直接拿⼀个内存对象返回；
（3）如果_freeLists[i]中没有对象时，则按照慢启动算法，批量从central cache中获取⼀定数量的对象，插入到自由链表并返回⼀个对象。

NumMoveSize()：计算 Central Cache 每次批量分配给 ThreadCache 的对象个数上限，

区间在[2, 512]，实现慢启动算法，减少内存浪费。

void* ThreadCache::FetchFromCentralCache(size_t index, size_t size) { // 1. 慢启动算法：计算本次批量获取的数量 size_t& maxsize = _hash_bucket[index].GetMaxSize(); size_t batchNum = std::min(maxsize, SizeClass::NumMoveSize(size)); if (batchNum == maxsize) maxsize++; // 达到上限则增加下次获取量 // 2. 向 CentralCache 批量申请对象 void* start = nullptr, *end = nullptr; size_t actualNum = CentralCache::GetInstance()->FetchRangeObj(start, end, batchNum, size); // 3. 处理返回值：留第一个给用户，剩余的插入本地桶 if (actualNum > 1) { // 将第2个到最后一个节点头插回本地 FreeList _hash_bucket[index].PushRange(NextObj(start), end, actualNum - 1); } return start; // 返回第一个节点给用户 }

3、释放内存：
（1） 当释放内存小于 256KB 时将内存释放回thread cache，计算size映射自由链表桶位置i，将对象Push到_freeLists[i]。
（2）当链表的长度过长，则从链表中取出一个批量的内存块，还给central cache。

void ThreadCache::Deallocate(void* ptr, size_t size) { assert(ptr); assert(size <= MAX_NUMS); //插入回哈希桶即可 size_t index = SizeClass::Index(size); _hash_bucket[index].Push(ptr); //当归还的内存块大于等于一个批量时就让central cache回收 if (_hash_bucket[index].GetSize() >= _hash_bucket[index].GetMaxSize()) { ListTooLong(_hash_bucket[index], size); } }

2、Central Cache模块

1.结构介绍

central cahce框架设计：该模块是多线程共享的，会频繁访问，为了削减创建的开销，使用单例模式（饿汉模式，提前创建好对象）设计。

// 中心缓存 class CentralCache { public: // 获取单例 static CentralCache* GetInstance() { return &_sInst; } Span* GetOneSpan(SpanList& list, size_t size); //central cache分配给thread cache资源个数 size_t FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size); //回收thread cache一个批量的内存块 void ReleaseToSpans(void* start, size_t size); private: SpanList _spanlist[MAX_NUMS];//与thread cache大小相同的哈希桶 private: //多线程访问，没必要每次都创建对象，用一个单例即可 CentralCache(){} CentralCache(const CentralCache&) = delete; CentralCache& operator=(const CentralCache&) = delete; static CentralCache _sInst; }; // Span块设计 struct Span { void* _freeList = nullptr; //每个span块下挂的自由链表 PAGE_ID _pageid = 0; //span的页号 size_t _n = 0; //页的数量 size_t _objSize = 0; //小内存块的大小 Span* _prev = nullptr; //span的链表结构 Span* _next = nullptr; size_t _usecount = 0; //span下分割好的小内存分配给thread cache的数量 bool _isUse = false; //标记是否被Central Cache使用 };

2、申请内存

（1）当thread cache没有足够内存时就会想central cache申请内存对象了，此处使用了慢启动算法；且多线程共享，访问central cache需要加桶锁（加全局锁效率低下，加单个span块锁很可能发生死锁现象），以此提高申请效率。

（2）thread cache索引到的central cache也没有内存时，则向page cache申请一个span块，拿到span块之后将span管理的内存按对应大小切分好连接到自由链表上。最后central cache就能从span块中取出内存块给thread cache。

（3）central cache中的span块的use_count记录分配出去的内存块对象个数，为回收内存准备。

梳理流程：索引映射的span块 -> 有span则头删内存块返回，无则向page cache申请 -> 切分新span块连接至自由链表 -> 头删返回更新span // 1. 批量分配对象给 ThreadCache size_t CentralCache::FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size) { size_t index = SizeClass::Index(size); _spanlist[index].GetMutex().lock(); // 加桶锁 // 获取一个包含空闲对象的 Span（若为空则向 PageCache 申请） Span* curSpan = GetOneSpan(_spanlist[index], size); // 从 Span 的 FreeList 中向后遍历，提取 batchNum 个对象 start = curSpan->_freeList; end = start; size_t actualNum = 1; while (actualNum < batchNum && NextObj(end)) { end = NextObj(end); actualNum++; } // 断开链表，更新 Span 状态 curSpan->_freeList = NextObj(end); NextObj(end) = nullptr; curSpan->_usecount += actualNum; _spanlist[index].GetMutex().unlock(); // 释放桶锁 return actualNum; } // 2. 获取并切割 Span Span* CentralCache::GetOneSpan(SpanList& list, size_t size) { // 情况1：遍历当前桶，寻找有空闲对象的 Span Span* lt = list.Begin(); while (lt != list.End()) { if (lt->_freeList) return lt; lt = lt->_next; } // 情况2：当前桶为空，向 PageCache 申请前，先释放桶锁 // 目的是避免在向系统申请内存期间，阻塞其他线程释放内存 list.GetMutex().unlock(); PageCache::GetInstance()->GetMutex().lock(); // 加 PageCache的全局锁 Span* span = PageCache::GetInstance()->NewSpan(SizeClass::NumMoveSpan(size)); PageCache::GetInstance()->GetMutex().unlock(); // 释放全局锁 // 将申请到的 Span 按目标大小等分切割，串联成 FreeList（此处省略具体切割循环） // ... 切割逻辑 ... list.GetMutex().lock(); // 将 Span 插入回 CentralCache 需要重新加桶锁 list.PushFront(span); // 将切好的 Span 挂入当前桶 return span; }

3、释放内存

当thread cache某个链表内存块过多或线程被销毁时，将内存归还到central cache中并更新span块的use_count，若use_count降为0则将span归还到page cache中。

1. 页号反查地址：在Page Cache中建立了页号和内存块地址之间的映射（页号是每个span块的虚拟地址整除单页大小8kb），以此从自由链表中的内存块快速索引到span页号。
2. 锁竞争优化：当PageCache 回收Span 时，主动释放桶锁。因为PageCache回收 Span 可能合并前后页，这是一个耗时操作。释放桶锁可以保证其他线程在归还内存时不会被阻塞。

// 回收 ThreadCache 批量归还的内存对象 void CentralCache::ReleaseToSpans(void* start, size_t size) { size_t index = SizeClass::Index(size); _spanlist[index].GetMutex().lock(); // 加桶锁 while (start) { void* next = NextObj(start); // 1. 通过对象地址反查出它所属的 Span Span* span = PageCache::GetInstance()->MapPagetoSpans(start); // 2. 将对象头插回对应 Span 的 FreeList 中 NextObj(start) = span->_freeList; span->_freeList = start; span->_usecount--; // 更新使用计数 // 3. 重点：如果该 Span 上的对象全部释放（usecount == 0），触发回收 if (span->_usecount == 0) { // 从 CentralCache 的桶中提取该 Span _spanlist[index].Erase(span); span->_freeList = nullptr; // 先释放桶锁，再去向 PageCache 归还 _spanlist[index].GetMutex().unlock(); PageCache::GetInstance()->GetMutex().lock(); PageCache::GetInstance()->ReleaseToPageCache(span); PageCache::GetInstance()->GetMutex().unlock(); _spanlist[index].GetMutex().lock(); // 重新加桶锁，继续处理下一个对象 } start = next; } _spanlist[index].GetMutex().unlock(); // 释放桶锁 }

3、Page Cache模块

1、结构介绍

PageCache也设计为单例模式，保证多线程访时的效率与安全。

（1）设计一个定长内存池专门管理span：

Span 结构体的创建和销毁很频繁。如果直接使用new/delete，底层会触发操作系统的malloc/free，这不仅违背了内存池“减少系统调用”的初衷，还会引入额外的锁竞争。

因此，我们引入了ObjPool<Span>。它的底层原理与 ThreadCache 中的 FreeList 相同（利用空闲块头部存储下一个节点指针），但它专门用于管理 Span 这种固定大小的对象。预分配一大块内存，保证 PageCache 内部的自给自足。

（2）建立页号到Span块地址的映射：

页号：每个 Span 块的首地址除以页大小（8KB）之后得到的唯一编号。通过内存地址右移对应的位数（ptr >> PAGE_SHIFT）得出。例如，地址0x8000，页大小 8KB（即 213213 ），右移 13 位后，页号就是 1。

内存块地址：Span 被 CentralCache 按固定大小（如 32 字节）进一步切分后，挂在自由链表（FreeList）上，最终交付给用户malloc/free的各个小对象的具体虚拟地址。

页号到 Span 的反向映射：

page cache维护了一个哈希表，Key 是页号（Page ID），Value 是 Span 指针。

当用户释放内存时，内存池首先通过位运算（ptr >> PAGE_SHIFT）将内存块地址转换为所在的页号，然后拿着这个页号去哈希表中查找，反查出该内存块所属的 Span 块，并将其归还到对应的 FreeList 中。

//PageCache也设计为单例，多线程访问，保证效率与安全 class PageCache { public: Span* NewSpan(size_t npage); static PageCache* GetInstance() { return &_sInst; } std::mutex& GetMutex() { return _pagemtx; } Span* MapPagetoSpans(void* start); void ReleaseToPageCache(Span* span); private: PageCache() {} PageCache(const PageCache&) = delete; PageCache& operator=(const PageCache&) = delete; //单例 static PageCache _sInst; SpanList _spanlist[NPAGES]; //使用定长内存池替代new和delete ObjPool<Span> _objpool; //用一个哈希表记录一个页被截下来交付出去后，它的PAGE_ID和它的每个小内存块的Span* std::unordered_map<PAGE_ID, Span*> _Page2Span_map; std::mutex _pagemtx;//由于central申请内存会跨桶遍历，直到找到非空page，所以桶锁锁不住，需要一个大锁 };

2、申请内存

（1）能够看出central cache和page cache的结构很相似，但central cache的哈希桶是按照对齐规则映射的，page cache的哈希桶是按照页大小映射的，如128page下的span大小都是128page。

（2）当central cache向page cache申请内存时，page cache先检查映射位置是否有span，若没有则向后找到非空的位置下的span并切分出来：如申请的内存映射到5page，5page后没有span块，则向后找到了16page下的非空span，于是将它取出来切分为5page和11page，插入到对应位置。

（3）如果超过128page都未能找到非空span，则向操作系统直接申请一个128页的span挂到spanlist[128]的自由链表后，WIndows中使用VirtualAlloc，LINUX中使用mmap、brk。

Span* PageCache::NewSpan(size_t npage) { assert(npage > 0); // 1. 大于 128 页直接向操作系统申请 if (npage >= NPAGES) { void* ptr = SystemAlloc(npage); Span* span = _objpool.New(); span->_pageid = (PAGE_ID)ptr >> PAGE_SHIFT; span->_n = npage; _Page2Span_map[span->_pageid] = span; // 建立页号到 Span 的映射 return span; } // 2. 当前页数的 SpanList 不为空，直接弹出 if (!_spanlist[npage].Empty()) { Span* kspan = _spanlist[npage].PopFront(); kspan->_isUse = true; // (此处省略为 kspan 的每一页建立映射的逻辑) return kspan; } // 3. 【降级切分】当前页数为空，向上遍历寻找更大的 Span 进行切割 for (size_t i = npage + 1; i < NPAGES; ++i) { if (!_spanlist[i].Empty()) { Span* bigSpan = _spanlist[i].PopFront(); // 切分出需要的 npage 页 Span* kspan = _objpool.New(); kspan->_pageid = bigSpan->_pageid; kspan->_n = npage; // 调整剩余大块 Span 的属性 bigSpan->_pageid += npage; bigSpan->_n -= npage; // 将剩余部分插回对应的 SpanList _spanlist[bigSpan->_n].PushFront(bigSpan); // (此处省略为 kspan 和 bigSpan 建立页号映射的逻辑) kspan->_isUse = true; return kspan; } } // 4. 【兜底申请】所有桶都为空，向操作系统申请 128 页并递归调用 Span* bigspan = _objpool.New(); void* ptr = SystemAlloc(NPAGES - 1); bigspan->_pageid = (PAGE_ID)ptr >> PAGE_SHIFT; bigspan->_n = NPAGES - 1; _spanlist[bigspan->_n].PushFront(bigspan); return NewSpan(npage); // 递归复用切分逻辑 }

3、释放内存

当central cache归还了一个span后，前后合并span的page id的空闲span直到不能合并为止，将零散的小span滚雪球似的合并为一个大的span以此解决外碎片问题。

时刻注意合并条件：是否存在相邻页、相邻页是否空闲、合并后是否超过 128 页上限

//PageCache回收策略 void PageCache::ReleaseToPageCache(Span* span) { //直接回收大于128页的内存块到堆 if (span->_n > NPAGES - 1) { void* ptr = (void*)(span->_pageid >> PAGE_SHIFT); SystemFree(ptr); _objpool.Delete(span); return; } assert(span); //向前合并 while (true) { PAGE_ID prevId = span->_pageid - 1; auto ret = _Page2Span_map.find(prevId); //没有对应页号，不合并 if (ret == _Page2Span_map.end()) { break; } //前一个页号被使用中，不合并 Span* prevSpan = ret->second; if (prevSpan->_isUse == true) { break; } //前一个页号加当前页号的页数大于128页，无法管理，不合并 if (prevSpan->_n + span->_n > NPAGES - 1) { break; } //合并 span->_pageid = prevSpan->_pageid; span->_n += prevSpan->_n; //方便之后的合并 span->_isUse = false; _spanlist[prevSpan->_n].Erase(prevSpan); //delete prevSpan; _objpool.Delete(prevSpan); } //向后合并 while(true) { PAGE_ID nextId = span->_pageid + span->_n; auto ret = _Page2Span_map.find(nextId); //没有对应页号，不合并 if (ret == _Page2Span_map.end()) { break; } //后一个页号被使用中，不合并 Span* nextSpan = ret->second; if (nextSpan->_isUse == true) { break; } //后一个页号加当前页号的页数大于128页，无法管理，不合并 if (nextSpan->_n + span->_n > NPAGES - 1) { break; } //合并 span->_n += nextSpan->_n; //方便之后的合并 span->_isUse = false; _spanlist[nextSpan->_n].Erase(nextSpan); //delete nextSpan; _objpool.Delete(nextSpan); } //将span加入哈希桶和map方便之后的合并 _spanlist[span->_n].PushFront(span); //如span为2000页，有5页，将2000到2004加入 _Page2Span_map[span->_pageid] = span; _Page2Span_map[span->_pageid + span->_n - 1] = span; }

三、基数树优化span块查询

虽然哈希表能够O(1)快速查找，但高并发内存池真正耗时的操作是多线程的锁竞争问题，而STL库为了极致的效率是没有解决线程安全问题的（如哈希表的扩容），所以MapPagetoSpans() 接口相当耗时。为了解决这个问题，业界提出了一种被称为“基数树”的结构，通过将页号拆分到树结构中实现快速查找，并通过懒加载缓解内存消耗。

1、一层结构

如果开一个大数组，再把页号全部放进去映射当然也是可以的（如 32位环境，页数：2^19个，指针大小：4Byte，数组大小：2^19 * 4 = 4MB），但64位环境能达到恐怖的2^51 * 8 = 128PB，显然不合理，所以要考虑分层设计。

// 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); } // 传递页号k获取span地址 void* get(Number k) const { // 数组范围[0, 2^BITS - 1]，如果k >> BITS仍然大于0表示越界了 if ((k >> BITS) > 0) { return NULL; } return array_[k]; } // 设置映射 void set(Number k, void* v) { array_[k] = v; } };

2、二层结构

二层是比较尴尬的，32位能很好适应一层结构，64位有三层结构，这让它处于一种不上不下的地位。这里依旧32位举例，取高5位作为第一层存储的指针，19 - 5 = 14位作为第二次存储剩余的页号；但我们可以用它过渡三层结构，它的关键操作和三层结构是一致的。

key = ((key >> LEAF_BITS) + 1) << LEAF_BITS：

表示跳跃到第一层下一个节点并还原页号，保证循环逻辑一致性。

template <int BITS> class TCMalloc_PageMap2 { private: // 以32位环境为例： // 根节点（第一层）占用的位数 static const int ROOT_BITS = 5; // 根节点数组的长度（2^5 = 32个节点） static const int ROOT_LENGTH = 1 << ROOT_BITS; // 叶子节点（第二层）占用的位数（总位数减去根节点位数） static const int LEAF_BITS = BITS - ROOT_BITS; // 叶子节点数组的长度（例如2^14 = 16384个条目） static const int LEAF_LENGTH = 1 << LEAF_BITS; // 叶子节点结构体：真正存储映射数据（Span地址）的地方 struct Leaf { void* values[LEAF_LENGTH]; }; // 根节点数组：存放指向叶子节点的指针 Leaf* root_[ROOT_LENGTH]; // 内存分配器函数指针：用于按需动态分配叶子节点 void* (*allocator_)(size_t); public: // 将页号类型定义为无符号整数类型（与当前环境指针等长） typedef uintptr_t Number; // 初始化内存分配器，并将根节点数组全部置空 explicit TCMalloc_PageMap2(void* (*allocator)(size_t)) { allocator_ = allocator; memset(root_, 0, sizeof(root_)); } // 根据页号获取对应的 Span 地址 void* get(Number k) const { // 取高5位：计算该页号属于哪个根节点（第一层索引） const Number i1 = k >> LEAF_BITS; // 取低14位：计算该页号在叶子节点内的偏移量（第二层索引） const Number i2 = k & (LEAF_LENGTH - 1); // 若页号超出 BITS 表示的范围，或对应的叶子节点尚未分配，则返回 NULL if ((k >> BITS) > 0 || root_[i1] == NULL) { return NULL; } // 两次指针解引用，快速找到目标 Span return root_[i1]->values[i2]; } // 为指定页号设置对应的 Span 地址 // 注意：调用此函数前，必须确保已通过 Ensure 函数分配了对应的叶子节点 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; // 越界检查：如果根节点索引超出数组长度，返回失败 if (i1 >= ROOT_LENGTH) return false; // 核心：如果该根节点对应的叶子节点尚未创建，则动态分配一个 if (root_[i1] == NULL) { // 调用外部传入的分配器申请一块 Leaf 大小的内存 Leaf* leaf = reinterpret_cast<Leaf*>((*allocator_)(sizeof(Leaf))); // 若分配失败，直接返回 false if (leaf == NULL) return false; // 将新分配的叶子节点内存清零 memset(leaf, 0, sizeof(*leaf)); // 挂载到根节点上 root_[i1] = leaf; } // 将 key 移动到下一个叶子节点的起始位置 // 避免在同一个叶子节点内逐页循环检查 key = ((key >> LEAF_BITS) + 1) << LEAF_BITS; } return true; } // 预分配所有可能的页号对应的叶子节点 // 注意：这会退化成类似 PageMap1 的全量预分配模式，失去懒加载优势 void PreallocateMoreMemory() { // 确保从 0 到 2^BITS 的所有页号都有对应的叶子节点 Ensure(0, 1 << BITS); } };

3、三层结构

三层结构使用64位说明，这里采用的是三层均分，即每层存储的指针位数都是17位（64 - 13 = 51，51 / 3 = 17位）

template <int BITS> // 64位下 BITS = 51 class TCMalloc_PageMap3 { private: // 1. 均分位数：(51+2)/3 = 17位 static const int INTERIOR_BITS = (BITS + 2) / 3; static const int INTERIOR_LENGTH = 1 << INTERIOR_BITS; static const int LEAF_BITS = BITS - 2 * INTERIOR_BITS; static const int LEAF_LENGTH = 1 << LEAF_BITS; // 2. 节点定义：中间层只存指针，叶子层存真实数据 struct Node { Node* ptrs[INTERIOR_LENGTH]; }; struct Leaf { void* values[LEAF_LENGTH]; }; Node* root_; // 根节点（常驻内存） void* (*allocator_)(size_t); public: // 3. 快速查找 void* get(uintptr_t k) const { const uintptr_t i1 = k >> (LEAF_BITS + INTERIOR_BITS); // 高17位 const uintptr_t i2 = (k >> LEAF_BITS) & (INTERIOR_LENGTH - 1); // 中17位 const uintptr_t i3 = k & (LEAF_LENGTH - 1); // 低17位 // 防御性检查：懒加载可能导致中间节点为空 if ((k >> BITS) > 0 || !root_->ptrs[i1] || !root_->ptrs[i1]->ptrs[i2]) return NULL; return reinterpret_cast<Leaf*>(root_->ptrs[i1]->ptrs[i2])->values[i3]; } // 4. 懒加载 bool Ensure(uintptr_t start, size_t n) { for (uintptr_t key = start; key <= start + n - 1;) { const uintptr_t i1 = key >> (LEAF_BITS + INTERIOR_BITS); const uintptr_t i2 = (key >> LEAF_BITS) & (INTERIOR_LENGTH - 1); // 第一层：分配中间节点 if (!root_->ptrs[i1]) { root_->ptrs[i1] = NewNode(); } // 第二层：分配叶子节点 if (!root_->ptrs[i1]->ptrs[i2]) { Leaf* leaf = (Leaf*)(*allocator_)(sizeof(Leaf)); memset(leaf, 0, sizeof(*leaf)); // 将 Leaf* 强转为 Node* 存入中间层 root_->ptrs[i1]->ptrs[i2] = reinterpret_cast<Node*>(leaf); } // 5. 直接跳到下一个叶子节点的起点，避免逐页循环 key = ((key >> LEAF_BITS) + 1) << LEAF_BITS; } return true; } };

4、对比测试

接下来对比测试并发内存池和malloc/free的效率差距

编译器：Visual Studio 2022 (C++17) Release x64

操作系统：Windows 11

内存：16GB

硬件平台：AMD Ryzen 9 7945HX with Radeon Graphics (2.50 GHz)

使用哈希表映射：

使用基数树优化：

由于释放内存Page Cache回收时查询span块非常频繁，所以释放内存的优化是最明显的。

感谢观看~