当前位置：首页 > news >正文

高并发内存池（三）：PageCache（页缓存）的实现

news 2025/4/16 0:03:18

前言：

        在前两期内容中，我们深入探讨了内存管理机制中在 ThreadCache 和 CentralCache两个层级进行内存申请的具体实现。这两层缓存作为高效的内存分配策略，能够快速响应线程的内存需求，减少锁竞争，提升程序性能。

        本期文章将继续沿着这一脉络，聚焦于PageCache层级的内存申请逻辑。作为内存分配体系的更高层级，PageCache承担着从操作系统获取大块内存，并对其进行管理和再分配的重要职责。通过对PageCache内存申请流程的剖析，我们将完整地勾勒出整个内存分配体系的工作流程。这不仅有助于我们理解内存分配的底层机制，也为后续深入探讨内存释放策略奠定了基础。

        接下来，让我们一同走进PageCache的内存世界。

一、PageCache的概述

二、PageCache结构

三、内存申请的核心流程

四、PageCache类的设计

五、代码实现

1.GetOneSpan的实现

2.NewSpan的实现

3.加锁保护

六、源码

一、PageCache的概述

⻚缓存是在central cache缓存上⾯的⼀层缓存，存储的内存是以⻚为单位存储及分配的。当central cache没有内存对象时，从page cache分配出⼀定数量的page，并切割成定⻓⼤⼩的⼩块内存，分配给central cache。当⼀个span的⼏个跨度⻚的对象都回收以后，page cache会回收central cache满⾜条件的span对象，并且合并相邻的⻚，组成更⼤的⻚，缓解内存碎⽚的问题。

二、PageCache结构

PageCache的结构是一个哈希表，如上图所示，也就是一个储存span双链表的数组，这一点和CentralCache的结构完全类似，区别在于这里的Span是大块的页（1页~128页）没有进行分割，而CentralCache中的Span是分割好的小块内存（自由链表），这样做的好处是方便页与页之间的分割与合并，有效的缓解内存外碎片，在下文会详细讲解。

PageCache结构的哈希映射方法是直接定址法，下标为n的位置储存的Span双链表中每个节点有n个页。

这个哈希表也是所有线程共用一个，属于临界资源需要加锁，但不是桶锁，而是一整个大锁，为什么呢？同样在下文细讲。

注：通常以4KB或8KB为一页，这里我们以8KB为一页。

三、内存申请的核心流程

ThreadCache $\rightarrow$ CentralCache $\rightarrow$ PageCache $\rightarrow$ 系统

当ThreadCache的自由链表内没有内存时，会向CentralCache中的一个Span申请，如果没有可用的Span，就向PageCache申请一个Span大块页，如果没有Span大块页了，最后才会向系统申请。

那么CentralCache具体是如何向PageCache申请内存的呢？

同样的PageCache一次只给一个Span，然后把这个Span切割成小块连接到CentralCache的哈希桶中，要给几页的Span呢？，这需要根据要将它切割成多大为单位的小块内存来进行具体分配，如果要切较小的内存块，页数就给小一些，如果要切大内存块，页数就给多一些。这里我们就先记为n页的Span。

因为PageCache中的哈希映射是直接定址法，所以直接找到n下标位置的Span链，如果有Span则取出并切成小块连接到CentralCache的哈希桶中。

如果没有n页的Span并不是直接去系统申请，而是到更大的页单位去找Span。比如找到有(n+k)页的Span，k>0，n+k<=128，那么把(n+k)拆开为n页与k页，并连接到相应位置。这样就有了n页的Span，把它切割和连接就行。

如果直到找到128页也没有找到可用的Span那么就向系统申请128页的内存，即128*8*1024字节，然后再执行上面逻辑。这就是整个在PageCache申请内存的逻辑。

处理临界资源互斥问题

如上，PageCache的哈希桶中各个桶是互相联动的，主要体现在这三方面：

当前桶没Span要往后找。
后面的桶切割Span后要连接到前面的桶。
在内存回收时又需要前面桶的Span合并成大页的Span连接到后面的桶。

所以这里不像CentralCache结构的桶相互独立，如果用桶锁会造成频繁的锁申请和释放，效率反而变得很低。用一个大锁来管理整个哈希桶更为合理。

四、PageCache类的设计

把这个类的设计放在头文件PageCache.h里，它核心就两个成员变量：锁和哈希桶，然后再声明一个用来申请Span的成员函数，如下：

class PageCache
{
public:Span* NewSpan(size_t k);std::mutex _pageMtx;
private:SpanList _spanLists[NPAGES];
};

Span* NewSpan(size_t k)：申请一个k页的Span
mutex _pageMtx：锁，因为它要在外部使用，所以设为public。
SpanList _spanLists[NPAGES]：哈希桶，其中NPAGES是一个静态全局变量为129，在Common.h中定义。设为129是因为哈希表中存1页~128页的Span，要通过页数直接定址的方式找到Span链表，就需要开辟129大小的数组。

PageCache类和CentralCache类一样所有线程共用一份，所以把它创建为单例模式，它分为以下几步：

把构造函数设为私有。
把拷贝构造禁用。
声明静态的PageCache对象，并在PageCache.cpp中定义。
提供一个获取PageCache对象的静态成员函数，并设为public。

代码示例：

class PageCache
{
public:static PageCache* GetInstance(){return &_sInst;}Span* NewSpan(size_t k);std::mutex _pageMtx;
private:PageCache() {}PageCache(const PageCache&) = delete;SpanList _spanLists[NPAGES];static PageCache _sInst;
};

五、代码实现

1.GetOneSpan的实现

回顾上一期我们只是假设通过GetOneSpan从CentralCache中取到了Span，然后继续做后面的处理。接下来我们一起来实现GetOneSpan函数。

因为要在Span链表中取一个有用的Span节点，所以需要遍历Span链表，那么可以模拟一个迭代器。我们在SpanList类中封装这两个函数：

Span* Begin()
{return _head->_next;
}
Span* End()
{return _head;
}

注：SpanList是一个带头双向环形链表。

接下来遍历链表找到可用的Span并返回，如果没有则到PageCache中申请。

Span* CentralCache::GetOneSpan(SpanList& list, size_t size)
{Span* it = list.Begin();while (it != list.End()){if (it->_freeList != nullptr){return it;}it = it->_next;}//走到这里说明没有可用的Span了，向PageCache中申请。//......
}

SpanList& list：指定的一个哈希桶，即一个Span链表的头结点。
size_t size：进行内存对齐后实际需要申请的字节大小。

注意在PageCache中申请的是大块的Span页，还需要把它切割，然后连接到CentralCache的桶中并返回。

在PageCache类中我们准备实现的函数NewSpan就是用来实现这个功能，现在需要考虑的是要传入的参数是多少，即要申请多少页的Span。

这里我们是以8KB为一页，即 1页=8*1024字节（2^13字节）为方便后面做位运算，我们在Common.h文件定义一个这样一个变量：

static int const PAGE_SHIFT = 13;

当然了这里也可以做成宏定义。

接下来在SizeClass类里封装一个函数NumMovePage用来计算一个size需要申请几页的Span，如下：

//用来计算ThreadCache向CentralCache申请几个小内存块
static inline size_t NumMoveSize(size_t size)
{int ret = MAX_BYTES / size;if (ret < 2) ret = 2;if (ret > 512) ret = 512;return ret;
}
//用来计算CentralCache向PageCache申请几个页的Span
static inline size_t NumMovePage(size_t size)
{int num = NumMoveSize(size);int npage = num * size;npage >>= PAGE_SHIFT; //除以8KBif (npage < 1) npage = 1;return npage;
}

在此，大家无需过度纠结于该设计背后的原理。这或许是相关领域的资深专家在历经大量测试与实践后，所总结提炼出的有效策略，我们继续走下面的逻辑。

//申请Span
Span* span = PageCache::GetInstance()->NewSpan(SizeClass::NumMovePage(size));
//切割Span
//......

NewSpan函数我们待会再设计，现在假设已经申请到一个Span的大块页，接下来就是进行切割。

首先我们需要得到的是这个页的起始地址与终止地址，在这个范围内以size的跨度切割。

用char*的指针变量start，end分别来储存起始地址和终止地址，用char*类型是因为char*类型指针变量加多少，就移动多少字节，方便后面做切割运算。

起始页地址的获取：

取到Span的页号，用页号乘以8KB（2^13字节）就能得到页的起始地址，即：

char* start = (char*)(span->_pageId << PAGE_SHIFT);

终止页地址的获取：

取到Span的页数，用页数乘以8KB再加上起始页地址就能得到终止页地址，即：

int bytes = span->_n << PAGE_SHIFT;
char* end = start + bytes;

注：页号是通过申请到的内存的虚拟地址除以8KB得到的。

接下来把start连接到Span的自由链表中，然后使用循环把[start，end]这块内存以size为跨度进行切割，如下：

Span* CentralCache::GetOneSpan(SpanList& list, size_t size)
{Span* it = list.Begin();while (it != list.End()){if (it->_freeList != nullptr){return it;}it = it->_next;}//申请SpanSpan* span = PageCache::GetInstance()->NewSpan(SizeClass::NumMovePage(size));//切割Spanchar* start = (char*)(span->_pageId << PAGE_SHIFT);int bytes = span->_n << PAGE_SHIFT;char* end = start + bytes;span->_freeList = start;while (start < end){Nextobj(start) = start + size;start = (char*)Nextobj(start);}Nextobj(end) = nullptr;//连接到CentralCache中list.PushFront(span);return span;
}

这里直接以尾插的方式切，好处在于用户在使用内存时是连在一块的，相关的数据会被一并载入寄存器中，可以增加高速缓存命中率，提高效率。

最后实现一个类方法PushFront，用来把Span连接到哈希桶里。

void PushFront(Span* node)
{//在Begin()前插入node，即头插Insert(Begin(), node);
}

2.NewSpan的实现

NewSpan我们放在源文件PageCache.cpp中实现。

查找当前桶

首先可以用assert断言页数的有效性，然后直接定址找到Span链，如果不为空返回一个Span节点，如果为空到后面的大页去找，如下：

Span* PageCache::NewSpan(size_t k)
{//判断页数的有效性assert(k > 0 && k < NPAGES);if (!_spanLists[k].Empty()){return _spanLists[k].PopFront();}//到后面更大块的Span页切割//......//向系统申请内存//......
}

到大页切割

当k号桶没有Span后，从k+1号桶开始往后找，比如找到i号桶不为空，则把Span节点取出来，记为nSpan，然后切割为k页和i-k页，切割的本质就是改变页数_n和页号_pageId。

new一个Span空间，用kSpan变量指向，然后把nSpan头k个页切出来储存到一个kSpan中，即：

kSpan->_pageId = nSpan->_pageId：更新kSpan的页号。
kSpan->_n = k：更新kSpan的页数。
nSpan->_pageId += k：原nSpan的页起始页号后移k位。
nSpan->_n -= k：原nSpan的页数减少k。

这样就相当于把原来大块的nSpan切割成了两块，最后把割剩下的nSpan插入到对应的哈希桶中，把kSpan返回。

	for (int i = k + 1; i < NPAGES; i++){if (!_spanLists[i].Empty()){Span* nSpan = _spanLists[i].PopFront();Span* kSpan = new Span;kSpan->_pageId = nSpan->_pageId;kSpan->_n = k;nSpan->_pageId += k;nSpan->_n -= k;_spanLists[nSpan->_n].PushFront(nSpan);return kSpan;}}

向系统申请

当k号桶往后的桶都是空的，那么我们就需要向系统申请内存了，直接申请一个128的页，放在128号桶，再执行上面的逻辑。

对于内存申请，在windows下我们使用VirtualAlloc函数，与malloc相比VirtualAlloc直接与操作系统交互，无额外开销，效率高，通常用来申请超大块内存。

VirtualAlloc的声明
LPVOID VirtualAlloc(LPVOID  lpAddress,        SIZE_T  dwSize,           DWORD   flAllocationType, DWORD   flProtect         
);
LPVOID lpAddress：期望的起始地址（通常设为0由系统决定）
SIZE_T dwSize：分配的内存大小（字节）
DWORD flAllocationType: 分配的类型（如保留或提交）
DWORD flProtect：内存保护选项（如读写权限）

返回值：LPVOID ，本质是void*，需强制类型转换后使用。

flAllocationType

MEM_COMMIT：提交内存，使其可用。

MEM_RESERVE：保留地址空间，暂不分配物理内存。

二者可组合使用（MEM_RESERVE | MEM_COMMIT），同时保留并提交。

flProtect
控制内存访问权限，常用选项：

PAGE_READWRITE：可读/写。

PAGE_NOACCESS：禁止访问（触发访问违规）。

PAGE_EXECUTE_READ：可执行/读。

因为还要考虑其他系统的需求，我们在Common.h内封装一个向系统申请内存的函数，并使用条件编译来选择不同的函数调用，如下：

#ifdef _WIN32#include <windows.h>
#else//Linux...
#endifinline 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;
}

接下来继续NewSpan的执行逻辑，向系统申请内存后，new一个Span储存相关的页信息，即计算页号（地址/8KB），填写页数，然后把Span连接到128号哈希桶，最后需要做的就是把它切割为k页的Span和(128-k)页的Span和前面的逻辑一模一样，为提高代码复用率以递归的方式返回。如下：

Span* PageCache::NewSpan(size_t k)
{assert(k > 0 && k < NPAGES);if (!_spanLists[k].Empty()){return _spanLists[k].PopFront();}for (int i = k + 1; i < NPAGES; i++){if (!_spanLists[i].Empty()){Span* nSpan = _spanLists[i].PopFront();Span* kSpan = new Span;kSpan->_pageId = nSpan->_pageId;kSpan->_n = k;nSpan->_pageId += k;nSpan->_n -= k;_spanLists[nSpan->_n].PushFront(nSpan);return kSpan;}}//向系统申请内存Span* span = new Span;void* ptr = SystemAlloc(NPAGES - 1);span->_pageId = (PANGE_ID)ptr>>PAGE_SHIFT;span->_n = NPAGES - 1;_spanLists[span->_n].PushFront(span);return NewSpan(k);
}