thread cache回收内存

当从 thread cache 中申请的内存对象使用完毕需要还回来的时候, 只需要计算出该内存对象对应 thread cache 中的哪一个自由链表桶, 然后将该内存对象插入进去即可.

不过需要注意的是, 如果不断有内存对象释放回来, 那么可能就会导致 thread cache 当中的某一个或一些自由链表桶变得越来越长, 但是当前的 thread cache 暂时不会使用这些内存对象, 从而造成浪费.

所以我们规定, 当释放内存对象回 thread 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 cache的spanif (_freeLists[index].Size() >= _freeLists[index].MaxSize()){ListTooLong(_freeLists[index], size);}
}

当自由链表的长度大于一次批量申请内存对象的个数时, 我们需要从该自由链表中取出一段长度为一次批量申请个数的内存对象, 然后将其还给 central cache 当中对应的span.

// 释放对象链表过长时, 归还一段list给central cache的span
void ThreadCache::ListTooLong(FreeList& list, size_t size)
{void* start = nullptr;void* end = nullptr;// 从自由链表中取出一次批量申请个数的内存对象list.PopRange(start, end, list.MaxSize());// 将取出的一段内存对象还给central cache相应的spanCentralCache::GetInstance()->ReleaseListToSpans(start, size);
}

观察上面的代码我们会发现需要对自由链表FreeList类的定义进行补充, 需要补充头删一段范围的成员函数PopRange, 以及一个成员变量_size用于统计自由链表的长度, 还有与它相关的成员函数Size, 当然了, 在对自由链表执行增删操作的时候, 需要对成员变量_size进行加减运算.

//管理切分好的小对象的自由链表
class FreeList
{
public:// 头插void Push(void* obj){assert(obj);NextObj(obj) = _freeList;_freeList = obj;_size++;}// 头删void* Pop(){assert(_freeList);void* obj = _freeList;_freeList = NextObj(_freeList);_size--;return obj;}// 插入一段范围的对象到自由链表void PushRange(void* start, void* end, size_t n){assert(start);assert(end);NextObj(end) = _freeList;_freeList = start;_size += n;}// 从自由链表中取出一段范围的对象void PopRange(void*& start, void*& end, size_t n){assert(n <= _size);start = _freeList;end = start;for (size_t i = 0; i < n - 1; i++){end = NextObj(end);}_freeList = NextObj(end); // 自由链表指向end的下一个对象NextObj(end) = nullptr; // 将取出的一段链表的尾置空_size -= n;}bool Empty(){return _freeList == nullptr;}size_t& MaxSize(){return _maxSize;}size_t Size(){return _size;}private:void* _freeList = nullptr;size_t _maxSize = 1;size_t _size = 0;
};

补充说明:

实际上 TCMalloc 在判断 thread cache 是否应该还一部分对象给 central cache 时, 还有一个判断依据, 就是还会考虑 thread cache 整体的大小, 当其整体的大小超过某一个值时, 我们就会将 thread cache 中的一部分对象还给 central cache, 这也就有效避免了某个线程的 thread cache 占用了太多内存的情况.

central cache回收内存

前面我们说了当 thread cache 某个自由链表过长时, 会还一段内存对象给 central cache 对应的span, 但到底是 central cache 中哪一个span, 这是需要计算的, 因为 central cache 中有很多个SpanList双向链表结构, 每个双向链表里面又有很多个span, 所以不同的内存对象可能对应不同的span.

那么我们如何找到一个内存对象对应的span呢?

我们知道, 我们可以通过内存对象的地址计算出其所在的页号(内存对象的地址直接除以页的大小即可得出其所在的页号), 但是一个span可能有多个页, 所以我们只需要根据页号计算出其对应的span即可得出正确结果.

有了上面的推论, 我们可以建立页号和span的映射, 因为在 page cache 尝试做前后页的span的合并的时候也会用到这个映射关系, 所以将这个映射关系定义在 page cache 中.

// page cache本质是一个按页数映射的哈希桶
class PageCache
{
public://获取从对象到span的映射Span* MapObjectToSpan(void* obj);
private:std::unordered_map<PAGE_ID, Span*> _idSpanMap; // 建立页号到span的映射
};

建立页号到span映射关系之后, 我们需要注意, 每当 page cache 分配大块span给 central cache 时, 都需要建立页号到span的映射, 这样一来, 之后 thread cache 再归还内存对象给 central cache 时, 就可以直接找到其所对应的span了.

所以我们需要对之前 page cache 中NewSpan函数中的代码进行修改:

// 获取一个K页的span
Span* PageCache::NewSpan(size_t k)
{assert(k > 0 && k < NPAGES);// 先检查第K个桶里面有没有spanif (!_spanLists[k].Empty()){Span* kSpan = _spanLists[k].PopFront();//建立页号与span的映射，方便central cache回收小块内存时查找对应的spanfor (PAGE_ID i = 0; i < kSpan->_n; i++){_idSpanMap[kSpan->_pageId + i] = kSpan;}return kSpan;}// 如果第k个桶里没有, 再找第K+1到最后一个桶for (size_t i = k + 1; i < NPAGES; i++){if (!_spanLists[i].Empty()){Span* nSpan = _spanLists[i].PopFront();Span* kSpan = new Span;// 在nSpan的头部切一个k页下来// k页被返回// nSpan被挂回去kSpan->_pageId = nSpan->_pageId;kSpan->_n = k;nSpan->_pageId += k;nSpan->_n -= k;_spanLists[nSpan->_n].PushFront(nSpan);//建立页号与span的映射，方便central cache回收小块内存时查找对应的spanfor (size_t 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的映射
Span* PageCache::MapObjectToSpan(void* obj)
{assert(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;}
}

将 thread cache 中的一段内存对象还给 central cache, 是一个一个内存对象归还给 central cache 对应的span, 首先要计算出该内存对象对应的span, 然后将其头插到对应span的自由链表中, 然后_useCount–.

需要注意的是, 如果_useCount减到0了, 那么这个span切给 thread cache 用的内存对象全部还回来了, 此时这个span属于空闲的span, 会被 page cache 回收, 在此之前需要将该span从对应的双向链表中剥离下来, 然后将其对应的自由链表置空(因为page cache中的span不需要切成一个个内存对象), 以及双向链表的前后指针置空(因为之后要将其插入到page cache对应的双向链表中). 但是哦, span的大块内存的起始页号和页数是不能变的, 因为它们要用于表征这个span的位置及大小.

// 将一定数量的内存对象释放回span
void CentralCache::ReleaseListToSpans(void* start, size_t size)
{assert(start);size_t index = SizeClass::Index(size);_spanLists[index]._mtx.lock();while (start){void* next = NextObj(start);// 先计算出内存对象到span的映射Span* span = PageCache::GetInstance()->MapObjectToSpan(start);NextObj(start) = span->_freeList;span->_freeList = start;span->_usecount--;// _useCount减到0说明span分配给thread cache的内存对象都回来了// 释放空闲的span, 并合并前后页spanif (span->_usecount == 0){_spanLists[index].Erase(span);span->_freeList = nullptr;span->_prev = nullptr;span->_next = nullptr;// 先解除桶锁, 使用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();
}

最后就是关于加锁解锁的问题了, 在 central cache 还span给 page cache 时也是存在加锁解锁的, 首先要把 central cache 中对应的桶锁给解掉, 只使用 page cache 的大锁就可以了, 这样一来方便其他执行流向该桶中申请和释放内存不被阻塞, 当归还span给 page cache 之后, 记得要重新加上桶锁, 然后再执行上述过程将还未还完的内存对象继续还给 central cache.

page cache回收内存

前面我们说到, 当_useCount减到0的时候, 也就是span分配给 thread cache 使用的内存对象全部还回来了, 这个span就可以还给 page cache 了.

但是为了缓解内存碎片, 在回收span的基础上, 还需要尝试合并前后页的span, 以达到缓解内存碎片的目的.

page cache 如何进行前后页的合并?

前后页的合并也就是向前合并和向后合并, 向前合并首先需要计算出该span前一页的页号, 如果归还回来的span的起始页号是begin, 页数是n页, 那么它的前一页的页号就是begin-1, 然后再判断前一页对应的span是否适合合并, 如果适合合并, 就合并前一页的span, 然后继续向前合并; 向后合并跟前面的操作基本一致, 但需要注意的是, 后一页的起始页号是begin+n.

还有一点需要注意, 当我们通过页号计算出其所在的span时, 这个span可能挂在 page cache 中, 也可能挂在 central cache 中正在被使用, 正在 central cache 中被使用的span是不能合并的, 我们要合并的是 page cache 空闲的span, 所以仅根据_useCount减为0就将这个span合并是存在问题的, 因为当 page cache 刚分配span到 central cache 中_useCount就是0, 它还没有分配内存对象给 thread cache 使用, 所以我们应该在span的定义中增加一个成员变量_isUse用于表征当前的span是否正在被使用.

// Span 管理一个以页为单位的大块内存
struct Span
{PAGE_ID _pageId = 0; // 大块内存起始页的页号size_t _n = 0; // 页的数量Span* _prev = nullptr; // 双向链表的结构Span* _next = nullptr;void* _freeList = nullptr; // 切好的小块内存的自由链表size_t _usecount = 0; // 切好的小块内存被分配给thread cache的计数bool _isUse = false; // span是否正在被使用
};

前面我们提到 page cache 在做前后页的合并时, 需要根据首尾页号计算出对应的span, 所以我们在挂span到 page cache中时, 只需要建立该span的首尾页号到span的映射即可, 因为向前合并只需要根据当前span前一页的页号(也是该span的尾页)即可得到对应的span, 向后合并只需要根据当前span后一页的页号(也是该span的首页)即可计算出对应的span.

所以在NewSpan成员函数中, 我们在获取k页的span后, 不但要将k页span中的页号与span建立映射关系, 还需要将n-k页span的首尾页号与其建立映射关系, 方便合并前后页的span.

// 获取一个K页的span
Span* PageCache::NewSpan(size_t k)
{assert(k > 0 && k < NPAGES);// 先检查第K个桶里面有没有spanif (!_spanLists[k].Empty()){Span* kSpan = _spanLists[k].PopFront();//建立页号与span的映射，方便central cache回收小块内存时查找对应的span// 分配给central cache使用的span, 每一页都要建立页号到span的映射for (PAGE_ID i = 0; i < kSpan->_n; i++){_idSpanMap[kSpan->_pageId + i] = kSpan;}return kSpan;}// 如果第k个桶里没有, 再找第K+1到最后一个桶for (size_t i = k + 1; i < NPAGES; i++){if (!_spanLists[i].Empty()){Span* nSpan = _spanLists[i].PopFront();Span* kSpan = new Span;// 在nSpan的头部切一个k页下来// k页被返回// nSpan被挂回去kSpan->_pageId = nSpan->_pageId;kSpan->_n = k;nSpan->_pageId += k;nSpan->_n -= k;_spanLists[nSpan->_n].PushFront(nSpan);// 挂到page cache中的span, 只需要建立其前后页到span的映射即可, 用于前后页的合并_idSpanMap[nSpan->_pageId] = nSpan;_idSpanMap[nSpan->_pageId + nSpan->_n - 1] = nSpan;for (size_t 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到page cache, 并合并相邻的span
void PageCache::ReleaseSpanToPageCache(Span* span)
{assert(span);// 尝试合并前后页span, 缓解内存碎片的问题// 向前合并while (1){PAGE_ID prevId = span->_pageId - 1; // 前一页// 1. 前一页对应的span不存在, 不合并了auto ret = _idSpanMap.find(prevId);if (ret == _idSpanMap.end()){break;}// 2. 前一页对应的span被使用了, 不合并了Span* prevSpan = ret->second;if (prevSpan->_isUse == true){break;}// 3. 合并出比128页大的span没办法管理, 不合并了if (span->_n + prevSpan->_n > NPAGES - 1){break;}// 更新span的起始页号和页数span->_pageId = prevSpan->_pageId;span->_n += prevSpan->_n;// 将前一页的span从对应的双向链表中剥离_spanLists[prevSpan->_n].Erase(prevSpan);delete prevSpan;}// 向后合并while (1){PAGE_ID nextId = span->_pageId + span->_n; // 后一页// 1. 后一页的span不存在, 不合并了auto ret = _idSpanMap.find(nextId);if (ret == _idSpanMap.end()){break;}// 2. 后一页的span被使用了, 不合并了Span* nextSpan = ret->second;if (nextSpan->_isUse == true){break;}// 3. 合并出了超过128页的span不能管理, 不合并了if (span->_n + nextSpan->_n > NPAGES - 1){break;}span->_n += nextSpan->_n;_spanLists[nextSpan->_n].Erase(nextSpan);delete nextSpan;}// 将合并后的span挂到page cache对应的双向链表中_spanLists[span->_n].PushFront(span);span->_isUse = false; // 将状态设置为未被使用// 建立span与其首尾页的映射_idSpanMap[span->_pageId] = span;_idSpanMap[span->_pageId + span->_n - 1] = span;
}