【项目日记(九)】细节优化与对比测试

前言

上面我们对申请和释放的过程都已写完，并进行了单线程的联调。本期我们来对一些细节进行优化以及与malloc 进行对比测试。

目录

前言

一、大于256KB的内存申请问题

• 申请过程

• 释放过程

• 简单测试

二、使用定长内存池脱离使用new

三、优化释放对象时传递参数问题

 四、多线程环境下与malloc对比测试

---

一、大于256KB的内存申请问题

• 申请过程

        之前每个线程的thread cache都是用于申请小于等于 256KB的内存的，而线程有可能申请大于 256KB 的内存，大于256KB的我们可以直接考虑去page cache或者堆上申请。具体的：当申请内存的大小大于 256KB(32页) 小于128页时，在page cache中申请。当申请内存的大小大于128页时，去找堆申请。

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

---

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

static inline size_t RoundUp(size_t bytes)
{if (bytes <= 128){return _RoundUp(bytes, 8);}else if (bytes <= 1024){return _RoundUp(bytes, 16);}else if (bytes <= 8 * 1024){return _RoundUp(bytes, 128);}else if (bytes <= 64 * 1024){return _RoundUp(bytes, 1024);}else if (bytes <= 256 * 1024){return _RoundUp(bytes, 8 * 1024);}else{// 大于 256KB 的进行按照页进行对齐return _RoundUp(bytes, 1 << PAGE_SHIFT);}
}

        这里就还需要对之前的申请逻辑进行修改了，当申请对象的大小大于256KB时，就不用向thread cache 申请了，而是计算好对齐规则转为页数 k 去page cache 调用NewSpan申请。

static void* ConcurrentAlloc(size_t size)
{// 大于 256KB 去对齐后算出页数，去page cache申请if (size > MAX_BYTES){// 仅从向上对齐size_t alignSize = SizeClass::RoundUp(size);// 计算出对应的页数size_t kPage = alignSize >> PAGE_SHIFT;// 去page cache申请PageCache::GetInstance()->_page_mtx.lock();Span* span = PageCache::GetInstance()->NewSpan(kPage);PageCache::GetInstance()->_page_mtx.unlock();// 根据也好计算出起始地址void* ptr = (void*)(span->_page_id << PAGE_SHIFT);return ptr;}else{// 第一次调用时，通过 TLS 让每个线程获取自己专属的 thread cache 对象if (pTLSThreadCache == nullptr){pTLSThreadCache = new ThreadCache;}// 到自己的thread cache 获取内存对象return pTLSThreadCache->Allocate(size);}
}

        这里我们当256KB时都是去page cache调用NewSpan了，也是就是说不管 是不是属于 128页以内的 都去NewSpan申请，因此我们需要对NewSpan进行改造。

        当申请到页数小于等于128页，在page cache进行申请，否则去堆上申请。这里只需要处理一下，大于 128 页的情况即可。

Span* PageCache::NewSpan(size_t k)
{assert(k > 0);// 特殊处理 大于 128 页的情况if (k > NPAGES - 1)// 直接找系统{void* ptr = SystemAlloc(k);Span* span = new Span;span->_page_id = (PAGE_ID)ptr >> PAGE_SHIFT;span->_n = k;// 建立页号与span的映射_idSpanMap[span->_page_id] = span;return span;}// 1、page cache 的第 k 号 桶中存在 非空的 k页 span，直接头删拿下来返回if (!_spanlists[k].Empty()){Span* kSpan = _spanlists[k].PopFront();// 建立页号与span的映射，方便central cache回收小块内存时查找对应的spanfor (PAGE_ID i = 0; i < kSpan->_n; i++){_idSpanMap[kSpan->_page_id + i] = kSpan;}// 将分配给central cache 的 span 设置为 ture 表示 使用kSpan->_isUse = true;return kSpan;}// 2、当前的 第 k 号 桶中 不 存在 非空的 k页 span，到[k+1, NPAGES - 1] 中找for (int i = k + 1; i < NPAGES; i++){if (!_spanlists[i].Empty()){Span* nSpan = _spanlists[i].PopFront();Span* kSpan = new Span;// 将 nSpan 分割成 k 页 和 n - k页kSpan->_page_id = nSpan->_page_id;kSpan->_n = k;nSpan->_page_id += k;nSpan->_n -= k;// 将 n - k 页插入到 第 n - k 个桶中，将 k 页返回_spanlists[nSpan->_n].PushFront(nSpan);// 将n - k 页的 span 与其 首页号 与 尾页号 进行映射_idSpanMap[nSpan->_page_id] = nSpan;_idSpanMap[nSpan->_page_id + nSpan->_n - 1] = nSpan;// 这里需要减 1 例如：页号100 5页，那尾页号就是104 ==》100 + 5 - 1// 建立页号与span的映射，方便central cache回收小块内存时查找对应的spanfor (PAGE_ID i = 0; i < kSpan->_n; i++){_idSpanMap[kSpan->_page_id + i] = kSpan;}// 将分配给central cache 的 span 设置为 ture 表示 使用kSpan->_isUse = true;// 将 kSpan返回即可return kSpan;}}// 3、走到这，说明[k+1，NPAGES-1]个桶中都没有大块内存，此时就需要向 OS 申请 NAGES-1 页的大块内存了Span* bigSpan = new Span;void* ptr = SystemAlloc(NPAGES - 1);bigSpan->_page_id = (PAGE_ID)ptr >> PAGE_SHIFT;// 将起始地址除以每一页的大小，就是页号bigSpan->_n = NPAGES - 1;// 将 NPAGES-1 页插入到 Page cache 对应哈希桶的位置_spanlists[NPAGES - 1].PushFront(bigSpan);// 4、直接递归，复用上面的逻辑，分割小对象return NewSpan(k);
}

• 释放过程

当释放时，我们需要判断释放对象的大小决定还给谁：

        在释放的时候需要先找到对应的span，但是在释放对象时我们知道他们的起始地址，如何知道他们的span呢？我们可以通过页号与span的映射找到对应的span，小于等于 128 页的，我们直接就可以映射找到，而大于128页的，我们在申请时专门搞了个span并建立了映射，所以也是可以找到的。

static void ConcurrentFree(void* ptr, size_t size)
{// 大于 256 KB if (size > MAX_BYTES){Span* span = PageCache::GetInstance()->MapObjectToSpan(ptr);PageCache::GetInstance()->_page_mtx.lock();PageCache::GetInstance()->ReleaseSpanToPage(span);PageCache::GetInstance()->_page_mtx.unlock();}else{assert(pTLSThreadCache);pTLSThreadCache->Deallocate(ptr, size);}
}

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

// 将central cache 还回来的 span 进行合并 并 挂到对应的 桶中
void PageCache::ReleaseSpanToPage(Span* span)
{// 大于128 页的直接还给OSif (span->_n > NPAGES - 1){void* ptr = (void*)(span->_page_id << PAGE_SHIFT);SystemFree(ptr);delete span;return;}// 对spand的前后页进行尝试合并缓解内存外碎片化的问题// 1、向前合并while (1){// 获取前一个span 的尾页号PAGE_ID prevId = span->_page_id - 1;// 查找该页号对应的span是否在映射的哈希表中auto ret = _idSpanMap.find(prevId);// 1、如果前一个span 的尾页号不存在（还未向OS申请），直接停止合并if (ret == _idSpanMap.end()){break;}// 2、前面页号对应的span正在被使用，停止合并Span* prevSpan = ret->second;if (prevSpan->_isUse == true){break;}// 3、加上前面页合并出的span超过了 NPAGES - 1无法管理，停止合并if (span->_n + prevSpan->_n > NPAGES - 1){break;}// 进行合并span->_page_id = prevSpan->_page_id;span->_n += prevSpan->_n;// 将 prevSpan 从双链表中 移除_spanlists[prevSpan->_n].Erase(prevSpan);delete prevSpan;}// 2、向后合并while (1){// 获取后一个span 的首页号PAGE_ID nextId = span->_page_id + span->_n;// 判断 该页号 对应的 span 是否 和他进行了映射auto ret = _idSpanMap.find(nextId);// 1、如果没有建立映射（还未向OS申请），停止合并if (ret == _idSpanMap.end()){break;}// 2、如果后一个span正在被使用，停止合并Span* nextSpan = ret->second;if (nextSpan->_isUse == true){break;}// 3、如过加上后面合并出的span超出了 NPAGES - 1无法管理，停止合并if (span->_n + nextSpan->_n > NPAGES - 1){break;}// 合并span->_n += nextSpan->_n;// 将 nextSpan 从链表中移除_spanlists[nextSpan->_n].Erase(nextSpan);delete nextSpan;}// 将该合并号的 span 挂到 对应的双链表中_spanlists[span->_n].PushFront(span);// 将合并的span进行 与 首尾页号的映射_idSpanMap[span->_page_id] = span;_idSpanMap[span->_page_id + span->_n - 1] = span;// 将合并的span设置为未被使用的状态span->_isUse = false;
}

        直接向堆申请内存时我们调用的接口是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
}

• 简单测试

下面我们对大于 256 KB 的申请和释放流程再来联调一遍

void TestBigPage()
{//找page cache申请void* p1 = ConcurrentAlloc(257 * 1024); //257KBConcurrentFree(p1, 257 * 1024);//找堆申请void* p2 = ConcurrentAlloc(129 * 8 * 1024); //129页ConcurrentFree(p2, 129 * 8 * 1024);
}

        当申请257KB的内存时，由于257KB的内存按页向上对齐后是33页，并没有大于128页，因此不会直接向堆进行申请，会向page cache申请内存，但此时page cache当中实际是没有内存的，最终page cache就会向堆申请一个128页的span，将其切分成33页的span和95页的span，并将33页的span进行返回。

        而在释放内存时，由于该对象的大小大于了256KB，因此不会将其还给thread cache，而是直接调用的page cache当中的释放接口。

        由于该对象的大小是33页，不大于128页，因此page cache也不会直接将该对象还给堆，而是尝试对其进行合并，最终就会把这个33页的span和之前剩下的95页的span进行合并，最终将合并后的128页的span挂到第128号桶中。

---

         当申请129页的内存时，由于是大于256KB的，于是还是调用的page cache对应的申请接口，但此时申请的内存同时也大于128页，因此会直接向堆申请。在申请后还会建立该span与其起始页号之间的映射，便于释放时可以通过页号找到该span。

     在释放内存时，通过对象的地址找到其对应的span，从span结构中得知释放内存的大小大于128页，于是会将该内存直接还给堆 。

二、使用定长内存池脱离使用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* span = _spanPool.New();
//释放span对象
_spanPool.Delete(span);

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

---

        此外，每个线程第一次申请内存时都会创建其专属的thread cache，而这个thread cache目前也是new出来的，我们也需要对其进行替换。

// 第一次调用时，通过 TLS 让每个线程获取自己专属的 thread cache 对象
if (pTLSThreadCache == nullptr)
{// pTLSThreadCache = new ThreadCache;static std::mutex tcMtx;static ObjectPool<ThreadCache> tcPool;tcMtx.lock();pTLSThreadCache = tcPool.New();tcMtx.unlock();
}

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

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

---

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

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

        但是在这里就会有一个很尴尬的问题：我们的 SpanList 是在Commit.h 中的，而把向系统申请和释放内存的两函数放在了Common.h，ObjectPool.h 中依赖的是 Common.h，而现在SpanList想脱离new使用ObjectPool，就需要在Common.h包含 ObjectPool.h。OK，造成头文件的循环引用了！

        此时，解决的办法很多。我们这里因为 ObjectPool.h 中只是需要向申请内存的函数。基于此，我们只需要把那部分单独拿出来放到一个SystemCtrl.h 中，让ObjectPool.h包含SystemCtrl.h，再让Common.h包含ObjectPool.h，这样接解决了头文件循环引用导致的链接错误。

SystemCtrl.h

#pragma once#include <iostream>#ifdef _WIN32#include<windows.h>
#else// ... linux brk mmap...的头文件
#endif// 直接去堆上按页申请空间
inline static void* SystemAlloc(size_t kpage)
{
#ifdef _WIN32void* ptr = VirtualAlloc(0, kpage << 13, MEM_COMMIT | MEM_RESERVE, PAGE_READWRITE);
#else// linux下brk mmap等
#endifif (ptr == nullptr)throw std::bad_alloc();return ptr;
}//直接将内存还给堆
inline static void SystemFree(void* ptr)
{
#ifdef _WIN32VirtualFree(ptr, 0, MEM_RELEASE);
#else//linux下sbrk unmmap等
#endif
}

然后只需要在 Common.h 中包含 ObjectPool.h 就 OK，后面page cache包含了Common.h，也就简介包含了SystemCtrl.h.

---

三、优化释放对象时传递参数问题

        当我们使用malloc函数申请内存时，需要指定申请内存的大小；而当我们使用free函数释放内存时，只需要传递是释放内存的起始地址即可。而我们目前的内存是在释放时依然需要传递释放内存对象的大小的，这里的原因如下：

        我们需要传递对象的判断他是要还给系统还是内存池。

        1、如果size大于256KB则需要size判断具体是还给page cache还是还给OS

        2、如果size小于256KB则需要size计算应该还给thread cache的哪一个桶

        我们也想做到在释放时只传递起始地址，不传递对象的大小，我们就需要建立对象的地址与大小的映射。而我们现在已经可以使用地址找到对象对应的span了，而每个span中的对象都是确定的，所以我们这里不用专门在整一个哈希映射了，我们只需要在span中添加一个字段_objectSize表示当前span中对象的大小。后期通过地址找到span然后查找_objectSize就可以知道释放对象的大小了。

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

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

// @ 向Page cache 申请若干页的大块内存时，加 互斥锁
PageCache::GetInstance()->_page_mtx.lock();// 申请前加锁
Span* span = PageCache::GetInstance()->NewSpan(SizeClass::NumMovePage(byte_size));
span->_objectSize = byte_size;// 当从page cache获取到k页的span时，将当前span管理的对象的大小设置为size
PageCache::GetInstance()->_page_mtx.unlock();// 申请后解锁

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

---

        此时我们进行释放时，就不用传递该对象的大小了，可以直接根据地址获取取span间接获取到size，准确来说是对齐后的size。

static void ConcurrentFree(void* ptr)
{Span* span = PageCache::GetInstance()->MapObjectToSpan(ptr);size_t size = span->_objectSize;// 大于 256 KB if (size > MAX_BYTES){PageCache::GetInstance()->_page_mtx.lock();PageCache::GetInstance()->ReleaseSpanToPage(span);PageCache::GetInstance()->_page_mtx.unlock();}else{assert(pTLSThreadCache);pTLSThreadCache->Deallocate(ptr, size);}
}

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

        我们将页号与span之间的映射关系是存储在PageCache类当中的，在page cache进行相关操作，访问这个映射关系是安全的，因为当进入page cache之前是需要加锁的，因此可以保证此时只有一个线程在进行访问。

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

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

// 获取从 对象到span的映射
Span* PageCache::MapObjectToSpan(void* obj)
{// 地址 转为 页号PAGE_ID id = (PAGE_ID)obj >> PAGE_SHIFT;// 采用RAII风格的锁std::unique_lock<std::mutex> lock(_page_mtx);auto ret = _idSpanMap.find(id);if (ret != _idSpanMap.end()){return ret->second;}else{// 理论上不可能走到这里我们直接断言 错误assert(false);return nullptr;}
}

 四、多线程环境下与malloc对比测试

        之前我们只是在单线程的情况下进行了一些基础的单元测试，下面我们在多线程的情况下进行与malloc对比测试。我们将这些测试代码放在 BenchMark.cpp 中：

#include "ConcurrentAlloc.h"using std::cout;
using std::endl;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([&]() {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, 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;
}

代码解释：

• ntimes：单轮次申请和释放内存次数。

• nworks：线程数。

• rounds：轮次。

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

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

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

固定大小内存的申请和释放

v.push_back(malloc(16));
v.push_back(ConcurrentAlloc(16));

        此时4个线程执行10轮操作，每轮申请释放10000次，总共申请释放了40万次，运行后可以看到，malloc的效率还是更高的。

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

不同大小内存的申请和释放

        下面我们再来测试一下不同大小内存的申请和释放：

v.push_back(malloc((16 + i) % 8192 + 1));
v.push_back(ConcurrentAlloc((16 + i) % 8192 + 1));

        运行后可以看到，由于申请和释放内存的大小是不同的，此时central cache当中的桶锁就起作用了，ConcurrentAlloc的效率也有了较大增长。

---