【高并发内存池】第六弹---深入理解内存管理机制：ThreadCache、CentralCache与PageCache的回收奥秘

✨个人主页：熬夜学编程的小林

💗系列专栏：【C语言详解】【数据结构详解】【C++详解】【Linux系统编程】【Linux网络编程】【项目详解】

1、threadcache回收内存

2、centralcache回收内存

3、pagecache回收内存

1、threadcache回收内存

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

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

3、如果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 cache
	if (_freeLists[index].Size() >= _freeLists[index].MaxSize())
	{
		ListTooLong(_freeLists[index], size);
	}
}

当自由链表的长度大于一次批量申请的对象时，我们具体的做法就是，从该自由链表中取出一次批量个数的对象，然后将取出的这些对象还给central cache中对应的span即可。

// 释放对象时，链表过长时，回收内存回到中心缓存
void ThreadCache::ListTooLong(FreeList& list, size_t size)
{
	void* start = nullptr;
	void* end = nullptr;
	// 从list取出批量个对象
	list.PopRange(start, end, list.MaxSize());

	// 将取出的对象还给central cache对应的span
	CentralCache::GetInstance()->ReleaseListToSpans(start, size);
}

从上述代码可以看出，FreeList类需要支持用Size函数获取自由链表中对象的个数，还需要支持用PopRange函数从自由链表中取出指定个数的对象。因此我们需要给FreeList类增加一个对应的PopRange函数，然后再增加一个_size成员变量，该成员变量用于记录当前自由链表中对象的个数，当我们向自由链表插入或删除对象时，都应该更新_size的值。

// 管理切分好的小块对象的自由链表
class FreeList
{
public:
	// 插入，头插
	void Push(void* obj)
	{
		assert(obj);

		//*(void**)obj = _freeList;
		Next(obj) = _freeList; 
		_freeList = obj;

		++_size;
	}
	// 删除，头删
	void* Pop()
	{
		assert(_freeList);

		void* obj = _freeList;
		_freeList = Next(obj);
		--_size;

		return obj;
	}
	// 头插一段范围的对象到自由链表
	void PushRange(void* start, void* end, size_t n)
	{
		assert(start);
		assert(end);
		// 链接到自由链表头结点
		Next(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 = Next(end);
		}
		_freeList = Next(end); // 自由链表指向end的下一个对象
		Next(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; // 链表结点个数
};

PopRange

对于FreeList类当中的PushRange成员函数，我们最好也像PopRange一样给它增加一个参数，表示插入对象的个数，不然我们这时还需要通过遍历统计插入对象的个数。

因此之前在调用PushRange的地方就需要修改一下，而我们实际就在一个地方调用过PushRange函数，并且此时插入对象的个数也是很容易知道的。当时thread cache从central cache获取了actualNum个对象，将其中的一个返回给了申请对象的线程，剩下的actualNum-1个挂到了thread cache对应的桶当中，所以这里插入对象的个数就是actualNum-1。

FetchFromCentralCache()函数修改：

// 从中心缓存获取对象
void* ThreadCache::FetchFromCentralCache(size_t index, size_t size)
{
	size_t batchNum = min(_freeLists[index].MaxSize(), SizeClass::NumMoveSize(size));
	if (_freeLists[index].MaxSize() == batchNum)
	{
		_freeLists[index].MaxSize() += 1;
	}
	void* start = nullptr;
	void* end = nullptr;
	size_t actualNum = CentralCache::GetInstance()->FetchRangeObj(start, end, batchNum, size);
	assert(actualNum >= 1); 
	if (actualNum == 1)
	{
		assert(start == end);
		return start;
	}
	else
	{
		_freeLists[index].PushRange(Next(start),end, actualNum - 1); // 修改部分
		return start;
	}
}

说明一下：

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

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

2、centralcache回收内存

当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类当中添加一个映射关系了，这里可以用C++当中的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之间的映射关系。

// 获取一个K页的span(加映射版本)
Span* PageCache::NewSpan(size_t k)
{
	assert(k > 0 && k < NPAGES); // 保证在桶的范围内

	// 1.检查第k个桶里面有没有Span，有则返回
	if (!_spanLists[k].Empty())
	{
		Span* kSpan = _spanLists[k].PopFront();

		// 建立页号与span的映射，方便central cache回收小块内存查找对应的span
		for (PAGE_ID i = 0; i < kSpan->_n; i++)
		{
			_idSpanMap[kSpan->_pageID + i] = kSpan;
		}
		return kSpan;
	}
	// 2.检查一下后面的桶里面有没有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页下来
			kSpan->_pageID = nSpan->_pageID;
			kSpan->_n = k;

			// 更新nSpan位置
			nSpan->_pageID += k;
			nSpan->_n -= k;

			// 将剩下的挂到对应映射的位置
			_spanLists[nSpan->_n].PushFront(nSpan);

			for (PAGE_ID i = 0; i < kSpan->_n; i++)
			{
				_idSpanMap[kSpan->_pageID + i] = kSpan;
			}

			return kSpan;
		}
	}
	// 3.没有大页的span，找堆申请128页的span
	Span* bigSpan = new Span;
	void* ptr = SystemAlloc(NPAGES - 1); // 申请128页内存
	bigSpan->_pageID = (PAGE_ID)ptr >> PAGE_SHIFT;
	bigSpan->_n = NPAGES - 1; // 页数

	// 将128页span挂接到128号桶上
	_spanLists[bigSpan->_n].PushFront(bigSpan);

	// 递归调用自己(避免代码重复)
	return NewSpan(k);
}

此时我们就可以通过对象的地址找到该对象对应的span了，直接将该对象的地址除以页的大小得到页号，然后在unordered_map当中找到其对应的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;
	}
}

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

central cache回收内存

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

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

// 将一定数量的对象释放到span
void CentralCache::ReleaseListToSpans(void* start, size_t size)
{
	size_t index = SizeClass::Index(size); // 计算桶的下标
	_spanLists[index]._mtx.lock(); // 加桶锁
	while (start)
	{
		void* next = Next(start); // 记录下一个结点
		// 获取从对象到span的映射
		Span* span = PageCache::GetInstance()->MapObjectToSpan(start);
		// 将对象头插到span自由链表
		Next(start) = span->_freeList;
		span->_freeList = start;

		span->_useCount--; // 更新被分配给thread cache的计数
		// 说明这个span分配出去的对象全部都还回来了
		if (span->_useCount == 0)
		{
			// 此时这个span就可以再回收给page cache，page cache可以再尝试去做前后页合并
			_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(); // 解除桶锁
}

引用计数不为0

注意：

如果要把某个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。

3、pagecache回收内存

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

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

page cache进行前后页的合并分析

合并的过程可以分为向前合并和向后合并。如果还回来的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时(CentralCache::GetOneSpan函数)，需要立即将该span的_isUse改为true。

// 获取一个非空的span
Span* CentralCache::GetOneSpan(SpanList& list, size_t size)
{
    //...
	Span* span = PageCache::GetInstance()->NewSpan(SizeClass::NumMovePage(size));
	span->_isUse = true; // 设置为true
	PageCache::GetInstance()->_pageMtx.unlock();
    //...
}

当central cache将某个span还给page cache时(PageCache::ReleaseSpanToPageCache函数)，也就需要将该span的_isUse改成false。

// 释放空闲的span回到PageCache，并合并相邻的span
void PageCache::ReleaseSpanToPageCache(Span* span)
{
    //...
    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与其首尾页之间的映射关系。

// 获取一个K页的span(加映射版本)
Span* PageCache::NewSpan(size_t k)
{
	assert(k > 0 && k < NPAGES); // 保证在桶的范围内

	// 1.检查第k个桶里面有没有Span，有则返回
	if (!_spanLists[k].Empty())
	{
		Span* kSpan = _spanLists[k].PopFront();

		// 建立页号与span的映射，方便central cache回收小块内存查找对应的span
		for (PAGE_ID i = 0; i < kSpan->_n; i++)
		{
			_idSpanMap[kSpan->_pageID + i] = kSpan;
		}
		return kSpan;
	}
	// 2.检查一下后面的桶里面有没有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页下来
			kSpan->_pageID = nSpan->_pageID;
			kSpan->_n = k;

			// 更新nSpan位置
			nSpan->_pageID += k;
			nSpan->_n -= k;

			// 将剩下的挂到对应映射的位置
			_spanLists[nSpan->_n].PushFront(nSpan);

			for (PAGE_ID i = 0; i < kSpan->_n; i++)
			{
				_idSpanMap[kSpan->_pageID + i] = kSpan;
			}

			return kSpan;
		}
	}
	// 3.没有大页的span，找堆申请128页的span
	Span* bigSpan = new Span;
	void* ptr = SystemAlloc(NPAGES - 1); // 申请128页内存
	bigSpan->_pageID = (PAGE_ID)ptr >> PAGE_SHIFT;
	bigSpan->_n = NPAGES - 1; // 页数

	// 将128页span挂接到128号桶上
	_spanLists[bigSpan->_n].PushFront(bigSpan);

	// 递归调用自己(避免代码重复)
	return NewSpan(k);
}