分层架构的C++高并发内存池性能优化

上篇文章详细讲述了高性能内存池项目，从内存池基础概念出发，明晰其解决传统内存分配弊端、利用池化技术优化性能的核心逻辑，又深入整体框架，剖析 thread cache 线程专属、快速分配的精巧设计，central cache 协调调度、均衡资源的关键作用，以及 page cache 面向大块内存、把控物理页管理的底层支撑。本篇文章将讲述如何使⽤tcmalloc源码中基数树进⾏优化高并发内存池代码，从而使高并发内存池从效率、并发、内存管理、扩展性等多个关键方面带来显著优势，小伙伴们可以多拓展学习一下哦~

一 基数树相关知识

1.1 基数树原理

基数树是一种多叉树，以键的二进制位分段作为节点索引，通过逐层匹配位段实现快速查找。其结构随键的长度动态调整，能高效映射和管理大范围地址或数据。

1.2 基数树优缺点

优点：基数树查找效率高，支持细粒度操作，内存布局友好，适合高并发场景下的快速数据定位与管理。​
缺点：基数树可能存在空间浪费，且树结构调整较复杂，在数据规模较小时优势不明显。​

1.3 基数树对高并发内存池的作用

引入 tcmalloc 源码中的基数树，能从效率、并发、内存管理、扩展性等多方面优化高并发内存池：
1）提升内存分配与释放的效率：基数树查找快，能快速定位内存块信息，减少操作耗时，让内存池高效处理密集请求。
2）减少并发冲突：支持细粒度锁定，不同线程可并行操作不同分支，降低锁竞争，提高并发处理能力。
3）优化内存碎片管理：精细跟踪管理内存块，便于分配合适内存块及回收合并空闲块，减少碎片，提高利用率。
4）增强内存池的可扩展性：可通过动态增加树的深度管理更多内存，不增加操作复杂度，适应不同规模并发场景。
5）提升缓存利用率：节点分布规整，利于 CPU 缓存预加载，提高缓存命中率，降低操作延迟，提升性能。
6）简化内存地址映射逻辑：利用内存地址二进制特性索引，无需复杂映射机制，降低系统开销。

1.4 基数树的详细代码

//容器内存映射，页映射的目标是给定一个内存页号（Page Number），快速找到对应的内存对象或元数据。
//通过分层设计和懒加载策略，在不同场景下均能保持高效的内存管理。
#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)) {explicit TCMalloc_PageMap1() {//array_ = reinterpret_cast<void**>((*allocator)(sizeof(void*) << BITS));size_t size = sizeof(void*) << BITS;size_t alignSize = SizeClass::_RoundUp(size, 1 << PAGE_SHIFT);array_ = (void**)SystemAlloc(alignSize >> PAGE_SHIFT);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;}
};//根节点包含 32 个指针（2^5），每个指向一个叶子节点。每个叶子节点包含2 ^ (BITS - 5)个值。
// 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 nodestruct Leaf {void* values[LEAF_LENGTH];};Leaf* root_[ROOT_LENGTH];             // Pointers to 32 child nodesvoid* (*allocator_)(size_t);          // Memory allocatorpublic:typedef uintptr_t Number;//explicit TCMalloc_PageMap2(void* (*allocator)(size_t)) {explicit TCMalloc_PageMap2() {//allocator_ = allocator;memset(root_, 0, sizeof(root_));PreallocateMoreMemory();}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 overflowif (i1 >= ROOT_LENGTH)return false;// Make 2nd level node if necessaryif (root_[i1] == NULL) {//Leaf* leaf = reinterpret_cast<Leaf*>((*allocator_)(sizeof(Leaf)));//if (leaf == NULL) return false;static ObjectPool<Leaf>	leafPool;Leaf* leaf = (Leaf*)leafPool.New();memset(leaf, 0, sizeof(*leaf));root_[i1] = leaf;}// Advance key past whatever is covered by this leaf nodekey = ((key >> LEAF_BITS) + 1) << LEAF_BITS;}return true;}void PreallocateMoreMemory() {// Allocate enough to keep track of all possible pagesEnsure(0, 1 << BITS);}
};//根节点和中间节点各有2^INTERIOR_BITS个指针。叶子节点包含2^ LEAF_BITS个值。
// Three-level radix tree（三层基数树）
template <int BITS>
class TCMalloc_PageMap3 {
private:// How many bits should we consume at each interior levelstatic const int INTERIOR_BITS = (BITS + 2) / 3; // Round-upstatic const int INTERIOR_LENGTH = 1 << INTERIOR_BITS;// How many bits should we consume at leaf levelstatic const int LEAF_BITS = BITS - 2 * INTERIOR_BITS;static const int LEAF_LENGTH = 1 << LEAF_BITS;// Interior nodestruct Node {Node* ptrs[INTERIOR_LENGTH];};// Leaf nodestruct Leaf {void* values[LEAF_LENGTH];};Node* root_;                          // Root of radix treevoid* (*allocator_)(size_t);          // Memory allocatorNode* 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();}//获取页号k对应的对象指针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];}//设置页号k对应的对象指针为v。通过直接更新数组或树节点中的值实现。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;}//确保从start开始的n个页号对应的节点已分配内存。通过遍历范围内的每个页号，按需创建中间节点和叶子节点实现。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 overflowif (i1 >= INTERIOR_LENGTH || i2 >= INTERIOR_LENGTH)return false;// Make 2nd level node if necessaryif (root_->ptrs[i1] == NULL) {Node* n = NewNode();if (n == NULL) return false;root_->ptrs[i1] = n;}// Make leaf node if necessaryif (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 nodekey = ((key >> LEAF_BITS) + 1) << LEAF_BITS;}return true;}//预分配更多内存，提高后续操作的效率。void PreallocateMoreMemory() {}
};

作用：三层基数树能管理更大范围的内存地址空间，通过精细的层级索引实现高效的内存页映射与查询，同时支持懒加载和动态扩展，适配高并发场景下的并行操作，还为内存碎片跟踪回收提供支撑。

二 传统内存池与高并发内存池性能对比

对比系统标准库的 malloc/free 和自定义并发内存分配器ConcurrentAlloc/ConcurrentFree 在多线程环境下的性能差异。

//基准测试/性能测试
//对比系统标准库的 malloc/free 和自定义并发内存分配器 ConcurrentAlloc/ConcurrentFree 在多线程环境下的性能差异。
#include"ConcurrentAlloc.h"//测试标准库 malloc/free 的并发性能
// ntimes 一轮申请和释放内存的次数
// rounds 轮次
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, (size_t)malloc_costtime);printf("%u个线程并发执行%u轮次，每轮次free %u次: 花费：%u ms\n",nworks, rounds, ntimes, (size_t)free_costtime);printf("%u个线程并发malloc&free %u次，总计花费：%u ms\n",nworks, nworks * rounds * ntimes, (size_t)malloc_costtime + free_costtime);
}//试自定义并发分配器 ConcurrentAlloc/ConcurrentFree 的性能
// 单轮次申请释放次数 线程数 轮次
//与 BenchmarkMalloc 完全相同，仅将 malloc/free 替换为 ConcurrentAlloc/ConcurrentFree
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([&]() {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(ConcurrentAlloc(16));v.push_back(ConcurrentAlloc((16 + i) % 8192 + 1));//每次分配的内存大小为 (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);}});}for (auto& t : vthread){t.join();}printf("%u个线程并发执行%u轮次，每轮次concurrent alloc %u次: 花费：%u ms\n",nworks, rounds, ntimes,(size_t)malloc_costtime);printf("%u个线程并发执行%u轮次，每轮次concurrent dealloc %u次: 花费：%u ms\n",nworks, rounds, ntimes, (size_t)free_costtime);printf("%u个线程并发concurrent alloc&dealloc %u次，总计花费：%u ms\n",nworks, nworks * rounds * ntimes, (size_t)malloc_costtime + free_costtime);
}int main()
{size_t n = 10000;cout << "==========================================================" << endl;BenchmarkConcurrentMalloc(n, 4, 10);//测试自定义并发分配器（4 个线程，每线程 10 轮，每轮 10000 次操作）cout << endl << endl;BenchmarkMalloc(n, 4, 10);cout << "==========================================================" << endl;return 0;
}

注：BenchmarkMalloc相关的详细代码🔗，自定义并发内存分配器ConcurrentAlloc/ConcurrentFree 已经用基数树优化后的详细代码也在这个链接中喔(ง •_•)ง

写在最后：本篇文章通过对基数树原理、优缺点以及其在高并发内存池优化中作用的深入探讨，我们清晰地看到了基数树在高性能内存池领域的重要价值。从代码层面的详细解析，再到与传统内存池的性能对比，每一个环节都在诉说着基数树为内存池性能提升所做出的贡献。相信在未来，随着技术的不断发展，基数树以及基于它优化的高性能内存池，会在更多的场景中发挥出巨大的潜力，为各类应用的高效运行保驾护航。