C++项目：高并发内存池_下

目录

8. thread cache回收内存

9. central cache回收内存

10. page cache回收内存

11. 大于256KB的内存申请和释放

11.1 申请

11.2 释放

12. 使用定长内存池脱离使用new

13. 释放对象时优化成不传对象大小

14. 多线程环境下对比malloc测试

15. 调试和复杂问题的调试技巧

16. 性能瓶颈分析

17. 使用基数树进行优化

---

8. thread cache回收内存

当某个线程申请的对象不用了，可以将其释放给thread cache，然后thread cache将该对象插入到对应哈希桶的自由链表当中即可。

但是随着线程不断的释放，对应自由链表的长度也会越来越长，这些内存堆积在一个thread cache中就是一种浪费，应该将这些内存还给central cache，这样一来，这些内存对其他线程来说也是可申请的，因此当thread cache某个桶当中的自由链表太长时可以进行一些处理。

如果thread cache某个桶当中自由链表的长度超过它一次批量向central cache申请的对象个数，那么此时就要把该自由链表当中的这些对象还给central 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 cacheif (_freeLists[index].Size() >= _freeLists[index].MaxSize()) // 要增加Size接口{ListTooLong(_freeLists[index], size);}
}void ThreadCache::ListTooLong(FreeList& list, size_t size)
{void* start = nullptr;void* end = nullptr;list.PopRange(start, end, list.MaxSize()); // 取批量的内存，要增加PopRange接口CentralCache::GetInstance()->ReleaseListToSpans(start, size);
}

要提供Size接口，并维护_size，再增加PopRange接口

调用的地方增加传入参数（倒数第三行）：

当thread cache的某个自由链表过长时，实际就是把这个自由链表当中全部的对象都还给central cache了，但这里在设计PopRange接口时还是设计的是取出指定个数的对象，因为在某些情况下当自由链表过长时，我们可能并不一定想把链表中全部的对象都取出来还给central cache，这样设计就是为了增加代码的可修改性。

当判断thread cache是否应该还对象给central cache时，还可以综合考虑每个thread cache整体的大小。比如当某个thread cache的总占用大小超过一定阈值时，就将该thread cache当中的对象还一些给central cache，这样就尽量避免了某个线程的thread cache占用太多的内存。对于这一点，在tcmalloc当中就是考虑到了的，但这里就简化了。

---

9. central cache回收内存

当thread cache中某个自由链表太长时，会将自由链表当中的这些对象还给central cache中的span。需要注意的是，还给central cache的这些对象不一定都是属于同一个span的。central cache中的每个哈希桶当中可能都不止一个span，因此计算出还回来的对象应该还给central cache的哪一个桶后，还需要知道还回来的对象到底应该还给这个桶当中的哪一个span。

如何根据对象的地址得到对象所在的页号？

某个页当中的所有地址除以页的大小都等该页的页号。比如假设一页的大小是100，那么地址0~99都属于第0页，它们除以100都等于0，而地址100~199都属于第1页，它们除以100都等于1。

如何找到一个对象对应的span？

虽然现在可以通过对象的地址得到其所在的页号，但是还是不能知道这个对象到底属于哪一个span。因为一个span管理的可能是多个页。

为了解决这个问题，可以建立页号和span之间的映射。由于这个映射关系在page cache进行span的合并时也需要用到，因此直接将其存放到page cache里面。这时就需要在PageCache类当中添加一个映射关系了，这里可以用STL当中的unordered_map进行实现，并且添加一个函数接口，用于让central cache获取这里的映射关系。（下面代码中只展示了PageCache

类当中新增的成员）

class PageCache
{
public:// 获取从对象到span的映射Span* MapObjectToSpan(void* obj);
private:std::unordered_map<PAGE_ID, Span*> _idSpanMap;
};

每当page cache分配span给central cache时，都需要记录一下页号和span之间的映射关系。此后当thread cache还对象给central cache时，才知道应该具体还给哪一个span。

因此当central cache在调用NewSpan接口向page cache申请k页的span时，page cache在返回这个k页的span给central cache之前，应该建立这k个页号与该span之间的映射关系，这里改一下之前写的NewSpan（多了两个for循环）：

Span* PageCache::NewSpan(size_t k) // 获取一个K页的span
{assert(k > 0 && k < NPAGES);if (!_spanLists[k].Empty()) // 先检查第k个桶里面有没有span{//return _spanLists[k].opFront(); // 有就直接返回Span* kSpan = _spanLists[k].PopFront();//建立页号与span的映射，方便central cache回收小块内存时查找对应的spanfor (PAGE_ID i = 0; i < kSpan->_n; i++){_idSpanMap[kSpan->_pageId + i] = kSpan;}return kSpan;}// 检查一下后面的桶里面有没有span，如果有->进行切分for (size_t i = k + 1; i < NPAGES; ++i){if (!_spanLists[i].Empty()){Span* nSpan = _spanLists[i].PopFront();Span* kSpan = new Span;// 在nSpan的头部切一个k页span下来返回，nSpan再挂到对应映射的位置kSpan->_pageId = nSpan->_pageId; // _pageId类似地址kSpan->_n = k;nSpan->_pageId += k; // 起始页的页号往后走nSpan->_n -= k; // 页数减等k_spanLists[nSpan->_n].PushFront(nSpan);//建立页号与span的映射，方便central cache回收小块内存时查找对应的spanfor (PAGE_ID i = 0; i < kSpan->_n; i++){_idSpanMap[kSpan->_pageId + i] = kSpan;}return kSpan;}}// 走到这说明后面没有大页的span了->去找堆要一个128页的spanSpan* bigSpan = new Span;void* ptr = SystemAlloc(NPAGES - 1);bigSpan->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT; // 转换为页号bigSpan->_n = NPAGES - 1;_spanLists[bigSpan->_n].PushFront(bigSpan);return NewSpan(k); // 把新申请的span插入后再递归调用一次自己(避免代码重复)
}

此时就可以通过对象的地址找到该对象对应的span了，直接将该对象的地址除以页的大小得到页号，然后在unordered_map当中找到其对应的span即可。

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;}
}

通过某个页号查找其对应的span时，该页号与其span之间的映射一定是建立过的，如果此时没有在unordered_map当中找到，则说明之前的代码逻辑有问题，因此当没有找到对应的span时可以直接用断言结束程序，以表明程序逻辑出错。

---

此时就能写central cache的回收内存了，当thread cache还对象给central cache时，就可以依次遍历这些对象，将这些对象插入到其对应span的自由链表当中，并且及时更新该span的_usseCount计数即可。

在thread cache还对象给central cache的过程中，如果central cache中某个span的_useCount减到0时，说明这个span分配出去的对象全部都还回来了，那么此时就可以将这个span再进一步还给page cache。

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的切分出去的所有小块内存都回来了// 这个span就可以再回收给page cache，pagecache可以再尝试去做前后页的合并if (span->_useCount == 0){_spanLists[index].Erase(span);span->_freeList = nullptr;span->_next = nullptr;span->_prev = nullptr;// 释放span给page cache时，使用page cache的锁就可以了_spanLists[index]._mtx.unlock(); // 这时把桶锁解掉PageCache::GetInstance()->_pageMtx.lock();PageCache::GetInstance()->ReleaseSpanToPageCache(span);PageCache::GetInstance()->_pageMtx.unlock();_spanLists[index]._mtx.lock();}start = next;}_spanLists[index]._mtx.unlock();
}

需要注意，如果要把某个span还给page cache，需要先将这个span从central cache对应的双链表中移除，然后再将该span的自由链表置空，因为page cache中的span是不需要切分成一个个的小对象的，以及该span的前后指针也都应该置空，因为之后要将其插入到page cache对应的双链表中。但span当中记录的起始页号以及它管理的页数是不能清除的，否则对应内存块就找不到了。

并且在central cache还span给page cache时也存在锁的问题，此时需要先将central cache中对应的桶锁解掉，然后再加上page cache的大锁之后才能进入page cache进行相关操作，当处理完毕回到central cache时，除了将page cache的大锁解掉，还需要立刻获得central cache对应的桶锁，然后将还未还完对象继续还给central cache中对应的span。

---

10. page cache回收内存

如果central cache中有某个span的_useCount减到0了，那么central cache就需要将这个span还给page cache了。

这个过程看似是非常简单的，page cache只需将还回来的span挂到对应的哈希桶上就行了。但实际为了缓解内存碎片的问题，page cache还需要尝试将还回来的span与其他空闲的span进行合并。

合并的过程可以分为向前合并和向后合并。如果还回来的span的起始页号是num，该span所管理的页数是n。那么在向前合并时，就需要判断第num-1页对应span是否空闲，如果空闲则可以将其进行合并，并且合并后还需要继续向前尝试进行合并，直到不能进行合并为止。而在向后合并时，就需要判断第num+n页对应的span是否空闲，如果空闲则可以将其进行合并，并且合并后还需要继续向后尝试进行合并，直到不能进行合并为止。

因此page cache在合并span时，是需要通过页号获取到对应的span的，这就是要把页号与span之间的映射关系存储到page cache的原因。

需要注意的是，当通过页号找到其对应的span时，这个span此时可能挂在page cache，也可能挂在central cache。而在合并时只能合并挂在page cache的span，因为挂在central cache的span当中的对象正在被其他线程使用。

可是我们不能通过span结构当中的_useCount成员，来判断某个span到底是在central cache还是在page cache。因为当central cache刚向page cache申请到一个span时，这个span的_useCount就是等于0的，这时可能当我们正在对该span进行切分的时候，page cache就把这个span拿去进行合并了，这显然是不合理的。

所以可以在span结构中再增加一个_isUse成员，用于标记这个span是否正在被使用，而当一个span结构被创建时我们默认该span是没有被使用的。

struct Span // 管理多个连续页大块内存跨度的结构
{PAGE_ID _pageId = 0; // 大块内存起始页的页号size_t  _n = 0;      // 页的数量Span* _next = nullptr;	// 双向链表的结构Span* _prev = nullptr;size_t _useCount = 0; // 切好小块内存，被分配给thread cache的计数void* _freeList = nullptr;  // 切好的小块内存的自由链表bool _isUse = false; // 是否在被使用
};

当central cache向page cache申请到一个span时，需要立即将该span的_isUse改为true。

当central cache将某个span还给page cache时，也就需要将该span的_isUse改成false。

---

在合并page cache当中的span时，需要通过页号找到其对应的span，而一个span是在被分配给central cache时，才建立的各个页号与span之间的映射关系，因此page cache当中的span也需要建立页号与span之间的映射关系。

与central cache中的span不同的是，在page cache中，只需建立一个span的首尾页号与该span之间的映射关系。因为当一个span在尝试进行合并时，如果是往前合并，那么只需要通过一个span的尾页找到这个span，如果是向后合并，那么只需要通过一个span的首页找到这个span。也就是说，在进行合并时我们只需要用到span与其首尾页之间的映射关系就够了。

因此申请k页的span时，如果是将n页的span切成了一个k页的span和一个n-k页的span，除了需要建立k页span中每个页与该span之间的映射关系之外，还需要建立剩下的n-k页的span与其首尾页之间的映射关系。

Span* PageCache::NewSpan(size_t k) // 获取一个K页的span
{assert(k > 0 && k < NPAGES);if (!_spanLists[k].Empty()) // 先检查第k个桶里面有没有span{//return _spanLists[k].PopFront(); // 有就直接返回Span* kSpan = _spanLists[k].PopFront();//建立页号与span的映射，方便central cache回收小块内存时查找对应的spanfor (PAGE_ID i = 0; i < kSpan->_n; i++){_idSpanMap[kSpan->_pageId + i] = kSpan;}return kSpan;}// 检查一下后面的桶里面有没有span，如果有->进行切分for (size_t i = k + 1; i < NPAGES; ++i){if (!_spanLists[i].Empty()){Span* nSpan = _spanLists[i].PopFront();Span* kSpan = new Span;// 在nSpan的头部切一个k页span下来返回，nSpan再挂到对应映射的位置kSpan->_pageId = nSpan->_pageId; // _pageId类似地址kSpan->_n = k;nSpan->_pageId += k; // 起始页的页号往后走nSpan->_n -= k; // 页数减等k_spanLists[nSpan->_n].PushFront(nSpan); // 将剩下的挂到对应映射的位置//存储nSpan的首尾页号与nSpan之间的映射，方便page cache合并span时进行前后页的查找_idSpanMap[nSpan->_pageId] = nSpan;_idSpanMap[nSpan->_pageId + nSpan->_n - 1] = nSpan;//建立页号与span的映射，方便central cache回收小块内存时查找对应的spanfor (PAGE_ID i = 0; i < kSpan->_n; i++){_idSpanMap[kSpan->_pageId + i] = kSpan;}return kSpan;}}// 走到这说明后面没有大页的span了->去找堆要一个128页的spanSpan* bigSpan = new Span;void* ptr = SystemAlloc(NPAGES - 1);bigSpan->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT; // 转换为页号bigSpan->_n = NPAGES - 1;_spanLists[bigSpan->_n].PushFront(bigSpan);return NewSpan(k); // 把新申请的span插入后再递归调用一次自己(避免代码重复)
}

此时page cache当中的span就都与其首尾页之间建立了映射关系，现在就可以进行span的合并了，需要注意的是，在向前或向后进行合并的过程中：

• 如果没有通过页号获取到其对应的span，说明对应到该页的内存块还未申请，此时需要停止合并。
• 如果通过页号获取到了其对应的span，但该span处于被使用的状态，那我们也必须停止合并。
• 如果合并后大于128页则不能进行本次合并，因为page cache无法对大于128页的span进行管理。

void PageCache::ReleaseSpanToPageCache(Span* span)
{while (true) // 对span前后的页，尝试进行合并，缓解内存碎片问题{PAGE_ID prevId = span->_pageId - 1;auto ret = _idSpanMap.find(prevId);if (ret == _idSpanMap.end()) // 前面的页号没有，不合并{break;}Span* prevSpan = ret->second; // 前面相邻页的span在使用，不合并if (prevSpan->_isUse == true){break;}if (prevSpan->_n + span->_n > NPAGES - 1) // 合并出超过128页的span没办法管理，不合并{break;}span->_pageId = prevSpan->_pageId;span->_n += prevSpan->_n;_spanLists[prevSpan->_n].Erase(prevSpan);delete prevSpan;}while (true) // 向后合并{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;}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;
}

在合并span时，由于这个span是在page cache的某个哈希桶的双链表当中的，因此在合并后需要将其从对应的双链表中移除，然后再将这个被合并了的span结构进行delete。

除此之外，在合并结束后，除了将合并后的span挂到page cache对应哈希桶的双链表当中，还需要建立该span与其首位页之间的映射关系，便于此后合并出更大的span。

测试代码：

void TestConcurrentAlloc3()
{void* p1 = ConcurrentAlloc(6);void* p2 = ConcurrentAlloc(8);void* p3 = ConcurrentAlloc(1);void* p4 = ConcurrentAlloc(7);void* p5 = ConcurrentAlloc(8);cout << p1 << endl; // 申请的小块内存正好是连续的cout << p2 << endl;cout << p3 << endl;cout << p4 << endl;cout << p5 << endl;ConcurrentFree(p1, 6);ConcurrentFree(p2, 8);ConcurrentFree(p3, 1);ConcurrentFree(p4, 7);ConcurrentFree(p5 ,8);cout << endl;cout << p1 << endl;cout << p2 << endl;cout << p3 << endl;cout << p4 << endl;cout << p5 << endl;
}void MultiThreadAlloc1()
{std::vector<void*> v;for (size_t i = 0; i < 7; ++i){void* ptr = ConcurrentAlloc(6);v.push_back(ptr);}for (auto e : v){ConcurrentFree(e, 6);}
}
void MultiThreadAlloc2()
{std::vector<void*> v;for (size_t i = 0; i < 7; ++i){void* ptr = ConcurrentAlloc(16);v.push_back(ptr);}for (auto e : v){ConcurrentFree(e, 16);}
}
void TestMultiThread()
{std::thread t1(MultiThreadAlloc1);std::thread t2(MultiThreadAlloc2);t1.join();t2.join();
}

至此我们便完成了这三个对象的申请和释放流程。

---

11. 大于256KB的内存申请和释放

11.1 申请

每个线程的thread cache是用于申请小于等于256KB的内存的，而对于大于256KB的内存，我们可以考虑直接向page cache申请，但page cache中最大的页也就只有128页，因此如果是大于128页的内存申请，就只能直接向堆申请了。

当申请的内存大于256KB时，虽然不是从thread cache进行获取，但在分配内存时也是需要进行向上对齐的，对于大于256KB的内存我们可以直接按页进行对齐。

之前实现RoundUp函数时，对传入字节数大于256KB的情况直接做了断言处理，因此这里需要对RoundUp函数稍作修改。

	static inline size_t RoundUp(size_t size) // 返回size对齐以后的值{if (size <= 128) {return _RoundUp(size, 8);}else if (size <= 1024) {return _RoundUp(size, 16);}else if (size <= 8 * 1024) {return _RoundUp(size, 128);}else if (size <= 64 * 1024) {return _RoundUp(size, 1024);}else if (size <= 256 * 1024) {return _RoundUp(size, 8 * 1024);}else{ //大于256KB的按页对齐return _RoundUp(size, 1 << PAGE_SHIFT);}}

现在对于之前的申请逻辑就需要进行修改了，当申请对象的大小大于256KB时，就不用向thread cache申请了，这时先计算出按页对齐后实际需要申请的页数，然后通过调用NewSpan申请指定页数的span即可。

static void* ConcurrentAlloc(size_t size)
{if (size > MAX_BYTES) //大于256KB的内存申请{//计算出对齐后需要申请的页数size_t alignSize = SizeClass::RoundUp(size);size_t kPage = alignSize >> PAGE_SHIFT;//向page cache申请kPage页的spanPageCache::GetInstance()->_pageMtx.lock();Span* span = PageCache::GetInstance()->NewSpan(kPage);span->_objSize = size;PageCache::GetInstance()->_pageMtx.unlock();void* ptr = (void*)(span->_pageId << PAGE_SHIFT);return ptr;}else{// 通过TLS，每个线程无锁地获取自己专属的ThreadCache对象if (pTLSThreadCache == nullptr){//pTLSThreadCache = new ThreadCache;static std::mutex tcMtx;static ObjectPool<ThreadCache> tcPool;tcMtx.lock();pTLSThreadCache = tcPool.New();tcMtx.unlock();}//cout << std::this_thread::get_id() << ":" << pTLSThreadCache << endl;return pTLSThreadCache->Allocate(size);}
}

也就是说，申请大于256KB的内存时，会直接调用page cache当中的NewSpan函数进行申请，因此这里要再对NewSpan函数进行改造，当需要申请的内存页数大于128页时，就直接向堆申请对应页数的内存块。而如果申请的内存页数是小于128页的，那就在page cache中进行申请，因此当申请大于256KB的内存调用NewSpan函数时也是需要加锁的，因为可能是在page cache中进行申请的。

---

11.2 释放

当释放对象时，我们需要判断释放对象的大小：

当释放对象时，需要先找到该对象对应的span，但是在释放对象时只知道该对象的起始地址。这也就是在申请大于256KB的内存时，也要给申请到的内存建立span结构，并建立起始页号与该span之间的映射关系的原因。此时就可以通过释放对象的起始地址计算出起始页号，进而通过页号找到该对象对应的span。

static void ConcurrentFree(void* ptr, size_t size)
{if (size > MAX_BYTES) //大于256KB的内存释放{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);}
}

因此page cache在回收span时也需要进行判断，如果该span的大小是小于等于128页的，那么直接还给page cache进行了，page cache会尝试对其进行合并。而如果该span的大小是大于128页的，那么说明该span是直接向堆申请的，我们直接将这块内存释放给堆，然后将这个span结构进行delete就行了。

直接向堆申请内存时我们调用的接口是VirtualAlloc，与之对应的将内存释放给堆的接口叫做VirtualFree，而Linux下的brk和mmap对应的释放接口叫做sbrk和unmmap。此时也可以将这些释放接口封装成一个叫做SystemFree的接口，需要将内存释放给堆时直接调用SystemFree即可。

inline static void SystemFree(void* ptr) // 直接将内存还给堆
{
#ifdef _WIN32VirtualFree(ptr, 0, MEM_RELEASE);
#else//linux下sbrk unmmap等
#endif
}

---

12. 使用定长内存池脱离使用new

tcmalloc是要在高并发场景下替代malloc进行内存申请的，因此tcmalloc在实现的时，其内部是不能调用malloc函数的，我们当前的代码中存在通过new获取到的内存，而new在底层实际上就是封装了malloc。

为了完全脱离掉malloc函数，此时之前实现的定长内存池就起作用了，代码中使用new时基本都是为Span结构的对象申请空间，而span对象基本都是在page cache层创建的，因此可以在PageCache类当中定义一个_spanPool，用于span对象的申请和释放。

class PageCache
{
public://...
private://...ObjectPool<Span> _spanPool;
};

然后将代码中使用new的地方替换为调用定长内存池当中的New函数，将代码中使用delete的地方替换为调用定长内存池当中的Delete函数。

Span* span = _spanPool.New(); //申请span对象_spanPool.Delete(span); //释放span对象

注意，当使用定长内存池当中的New函数申请Span对象时，New函数通过定位new也是对Span对象进行了初始化的。ConcurrentAlloc接口改一下：

		// 通过TLS，每个线程无锁地获取自己专属的ThreadCache对象if (pTLSThreadCache == nullptr){//pTLSThreadCache = new ThreadCache;static std::mutex tcMtx;static ObjectPool<ThreadCache> tcPool;tcMtx.lock();pTLSThreadCache = tcPool.New();tcMtx.unlock();}cout << std::this_thread::get_id() << ":" << pTLSThreadCache << endl;return pTLSThreadCache->Allocate(size);

这里我们将用于申请ThreadCache类对象的定长内存池定义为静态的，保持全局只有一个，让所有线程创建自己的thread cache时，都在个定长内存池中申请内存就行了。

但注意在从该定长内存池中申请内存时需要加锁，防止多个线程同时申请自己的ThreadCache对象而导致线程安全问题。

---

最后在SpanList的构造函数中也用到了new，因为SpanList是带头循环双向链表，所以在构造期间我们需要申请一个span对象作为双链表的头结点。

class SpanList // 带头双向循环链表
{
public:SpanList(){_head = _spanPool.New();_head->_next = _head;_head->_prev = _head;}
private:Span* _head;static ObjectPool<Span> _spanPool;
};

由于每个span双链表只需要一个头结点，因此将这个定长内存池定义为静态的，保持全局只有一个，让所有span双链表在申请头结点时，都在一个定长内存池中申请内存就行了。

---

13. 释放对象时优化成不传对象大小

使用malloc函数申请内存时，需要指明申请内存的大小；而使用free函数释放内存时，只需要传入指向这块内存的指针即可。

目前实现的内存池，在释放对象时除了需要传入指向该对象的指针，还需要传入该对象的大小。

原因如下：

• 如果释放的是大于256KB的对象，需要根据对象的大小来判断这块内存到底应该还给page cache，还是应该直接还给堆。
• 如果释放的是小于等于256KB的对象，需要根据对象的大小计算出应该还给thread cache的哪一个哈希桶。

如果我们也想做到，在释放对象时不用传入对象的大小，那么就需要建立对象地址与对象大小之间的映射。由于现在可以通过对象的地址找到其对应的span，而span的自由链表中挂的都是相同大小的对象。

因此可以在Span结构中再增加一个_objSize成员，该成员代表着这个span管理的内存块被切成的一个个对象的大小。

struct Span // 管理多个连续页大块内存跨度的结构
{PAGE_ID _pageId = 0; // 大块内存起始页的页号size_t  _n = 0;      // 页的数量Span* _next = nullptr;	// 双向链表的结构Span* _prev = nullptr;size_t _useCount = 0; // 切好小块内存，被分配给thread cache的计数void* _freeList = nullptr;  // 切好的小块内存的自由链表bool _isUse = false; // 是否在被使用size_t _objSize = 0; //切好的小对象的大小
};

而所有的span都是从page cache中拿出来的，因此每当我们调用NewSpan获取到一个k页的span时，就应该将这个span的_objSize保存下来。

代码中有两处，一处是在central cache中获取非空span时，如果central cache对应的桶中没有非空的span，此时会调用NewSpan获取一个k页的span；另一处是当申请大于256KB内存时，会直接调用NewSpan获取一个k页的span。

此时当我们释放对象时，就可以直接从对象的span中获取到该对象的大小，准确来说获取到的是对齐以后的大小。

static void ConcurrentFree(void* ptr)
{Span* span = PageCache::GetInstance()->MapObjectToSpan(ptr);size_t size = span->_objSize;if (size > MAX_BYTES) //大于256KB的内存释放{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);}
}

---

读取映射关系时的加锁问题

我们将页号与span之间的映射关系是存储在PageCache类当中的，当我们访问这个映射关系时是需要加锁的，因为STL容器是不保证线程安全的。

对于当前代码来说，如果我们此时正在page cache进行相关操作，那么访问这个映射关系是安全的，因为当进入page cache之前是需要加锁的，因此可以保证此时只有一个线程在进行访问。

但如果我们是在central cache访问这个映射关系，或是在调用ConcurrentFree函数释放内存时访问这个映射关系，那么就存在线程安全的问题。因为此时可能其他线程正在page cache当中进行某些操作，并且该线程此时可能也在访问这个映射关系，因此在page cache外部访问这个映射关系时是需要加锁的。

实际就是在调用page cache对外提供访问映射关系的函数时需要加锁，这里可以考虑使用C++当中的unique_lock，当然也可以用普通的锁。

Span* PageCache::MapObjectToSpan(void* obj)
{PAGE_ID id = ((PAGE_ID)obj >> PAGE_SHIFT);std::unique_lock<std::mutex> lock(_pageMtx); // 构造时加锁，析构时自动解锁auto ret = _idSpanMap.find(id);if (ret != _idSpanMap.end()){return ret->second;}else{assert(false);return nullptr;}
}

---

14. 多线程环境下对比malloc测试

之前只是对代码进行了一些基础的单元测试，下面在多线程场景下对比malloc进行测试。

// ntimes 一轮申请和释放内存的次数 || nworks 线程数 ||  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, malloc_costtime + free_costtime);
}// ntimes 一轮申请和释放内存的次数 || nworks 线程数 ||  rounds 轮次
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("%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, malloc_costtime + free_costtime);
}int main()
{size_t n = 1000;cout << "=========================================================" << endl;BenchmarkConcurrentMalloc(n, 5, 10);cout << endl << endl;BenchmarkMalloc(n, 5, 10);cout << "=========================================================" << endl;return 0;
}

在测试函数中，通过clock函数分别获取到每轮次申请和释放所花费的时间，然后将其对应累加到malloc_costtime和free_costtime上。最后就得到了，nworks个线程跑rounds轮，每轮申请和释放ntimes次，这个过程申请所消耗的时间、释放所消耗的时间、申请和释放总共消耗的时间。

创建线程时让线程执行的是lambda表达式，而这里在使用lambda表达式时，以值传递的方式捕捉了变量k，以引用传递的方式捕捉了其他父作用域中的变量，因此可以将各个线程消耗的时间累加到一起。

将所有线程申请内存消耗的时间都累加到malloc_costtime上， 将释放内存消耗的时间都累加到free_costtime上，此时malloc_costtime和free_costtime可能被多个线程同时进行累加操作的，所以存在线程安全的问题。鉴于此，在定义这两个变量时使用了atomic类模板，这时对它们的操作就是原子操作了。

先来测试一下固定大小内存的申请和释放：（如果运行失败可以先看下面的调试技巧，这里调试完换到Release测试）

运行后可以看到，malloc的效率还是更高的。

由于此时申请释放的都是固定大小的对象，每个线程申请释放时访问的都是各自thread cache的同一个桶，当thread cache的这个桶中没有对象或对象太多要归还时，也都会访问central cache的同一个桶。此时central cache中的桶锁就不起作用了，因为我们让central cache使用桶锁的目的就是为了，让多个thread cache可以同时访问central cache的不同桶，而此时每个thread cache访问的却都是central cache中的同一个桶。

下面再来测试一下不同大小内存的申请和释放：（两个地方都要改）

在Release下已经超过malloc。

Debug运行后可以看到，相比malloc来说还是差一点。

15. 调试和复杂问题的调试技巧

多线程调试比单线程调试要复杂得多，调试时各个线程之间会相互切换，并且每次调试切换的时机也是不固定的，这就使得调试过程变得非常难以控制。

本人在此项目也在调试花了很多时间，下面举几个例子来记录一下调试过程和技巧。

下面是在上面多线程测试出现一些问题（问题已被本人调试找出，现改回有问题之前）：

运行一下：（断言报错了）

自己写的断言报错还好，可以定位到报错的地方，而且这个问题是必现的。

现在根据报错找到Common.hpp的339行：

此时并不能找到上一步的原因，所以要用到条件断点来帮助调试：

1、条件断点

一般情况下我们可以直接运行程序，通过报错来查找问题。如果此时报的是断言错误，那么我们可以直接定位到报错的位置，然后将此处的断言改为与断言条件相反的if判断，在if语句里面打上一个断点，但注意空语句是无法打断点的，这时随便在if里面加上一句代码就可以打断点了。

运行到条件断点处后，我们就可以对当前程序进行进一步分析，找出断言错误的被触发原因。

2、查看函数栈帧

当程序运行到断点处时，需要对当前位置进行分析，如果检查后发现当前函数是没有问题的，这时可能需要回到调用该函数的地方进行进一步分析，此时依次点击“调试→窗口→调用堆栈”。

点击调用堆栈后：

双击函数栈帧中的其他函数，就可以跳转到该函数对应的栈帧。

需要注意的是，监视窗口只能查看当前栈帧中的变量。如果要查看此时其他函数栈帧中变量的情况，就可以通过函数栈帧跳转来查看。

此时知道有函数调用了PopFront，把鼠标移到各个变量发现_head的next和prev都等于_head，所以此时_head是空的，说明逻辑错了。

此时再跳到上一层栈帧：

此时就知道是手误了，正确逻辑应该是这样的：

上面错误的代码就相当于访问了第0个桶，第0个肯定是空的。

把条件断点换回断言再调试运行一下：

又崩了，还是跳到上几层栈帧：

此时还是找不到问题就要分析了，PopRange出问题了，大概率是Push或者PushRange有问题：

Push单个对象就是一个头插，怎么看都没问题，那应该就是PushRange有问题了：

此时再看看ThreadCache里调用PushRange的地方：

此时对比就应该想到是不是第三个参数有问题，所以在PushRange里改一下代码：

还是验证+条件断点：

	void PushRange(void* start, void* end, size_t n) // 头插一段范围的内存给_freeList{NextObj(end) = _freeList; // 赋值给end的头4/8字节_freeList = start;// 测试验证+条件断点int i = 0;void* cur = start;while (cur){cur = NextObj(cur);++i;}if (n != i){int test = 0;}_size += n; // 这里在参数增加一个个数n}

此时鼠标移到 n 和 i 或者通过自动窗口就能看到 n 和 i 并不能对应上，你说给我的是94个，但我验证了才有51个，所以还是上一层的问题，应该就是FetchRangeObj的问题，但是是同级的函数跳不到上一层栈帧，此时就把PushRange的调试代码注释掉，在FetchRangeObj再打类似的调试代码：

此时发现还是对不上，问题就在附近了，此时就找了很久就发现span有问题：

最下面的_objSize一个对象的大小是16，一页的大小是8 * 1024 = 8192，_useCount最大才512，但此时已经到533了。

所以此时就是span切错了，是GetOneSpan的问题，再往上一层打条件断点：

如果不看之前的代码能找到哪里出错吗？

此时运行结果是疑似死循环了，再看一个技巧：

3、疑似死循环时中断程序

当你在某个地方设置断点后，如果迟迟没有运行到断点处，而程序也没有崩溃，这时有可能是程序进入到某个死循环了。这时我们可以依次点击“调试→全部中断”。

这时程序就会在当前运行的地方停下来。

再按F10就发现死循环的原因是 j 出问题了。

此时就发现是少了一句代码，最后忘记把tail的next置空了：

现在终于找出了两个问题，前面都是固定申请16字节的大小，现让申请随机一点，把n调大一点：

找到断言错误的地方打个断点然后查看栈帧：

跳到上一层栈帧发现指针错了：

此时就看_idSpanMap有没有错，然后把unordered_map改成map一个个看桶里的数据。

最后就发现了这个问题（第一个问题的时候可能也看出来了）：

在NewSpan这直接返回的也要建立和ID的映射：

调试技巧就先讲到这了，调试还是在于平时的积累和动手。

---

16. 性能瓶颈分析

下面分析当前项目的瓶颈在哪里，但这不能简单的凭感觉，应该用性能分析的工具来进行分析。

VS编译器中就带有性能分析的工具的，依次点击“调试→性能查看器”进行性能分析，注意该操作要在Debug模式下进行。

将代码中n的值由10000调成了1000，否则该分析过程可能会花费较多时间，并且将malloc的测试代码进行了屏蔽，因为要分析的是我们实现的高并发内存池。

点击后再选择检测选项查看各个函数的用时时间，然后点击开始就行了。

---

17. 使用基数树进行优化

tcmalloc源码中还有很多优化，其中之一就实现了基数树进行优化。

基数树实际上就是一个分层的哈希表，根据所分层数不同可分为单层基数树、二层基数树、三层基数树等。

单层基数树实际采用的就是直接定址法，每一个页号对应span的地址就存储数组中在以该页号为下标的位置。

最坏的情况下需要建立所有页号与其span之间的映射关系，因此这个数组中元素个数应该与页号的数目相同，数组中每个位置存储的就是对应span的指针。

下面对tcmalloc源码中的单层基数树进行简单修改：

// 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) // k的范围不在[0, 2^BITS-1]{return NULL;}return array_[k]; // 返回该页号对应的span}// 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; // 建立映射}
};

此时当我们需要建立映射时就调用set函数，需要读取映射关系时，就调用get函数就行了。

代码中的非类型模板参数BITS表示存储页号最多需要比特位的个数。在32位下我们传入的是32-PAGE_SHIFT，在64位下传入的是64-PAGE_SHIFT。而其中的LENGTH成员代表的就是页号的数目，即2^BITS。

比如32位平台下，以一页大小为8K为例，此时页的数目就是2^32 / 2^13 = 2^19。因此存储页号最多需要19个比特位，此时传入非类型模板参数的值就是32 − 13 = 19。由于32位平台下指针的大小是4字节，因此该数组的大小就是2^19 * 4 = 2M，内存消耗不大，是可行的。但如果是在64位平台下，此时该数组的大小是2^51 * 8 = 2^54 = 2^24 G，这显然是不可行的，实际上对于64位的平台，我们需要使用三层基数树。

这里还是以32位平台下，一页的大小为8K为例来说明，此时存储页号最多需要19个比特位。而二层基数树实际上就是把这19个比特位分为两次进行映射。

比如用前5个比特位在基数树的第一层进行映射，映射后得到对应的第二层，然后用剩下的比特位在基数树的第二层进行映射，映射后最终得到该页号对应的span指针。

在二层基数树中，第一层的数组占用2^5 * 4 = 2^7 Byte空间，第二层的数组最多占用2^5 * 2 ^14 * 4 = 2M，二层基数树相比一层基数树的好处就是，一层基数树必须一开始就把2M的数组开辟出来，而二层基数树一开始时只需将第一层的数组开辟出来，当需要进行某一页号映射时再开辟对应的第二层的数组就行了。

// 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) // k的范围不在[0, 2^BITS-1]{return NULL;}return array_[k]; // 返回该页号对应的span}// 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; // 建立映射}
};

因此在二层基数树中有一个Ensure函数，当需要建立某一页号与其span之间的映射关系时，需要先调用该Ensure函数确保用于映射该页号的空间是开辟了的，如果没有开辟则会立即开辟。

在32位平台下，就算将二层基数树第二层的数组全部开辟出来也就消耗了2M的空间，内存消耗也不算太多，因此可以在构造二层基数树时就把第二层的数组全部开辟出来。

---

上面一层基数树和二层基数树都适用于32位平台，而对于64位的平台就需要用三层基数树了。三层基数树与二层基数树类似，三层基数树实际上就是把存储页号的若干比特位分为三次进行映射。

此时只有当要建立某一页号的映射关系时，再开辟对应的数组空间，而没有建立映射的页号就可以不用开辟其对应的数组空间，此时就能在一定程度上节省内存空间。

因此当我们要建立某一页号的映射关系时，需要先确保存储该页映射的数组空间是开辟好了的，也就是调用代码中的Ensure函数，如果对应数组空间未开辟则会立马开辟对应的空间。