【高并发内存池】调试和优化 {编译和调试；单元测试；性能分析和优化：基数树替换哈希表，源码剖析；基准测试；tcmalloc库的安装和使用}
·
一、编译和调试
编译选项
makefile文件
test: Benchmark.cpp ThreadCache.cpp CentralCache.cpp PageCache.cpp
g++ $^ -o $@ -std=c++11 -O2 -lpthread
.PHONY:clean
clean:
rm -f test
g++优化等级
- -O0：这是默认的优化等级，关闭所有优化选项，编译时间最短，但运行效率最低。（不指定优化等级就是O0）
- -O1：提供基本的优化，如函数内联和循环优化，编译时间较短，运行效率有所提升，但不如-O2和-O3。
- -O2：在-O1的基础上增加更多的优化，如指令重排和库函数调用优化，通常能显著提升性能，编译时间适中。
- -O3：提供最高级别的优化，包括循环展开、指令重排和向量化等，编译时间最长，但运行效率最高。不过，它也可能引入不稳定性。
提示：在进行调试或功能测试时使用O0默认优化等级即可；而在进行正式的性能测试或是编译release版本的程序时应该使用O2甚至O3的优化等级。
调试技巧
- 条件断点：条件满足时断点才会被触发，方便程序员精确快速地定位到出问题地位置。
- 中断运行：当程序陷入死循环时，不会触发任何断点，此时需要中断运行才能定位到死循环的位置。
- 调用栈帧：断点被触发后发现不是本层函数调用的问题，可以通过栈帧窗口逐级返回上一层调用寻找问题。
- 性能和诊断：可以从CPU占用、内存占用、函数调用计时、等待线程，四个角度分析程序。方便开发人员找出程序的性能瓶颈，并有针对性地进行优化。
二、单元测试
一般测试：看内存池是否可以正常工作
void Test_ConcurrentAlloc1()
{
int *p1 = (int *)ConcurrentAlloc(6); //maxsize:2 allocnum:1 size:0
int *p2 = (int *)ConcurrentAlloc(7); //3 2 1
int *p3 = (int *)ConcurrentAlloc(8); //3 0
int *p4 = (int *)ConcurrentAlloc(1); //4 3 2
int *p5 = (int *)ConcurrentAlloc(1); //4 1
int *p6 = (int *)ConcurrentAlloc(1); //4 0
int *p7 = (int *)ConcurrentAlloc(1); //5 4 3
*p1 = 10;
*p2 = 20;
*p3 = 30;
*p4 = 40;
cout << p1 << ": " << *p1 << endl;
cout << p2 << ": " << *p2 << endl;
cout << p3 << ": " << *p3 << endl;
cout << p4 << ": " << *p4 << endl;
ConcurrentFree(p1); //maxsize:5 size:4 span._usecount:10
ConcurrentFree(p2); //5 0 5 ** ListTooLong Release To CentralCache
ConcurrentFree(p3); //5 1 5
ConcurrentFree(p4); //5 2 5
ConcurrentFree(p5); //5 3 5
ConcurrentFree(p6); //5 4 5
ConcurrentFree(p7); //5 0 0 ** ListTooLong Release To CentralCache
}
运行结果：
特殊测试：看CentralCache中的一页内存(4KB)分配完后，是否能正常从PageCache中分裂其他页
void Test_ConcurrentAlloc2()
{
std::vector<double*> arr;
for (int i = 0; i < 512; ++i)
{
double *pd = (double *)ConcurrentAlloc(8);
arr.push_back(pd);
*pd = 8.16;
}
cout << "Last ConcurrentAlloc!" << endl;
// 在第513次申请时应该从PageCache再切分1页内存
double *pd = (double *)ConcurrentAlloc(8);
arr.push_back(pd);
*pd = 8.16;
for(auto e : arr)
{
ConcurrentFree(e);
}
}
运行结果：（不显示CentralCache的申请释放过程）
- 为什么没有合并？因为归还的是第1页，而第2页仍在使用，不满足合并条件，因此不能合并。（删除）
- 新增了ThreadCache的析构，当线程退出时会自动释放其所有FreeList中的内存块，使得后续的合并的得以进行。
特殊测试：大内存块（大于MAX_BYTES）申请
- 对于 >256KB (MAX_BYTES) 的内存空间，则会直接向PageCache申请和释放内存。
- 事实上，PageCache能管理的最大内存块是128页大小，如果申请（或释放）超过128页的内存，PageCache会直接向系统申请（或释放）。
void Test_BigAlloc()
{
int *p1 = (int *)ConcurrentAlloc(MAX_BYTES + 100);
int *p2 = (int *)ConcurrentAlloc(NPAGES << PAGE_SHIFT);
*p1 = 10;
*p2 = 20;
cout << p1 << ": " << *p1 << endl;
cout << p2 << ": " << *p2 << endl;
ConcurrentFree(p1);
ConcurrentFree(p2);
}
运行结果：
并发测试：
void AllocAndFree1()
{
std::vector<void*> arr;
for (int i = 0; i < 7; ++i)
{
int *pi = (int *)ConcurrentAlloc(6); //对齐到8
*pi = 12;
arr.push_back(pi);
}
for (int i = 0; i < 7; ++i)
{
ConcurrentFree(arr[i]);
}
}
void AllocAndFree2()
{
std::vector<void*> arr;
for (int i = 0; i < 7; ++i)
{
int *pi = (int *)ConcurrentAlloc(10); //对齐到16
*pi = 12;
arr.push_back(pi);
}
for (int i = 0; i < 7; ++i)
{
ConcurrentFree(arr[i]);
}
}
void Test_ConcurrentAlloc3()
{
std::thread thread1(AllocAndFree1);
std::thread thread2(AllocAndFree2);
thread1.join();
thread2.join();
}
运行结果：
三、性能分析和优化
3.1 性能瓶颈分析
性能分析：
得出结论：该项目的性能瓶颈在于idSpanMap的查找和锁的竞争
3.2 优化方案
3.2.1 基数树替换哈希表
优化idSpanMap的数据结构，从哈希表改为基数树，下面简单介绍一下基数树：
基数树是如何提高效率的？
- 基数树的查找效率更高：在哈希表中，不同的键可能映射到相同的哈希桶，从而导致冲突。而基数树为每个页号都分配了空间，不存在冲突。因此虽然理论上二者的查找效率都为O(1)，但实际基数树的查找效率更高。
- 基数树几乎没有扩容消耗：当哈希表的负载因子超过阈值，通常会触发扩容操作，不仅需要申请和释放内存还要重新计算每个元素的哈希地址，消耗很大。而基数树一般是提前开好空间的，即使在多层基数树中存在未开空间的情况，也只是需要申请内存。
- 基数树的设计可以减少或避免一些锁的使用
- 新旧分离：哈希表或红黑树在进行写操作时可能会改变结构：哈希表会进行扩容，红黑树也要进行旋转操作自平衡。在并发访问的过程中如果发生数据结构的改变，则必然存在线程安全问题。而基数树的每个页号都有一个唯一且固定的映射存储位置，空间结构一旦建立，就不会发生变化（对于已经开好的空间），也就是说idSpanMap（基数树）申请新的Node或是Leaf，不会影响旧空间中的映射关系。
- 读写分离：Pageid和span的映射关系在span分裂和合并的过程当中可能会发生改变。但是只有在PageCache的申请和释放操作中存在对idSpanMap（基数树）的写操作，而PageCache的申请和释放操作一定是串行执行的，也就是说同一时间只能有一个线程对idSpanMap（基数树）进行写操作，且修改的是未分配或是将要分配的page的映射，因此不影响在
ConcurrentFree和CentralCache::ReleaseListToSpans中的对idSpanMap的并发读操作（已分配的将要释放的page的映射）。
3.2.2 源码剖析
具体解析看代码中的注释即可！
注意！以下代码定义了三个结构：单层基数树、双层基数树、三层基数树；我们使用三层基数树作为idSpanMap的存储结构，适用于32位和64位平台。
TCMalloc_PageMap1
PageMap.hpp
单层基数树，按页号直接哈希（适用于32位平台）
特点：简单、快捷
#pragma once
#include "Common.hpp"
#include "ObjectPool.hpp"
// Single-level array
// 大小：32位下：4*2^20=4MB；64位下：4*2^52就太大了
template <int BITS> // 存储页号需要多少位 32位下：32-PAGE_SHIFT=20 64位下：64-PAGE_SHIFT=52
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 = AlineRule::_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'.
// k就是页号，v就是Span*
void set(Number k, void *v)
{
array_[k] = v;
}
};
TCMalloc_PageMap2
PageMap.hpp
双层基数树，两次哈希（适用于32位平台）
特点：相比单层基数树更能节省空间，但也相对更复杂
// Two-level radix tree
// 双层基数树整体占用的空间和单层相同：4*2^5*2^15=4MB（32位下）
// 但是如果是单层基数树需要将4MB空间一次开好，而双层只需要先开好第一层，其余的空间用多少开多少
// 对于32位：4*2^5 + n*4*2^15；对于64位：4*2^5 + n*4*2^47(第二层的空间还是太大了)
template <int BITS>
class TCMalloc_PageMap2
{
private:
// Put 32 entries in the root and (2^BITS)/32 entries in each leaf.
// 取页号的前5位作为第一层索引
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)) {
explicit TCMalloc_PageMap2()
{
// allocator_ = allocator;
memset(root_, 0, sizeof(root_));
PreallocateMoreMemory();
}
void *get(Number k) const
{
const Number i1 = k >> LEAF_BITS; // 取页号的高5位
const Number i2 = k & (LEAF_LENGTH - 1); // 取页号的低15位
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;
}
// Ensure函数用于确保从start页号往后的n页，在基数树中的索引结构内存都已开好
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;
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 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);
}
};
TCMalloc_PageMap3
PageMap.hpp
三层基数树，三次哈希（适用于32位和64位平台）
// Three-level radix tree
// 大小：对于64位：8*2^18 + m*8*2^18 + n*8*2^16（2MB + m*2MB + n*0.5MB）
template <int BITS>
class TCMalloc_PageMap3
{
private:
// How many bits should we consume at each interior level
// 前两层各占18位（64位下）
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
// 第三层占16位
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)));
// 按页对齐，因为32位下有不足1页的情况
size_t alignSize = AlineRule::_RoundUp(sizeof(Node), 1 << PAGE_SHIFT);
Node *result = (Node *)SystemAlloc(alignSize >> PAGE_SHIFT);
if (result != NULL)
{
memset(result, 0, sizeof(*result));
}
return result;
}
public:
typedef uintptr_t Number;
// explicit TCMalloc_PageMap3(void *(*allocator)(size_t))
explicit TCMalloc_PageMap3()
{
// allocator_ = allocator;
root_ = NewNode(); // 先把第一层开好
}
void *get(Number k) const
{
const Number i1 = k >> (LEAF_BITS + INTERIOR_BITS); // 取页号的高18位
const Number i2 = (k >> LEAF_BITS) & (INTERIOR_LENGTH - 1); // 取页号的中间18位
const Number i3 = k & (LEAF_LENGTH - 1); // 取页号的低16位
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);
Ensure(k, 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)
{
// cout << "Make 2nd level node" << endl; //degug
Node *n = NewNode();
if (n == NULL)
return false;
root_->ptrs[i1] = n;
}
// Make leaf node if necessary
if (root_->ptrs[i1]->ptrs[i2] == NULL)
{
// cout << "SystemAlloc(sizeof(Leaf) >> PAGE_SHIFT)" << endl; //degug
// Leaf *leaf = reinterpret_cast<Leaf *>((*allocator_)(sizeof(Leaf)));
size_t alignSize = AlineRule::_RoundUp(sizeof(Leaf), 1 << PAGE_SHIFT); // 按页对齐
Leaf *leaf = (Leaf *)SystemAlloc(alignSize >> PAGE_SHIFT);
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()
{
}
};
四、基准测试
测试代码：
#include "ConcurrentAlloc.hpp"
#include <vector>
#include <atomic>
// ntimes 一轮申请和释放内存的次数
// rounds 轮次
// size_t pageCount = 0; // debug
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("%lu个线程并发执行%lu轮次，每轮次malloc %lu次: 花费：%lu ms\n",
nworks, rounds, ntimes, (size_t)malloc_costtime / 1000);
printf("%lu个线程并发执行%lu轮次，每轮次free %lu次: 花费：%lu ms\n",
nworks, rounds, ntimes, (size_t)free_costtime / 1000);
printf("%lu个线程并发malloc&free %lu次，总计花费：%lu ms\n",
nworks, nworks * rounds * ntimes, ((size_t)malloc_costtime + (size_t)free_costtime) / 1000);
}
// 单轮次申请释放次数 线程数 轮次
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));
}
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("%lu个线程并发执行%lu轮次，每轮次concurrent alloc %lu次: 花费：%lu ms\n",
nworks, rounds, ntimes, (size_t)malloc_costtime / 1000);
printf("%lu个线程并发执行%lu轮次，每轮次concurrent dealloc %lu次: 花费：%lu ms\n",
nworks, rounds, ntimes, (size_t)free_costtime / 1000);
printf("%lu个线程并发concurrent alloc&dealloc %lu次，总计花费：%lu ms\n",
nworks, nworks * rounds * ntimes, ((size_t)malloc_costtime + (size_t)free_costtime) / 1000);
}
int main()
{
size_t n = 10000;
cout << "==========================================================" << endl;
BenchmarkConcurrentMalloc(n, 4, 10);
// cout << pageCount << endl; // debug
cout << endl
<< endl;
BenchmarkMalloc(n, 4, 10);
cout << "==========================================================" << endl;
return 0;
}
测试结果：
Linux64 多线程单桶测试：
ConcurrentAlloc和malloc的多线程单桶测试结果相近！
Linux64 多线程分桶测试：
ConcurrentAlloc多线程分桶内存分配的效率显著提升！
五、tcmalloc库的安装和使用
gperftools工具集
gperftools是谷歌开源的性能分析工具包，主要包括了以下几个组件：
-
tcmalloc：一种内存分配器，旨在提高内存分配的效率和性能。tcmalloc通过优化内存管理算法和减少内存碎片来提高内存分配速度和减少内存占用。在高并发、大内存分配和多线程场景下，tcmalloc可以提升系统的性能表现。
-
Heap Profiler：一种堆分析工具，用于查看程序的内存分配情况和发现潜在的内存泄漏问题。Heap Profiler可以生成程序的内存使用情况报告，帮助开发人员定位内存分配问题并优化内存使用。
-
CPU Profiler：一种CPU性能分析工具，用于分析程序的CPU使用情况和性能瓶颈。CPU Profiler可以生成程序的CPU调用图和性能报告，帮助开发人员优化程序的性能表现。
-
Heap Checker：一种堆检查工具，用于检测程序中的内存错误和潜在的内存问题。Heap Checker可以帮助开发人员及时发现并修复内存错误，提高程序的稳定性和可靠性。
总的来说，gperftools提供了一套全面的性能分析工具，帮助开发人员优化程序的内存管理、CPU使用和性能表现，提高程序的稳定性和性能。
下载
GitHub - gperftools/gperftools: Main gperftools repository
- 直接下载最新的gperftools-2.16（Release），将链接中的
tar.gz或是zip文件下载到本地。 - 解压缩：
- tar.gz文件：
tar xzvf gperftools-2.16.tar.gz - zip文件：
unzip gperftools-2.16.zip
- tar.gz文件：
安装
cd gperftools-2.16
./configure
make
make install
注意：
- 如果遇到permission denied报错，表示没有权限，拒绝访问，请切换root或使用sudo提升权限。
- 安装完成后在目录
/usr/local/lib下，查看libtcmalloc_minimal.*等动静态库文件(.so, .a)是否成功安装。./configure命令用于配置编译选项，默认情况下编译生成gperftools中的所有组件（编译时间较长）。如果只想生成tcmalloc库，则通过命令./configure --disable-cpu-profiler --disable-heap-profiler --disable-heap-checker --enable-minimal进行最小化构建。
链接使用
- 配置动态库的搜索路径：
echo /usr/local/lib > /etc/ld.so.conf.d/libtcmalloc.conf - 更新动态链接器的运行时链接库缓存：
ldconfig - 编译链接：
g++ $^ -o $@ -std=c++11 -O2 -lpthread -ltcmalloc - 使用：无需修改代码，只要编译选项中链接了tcmalloc库，程序会自动使用tcmalloc替换ptmalloc(glibc)实现内存管理。
使用场景
-
高并发场景：在高并发的系统中，tcmalloc可以有效地减少内存碎片和提升内存分配效率，从而减少锁竞争和内存占用，提高系统的并发性能。
-
大内存分配场景：在需要大量内存分配的场景下，tcmalloc可以更高效地利用内存，减少内存分配的性能开销，从而提升系统的性能。
-
多线程场景：在多线程环境下，tcmalloc可以更好地处理多线程之间的内存管理和竞争问题，提高系统的并发性能和稳定性。
-
高性能需求场景：在对内存分配性能要求较高的场景下，tcmalloc可以更高效地管理内存，减少内存碎片，提高系统的性能表现。
参考文章
更多推荐
18
0- 0
已为社区贡献3条内容
所有评论(0)