【项目日记(八)】内存回收与联调

前言

我们前面实现了三层缓存申请的过程，并完成了三层缓存申请过程的联调！本期我们来介绍三层的缓存的回收机制以及三层整体联调释放的过程。

目录

前言

一、thread cache 回收内存

二、central cache 回收内存

• 如何确定一个对象对应的span

 • central cache 的内存回收

三、page cache 回收内存

 • page cache 合并页的策略

• page cache 进行合并 span 的实现

四、内存回收过程的整体联调

---

一、thread cache 回收内存

        当某个线程申请的内存对象不用了时，就可以将其还给thread cache，然后thread cache将该对象插入到对应的哈希表桶即可！

        随着线程的不断释放，thread cache对应的哈希桶中的内存对象的自由链表会越来越长，这些内存还回来堆积在一个thread cache中也是一种浪费，其他线程的thread cache 也用不到。此时，我们可以考虑将还回来的部分内存还给下一层central cache，这样一来还回去这部分内存其他线程也就可以申请了。

        我们这里采用的释放策略是：如果thread cache 某个桶中的自由链表的长度超过他一次批量向central cache申请的个数，那么我们就取thread cache对应桶中的批量数的内存对象，还给central cache。举个例子：比如当前线程thread cache的0号桶中的对象数是10，一次批量向central cache获取的数是8，那就将0号桶中的8个内存对象拿出来还给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);// 当前桶中的自由链表的长度超过 一次向 central cache 申请的数量 我们就还回去一个批量if (_freelists[index].Size() > _freelists[index].MaxSize()){ListTooLong(_freelists[index], size);}
}

        ListTooLong 的具体做法就是从当前的自由链表中取出一批内存，然后一批内存对象还给central cache 对应哈希桶位置的 span 对象，即插入到span 的自由链表中即可！

// 还给 central cache 一个批量的内存对象
void ThreadCache::ListTooLong(FreeList& list, size_t size)
{// 记录需要被 拿下来 还回去 的那一批内存的 起始和结束位置void* start = nullptr, *end = nullptr;// 从自由链表list 中获取list.PopRange(start, end, list.MaxSize());// 将取出来的这一批内存还给 central cache 对应的spanCentralCache::GetInstance()->ReleaseListToSpan(start, end);
}

        上面的代码中，我们需要在FreeList 中添加一个字段_size记录当前自由链表的长度，还需要新增一个接口 PopRange用于从自由链表中取出一定数量的内存对象。

// 用于管理切分好的小内存对象的自由链表
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){// 头插NextObj(end) = _freelist;_freelist = start;// 更新_size_size += n;}void PopRange(void*& start, void*& end, size_t n)// 这里使用引用，避免使用二级指针{assert(start);assert(end);// 头删start = _freelist;end = start;for (size_t i = 0; i < n - 1; i++)// 因为end 已经指向第一个了，所以走n-1步{end = NextObj(end);}// 将_freelist指向end 的下一个_freelist = NextObj(end);// 让 end 的下一个置空NextObj(end) = nullptr;// 更新_size_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; // 默认从 central cache 获取 内存对象的个数size_t _size = 0;// 用于记录当前自由链表的长度
};

        这里需要注意的是，我为了在FreeList中更好地更新_size我将PushRange的接口进行了修改，多加了一个参数n，表示要插入这一段的内存的数量，这个值在外面很好计算，但是在里面就需要遍历了，所以这里选择外部给。

        外部在使用PushRange的地方修改传参，即新增插入对象的个数。外部如何计算呢？其实很好计算！thread cache从central cache获取到一定批量的内存后，如果是一个直接返回不会调用PushRange的，所以在申请多个的时候才调用，而申请到多个内存时总数是 actualNum 个，第一个返回， actualNum - 1个插入到thread cache 的自由链表中，所以插入的个数就是actualNum-1个。

注意一下：

        我们判断thread cache给central cache归还内存时的条件是：当thread cache 的某个哈希桶中的自由链表太长时，需要归还一定的批量。其实我们还可考虑的更加全面一些：例如，加上每个对象的大小。当某个thread cache的总占用内存超过一定限制时，我们就应该将thread cache中的一部分对象还给central cache，这样就可以做到尽可能让每个thread cache占用的内存最小，这一点tcmalloc人家就是结合上述的两点设计的。

二、central cache 回收内存

        当 thread cache 中某个自由链表的长度过长时，会将该自由链表的一部分对象拿下来还给下一层 central cache 对应的span。

        但是这里有个问题是，还回来的这一批内存对象有可能不是属于一个span的。因此，我们除了确定他们是在哪一个哈希桶中之外，还需要确定每一个内存对象是属于哪一个span 的。

        具体做法是：我们可以根据对象的地址先计算出对应的页号，然后用页号找到对应的span。

• 如何确定一个对象对应的span

1、如何根据对象的地址找到对应的页号？

        可以通过当前对象的地址 除以 一页的大小就是对应的 页号。因为在某一页中所有地址除以该页的大小都等于该页的页号。举个例子：假设一页大小是 2000，那么地址就是[0,1999] 它们都除以 2000 都是等于 0 的，即页号就是0

2、如何通过页号找到对应的span？

       上面已经知道对象的页号了，但是依然无法确定他是属于哪一个span 的因为一个span有可能管理了多个页。

        基于上面，我们可以在page cache 向 central cache 分配若干页的span时，构建每一页的页号与span的映射。因为这个映射关系在page cache合并也需要，因此我们直接定义在 page cache层，这里我么直接使用 unordered_map进行做映射，并需要提供一个接口，用于让central cache获取映射关系。

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

        MapObjectToSpan的实现也是很简单，就是将对象的地址是转为页号，然后根据页号，在哈希表中找到对应的span

// 获取从 对象到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;}
}

        每当page cache分配给central cache时，都需要记录一下每一页的页号与span的映射关系，后面当thread cache还对象给central cache时，才可以知道是具体还给哪一个span。

        因此当central cache在调用NewSpan接口向page cache申请内存时，page cache在返回这个k页span之前，就应该将这k页与对应的span进行建立映射。

Span* PageCache::NewSpan(size_t k)
{assert(k > 0 && k < NPAGES);// 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;}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);// 建立页号与span的映射，方便central cache回收小块内存时查找对应的spanfor (PAGE_ID i = 0; i < kSpan->_n; i++){_idSpanMap[kSpan->_page_id + i] = kSpan;}// 将 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);
}

 • central cache 的内存回收

        当thread cache还一批内存对象给central cache时，就可以依次遍历这批对象，将这些对象插入到自己对象的span的自由链表中，并及时更新_useCount的计数。

        在这个归还的过程中，如果发现central cache的某个span的_useCount减到了 0 就说明该span分出去的所有对象都还回来，此时就是需要向下一步还该span 了。

// 将thread cache还回来的对象 挂到合适的 span 中
void CentralCache::ReleaseListToSpans(void* start, size_t size)
{// 根据size 找到对应的桶size_t index = SizeClass::Index(size);// 加桶锁_spanlists[index]._mtx.lock();// 将对象一个个的挂到合适的spanwhile (start){void* next = NextObj(start);// 找到当前对象对应的spanSpan* span = PageCache::GetInstance()->MapObjectToSpan(start);// 将对象头插到spanNextObj(start) = span->_freeList;span->_freeList = start;// 更新计数span->_useCount--;// 当前span 的所有对象都还回来了，此时就需要将span还给pageif (span->_useCount == 0){// 将当前的span从链表中 移除_spanlists[index].Erase(span);span->_freeList = nullptr;// 自由链表置空span->_next = nullptr;span->_prev = nullptr;// 将 span 还给page cache// 因为当前已经将 span 从链表中拿下来了，且要还给page cache了，为了不影响其他线程，我们当前可以把锁解了_spanlists[index]._mtx.unlock();// 给page cache 加锁PageCache::GetInstance()->_page_mtx.lock();PageCache::GetInstance()->ReleaseSpanToPage(span);PageCache::GetInstance()->_page_mtx.unlock();// 当前的span 还给span 后继续加锁，处理下一个对象_spanlists[index]._mtx.lock();}start = next;}// 解桶锁_spanlists[index]._mtx.unlock();
}

         需要注意的是，如果当把某个span还给page cache，我们需要先将这个span从cache对应的双链表中移除，然后再将span的自由链表置空、前驱后继指针置空。但是不敢将span的起始页号以及页数清理，因为他要向page cache的哈希桶中插入，如果清空就找不到了。

        这里你可能在想，这里每个span虽然内存都还回来了但是都是乱序的，这在page cach合并中没有影响嘛？没有的！因为span的起始地址（页号）知道，几页也是知道的。所以不用管内部是否乱序！

        在将span 还给page cache 时需要注意锁的问题，当将span从链表中移除下来之后，应该将该位置的桶锁给解除，因为此时有可能其他线程也需要在该位置申请或者释放。然后加上page cache 的大锁，归还span，然后解除page cache的大锁。然后在进行下对一个对象进行挂到合适的span，因为在解除page cache的大锁后，要立即申请 桶锁。

---

三、page cache 回收内存

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

        这个思路一看很简单，只需要将span挂到page cache对应的桶中就好了，实际不然。为了缓解内存外碎片的问题，page cache还需要尝试将还回来的span与其他空闲的span进行合并成更大页的span。

 • page cache 合并页的策略

        page cache 的合并策略是：将当前还回来span 对应前后页的 span 尽可能的合并成更大页的span。如果当前还回来的 span 的页号是 id 该 span 管理的页数是 n。向前合并时，就需要判断 id-1 页的span是否空闲，如果空闲就将其和当前的span合并，合并完成之后，继续向前合并直到不能合并为止。向后合并时，需要判断 id + n 页的 span 是否空闲，如果空闲就合并，合并完成之后，继续向后合并，直到不能合并为止。

这里不能合并的条件是啥呢？

不能合并的条件有三个：

1、前面/后面的页号在映射的哈希表中没有（还没有向OS申请）

2、前面/后面的页号正在被使用

3、当前合并的页的大小，超过NPAGES-1以至于无法在page cache管理（也就是合并的过程中大块的内存是需要 <= NPAGES-1的）

        因为上述page cache在进行合并相邻页的span时，需要通过根据页号获取相对应的span，所以这就是我们当时把页号与span之间的映射关系定义到page cache 层的原因。

        这里虽然可以使用页号找到对应的span了，但是如何保证当前的 span 在page cache层呢？如果当前的span是在central cache层的那是不能够进行合并的，因为central cache的span有可能正在被线程使用了。

        这里你可能想到说，可以是使用_useCount是否等于0来确定他在那一层的。这种想法是不对的，有一种情况：当central cache刚向page cache 申请了一个span此时他还没有来得及给上层thread cache了，即_useCount 是 0，此时如果page cache将其回收回去合并了，这显然是不合理的。

        因此基于上面的原因，我们可以在Span中添加一个布尔类型的字段 _isUse 标记当前的span是否在被使用，page cache 层的就是 false表示未被使用，central cache就是true表示在使用。当central cache找page cache申请成功时，将 _isUse 置为 true，等central cache将span还给page cache后将 _isUse 置为 false。

// 管理多个连续页大块内存的跨度结构
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 是否被使用 ，默认是未使用的
};

        由于在合并page cache的span的过程中，需要它通过页号找到其对应的span，但是一个span在被分配给central cache时，才会建立各个页与span之间的映射关系，而剩余挂在page cache 的那个span是没有进行页号和span的映射关系的建立的，因此在page cache之间的span也需要和其页号进行映射关系的建立。

        与central cache中span不同的是，在page cache中，只需要建立一个span的首尾页号与span的映射关系即可。因为当一个span在进行尝试合并时，如果是向前合并，只需要通过一个span的尾页号即可找到该span，同理如果是向后合并，只需要通过一个span的首页号即可找到该span。因此，在page cache中的映射只需要span与该span管理的 起始页号 和 尾页号即可。

        所以，当我们申请k页的span时，如果是将n页的span切成了k页 和 n - k 页的span，我们除了需要将k页span的每一页与kspan进行映射外，还需要将 n- k页的n-kspan进行首尾页号与n-kspan进行映射。

Span* PageCache::NewSpan(size_t k)
{assert(k > 0 && k < NPAGES);// 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);
}

• page cache 进行合并 span 的实现

        此时page cache中的span都与他的首页号和尾页号都建立了映射，我们就可以根据上述的合并策略进行合并了。

// 将central cache 还回来的 span 进行合并 并 挂到对应的 桶中
void PageCache::ReleaseSpanToPage(Span* span)
{// 对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;
}

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

---

四、内存回收过程的整体联调