|
|
page cache的设计及实现
page cache 本质上也是一个哈希桶, 它是按照页的数量进行映射的. 当 central cache 向 page cache 申请内存时, page cache 先检查对应位置是否有span, 如果没有则向更大页去寻找一个span, 如果找到则分裂成两个. 比如: 你申请7page的内存, 7页后面没有挂span, 则向后面寻找更大的span, 假设在10页位置找到了span, 则将10页的span分裂成7页的span和3页的span.
如果找到 _spanLists[128] 都没有合适的span, 则向系统堆申请128页的span挂到 _spanLists[128] 上.
可能大家会疑惑, 为什么是128, 而不是别的数字?
这个问题也就等同于为什么 page cache 中最大挂128页的span? 因为在这个项目里我们规定线程申请单个对象最大是256KB, 而128页可以被切成4个256KB的对象, 因此是足够的.
当然了, 具体情况具体分析, 在别的特殊场景下也可以自行设置.
//page cache 中哈希桶的个数
static const size_t NPAGES = 129;
为了让哈希桶的下标和页数对应起来, 我们将哈希桶的个数设置成129个, 也就是将0号下标舍弃.
page cache与central cache的不同:
我们知道 central cache 和 page cache 的核心结构都是 SpanList 的哈希桶, 但是它们的本质是有区别的, central cache 的哈希桶跟 thread cache 一样, 都是按照大小对齐关系映射的, 它的 SpanList 中挂的 span 中的内存都被按映射关系切好链接成小块内存的自由链表, 而 page cache 中的 SpanList 则是按照下标桶号映射的, 也就意味着第 i 号桶下面挂的span都是 i 页内存.
还有一点就是 central cache 中的多个桶可能会同时向 page cache 申请内存, 所以 page cache 是存在线程安全问题的, 因此在访问 page cache 时必须要加锁. 但是在page cache 这里我们不能像 central cache 一样使用桶锁, 因为当 central cache 向 page cache 申请内存时, page cache 可能会将其他桶当中大页span切小, 也就是说会访问到其他桶. 而且, 当central cache 将某个span归还给 page cache 时, page cache 也会尝试将该span与其他桶当中的span进行合并.
总之就是, 在访问 page cache 时, 我们可能需要访问 page cache 中的多个桶, 如果 page cache 用桶锁就会出现大量频繁的加锁和解锁操作, 导致程序的整体效率低下. 因此我们在访问 page cache 时用一把大锁将整个 page cache 给锁住.
而 thread cache 在访问 central cache 时, 也可能出现多个 thread cache 同时访问 central cache 的情况, 但是我们的 central cache 使用的是桶锁, 因为 central cache 的每个哈希桶中的span都被切分成了对应大小的内存对象并且通过自由链表链接起来了, thread cache 只需要根据自己所需对象的大小访问 central cache 中对应的哈希桶即可, 不会访问其他哈希桶, 因此 central cache 可以用桶锁.
page cache的实现方式
page cache在整个进程中也是只能存在一个的, 因此我们也需要将其设置为和 central cache 一样的单例模式.
// page cache本质是一个按页数映射的哈希桶
class PageCache
{
public:
// 提供一个全局访问点
static PageCache* GetInstance()
{
return &_sInst;
}
// 获取一个k页的span
Span* NewSpan(size_t k);
private:
// 构造和拷贝构造设置为私有
PageCache()
{
}
PageCache(const PageCache&) = delete;
static PageCache _sInst;
private:
SpanList _spanLists[NPAGES];
public:
std::mutex _pageMtx; // 一把大锁
};
程序运行起来直接创建出该对象:
PageCache PageCache::_sInst;
page cache中获取span
获取一个非空的span
我们知道 thread cache 在向 central cache 申请内存对象时, central cache 要先从对应的哈希桶中获取一个非空的span, 然后从这个span中取出若干个内存对象给 thread cache, 试问如何获取到这个非空的span呢?
首先遍历当前位置映射的双向链表, 如果该双向链表中有非空的span, 直接返回即可. 说到遍历, 需要在SpanList的定义当中给出Begin成员函数和End成员函数, 方便我们像使用迭代器一样遍历双向链表.
// 双向链表结构
class SpanList
{
public:
SpanList()
{
_head = new Span;
_head->_prev = _head;
_head->_next = _head;
}
// 第一个span
Span* Begin()
{
return _head->_next;
}
// 最后一个span的下一个位置
Span* End()
{
return _head;
}
private:
Span* _head = nullptr;
};
如果当前位置映射的SpanList中没有非空的span, 那么只能向 page cache 中申请大块内存了.
但是需要从 page cache 中获取多大的内存呢? 这个由具体对象的大小决定, 我们可以先根据对象的大小计算出 thread cache 一次向 central cache 获取对象数量的上限值, 然后将这个上限值乘以单个对象的大小, 计算出具体需要多少字节, 最后再将这个算出来的字节数转换为页数, 如果转换后不够一页, 那么我们就申请一页, 否则转换出来是几页就申请几页. 也就是说, central cache 向 page cache 申请内存时, 要求申请到的内存尽量能够满足 thread cache 向 central cache 申请时的上限.
// 管理对齐和映射的关系
class SizeClass
{
public:
// 一次向page cache要多少页
static size_t NumMovePage(size_t size)
{
size_t num = NumMoveSize(size); // thread cache一次向central cache获取对象的上限值
size_t npage = num * size; //num个size大小的对象所需的字节数
npage >>= PAGE_SHIFT; // 将字节数转化成页数
if (npage == 0)
npage = 1; // 最少给一页
return npage;
}
};
其中 PAGE_SHIFT 表示页大小转换偏移, 这里我们按每页8K的大小为例, 那么PAGE_SHIFT 的大小就是13.
// 页大小转换偏移，即一页定义为2^13Byte，也就是8KB
static const size_t PAGE_SHIFT = 13;
当 central cache 申请到了这个若干页的span之后, 还需要将其切成一个个所需要的内存对象链接起来, 然后挂到span的自由链表当中.
那怎么对这个大块span进行切分呢? 首先根据大块内存起始页的页号计算出大块内存的起始地址, 再根据大块内存的页数计算出大块内存的大小, 然后利用起始地址和大块内存的大小计算出大块内存的结束地址.
有了大块内存的起始地址和结束地址, 那么对其进行切分就简单多了, 但是有一点需要注意的是为什么我们这里选用的是尾插, 而不是头插呢?
因为将切好的对象尾插到自由链表, 这些对象看起来是按照链式结构链接起来的, 而实际它们在物理上是连续的, 这时当我们把这些连续内存分配给某个线程使用时, 可以提高该线程的CPU缓存利用率, 这样一来CPU高速缓存命中率会更高.
补充: CPU高速缓存命中率知识点
一般程序编译连接后, 生成可执行程序, CPU执行这个程序时需要去访问内存, 但是CPU太快了, 比内存快得多, 于是CPU总是在等内存.
所以CPU不会直接去访问内存, 因为它嫌内存的速度太慢了, 所以CPU会把数据加载到三级缓存或者寄存器中, 一般小数据加载到寄存器中, 大数据加载到缓存中.
那什么叫做命中呢? CPU会看数据是否在缓存, 在就叫命中, 可以直接访问, 不在就叫不命中, 此时需要先把数据从内存加载到缓存中, 之后再访问.
所以针对CPU高速缓存命中率这个知识点, 我们可以用之前学习的顺序表和链表来举例子:
首先我们知道顺序表在物理空间上是连续的, 而链表在物理空间上是不连续的.
当我们访问顺序表第一个元素时, 会看其是否在缓存中, 不在, 不命中, 然后就把后面的一段加载到缓存中去(具体加载多少, 看硬件), 然后再访问后面的元素就都命中了.
如果是链表, 虽然也会将后面的一段数据加载到缓存中, 但是链表在物理上是不连续, 随机存储的, 所以命中率会低一些.
有了上面的基础, 请看具体代码实现:
// 从SpanList或者page cache获取一个非空的span
Span* CentralCache::GetOneSpan(SpanList& list, size_t size)
{
// 判断当前的SpanList是否有非空的span
Span* it = list.Begin();
while (it != list.End())
{
if (it->_freeList != nullptr)
{
return it;
}
else
{
it = it->_next;
}
}
// 解除桶锁, 方便thread cache释放内存回来
list._mtx.unlock();
// 走到这说明需要从page cache获取一个非空的span
PageCache::GetInstance()->_pageMtx.lock();
Span* span = PageCache::GetInstance()->NewSpan(SizeClass::NumMovePage(size));
PageCache::GetInstance()->_pageMtx.unlock();
// 计算大块内存所在的地址和大小
char* start = (char*)(span->_pageId << PAGE_SHIFT); // 大块内存的起始地址
size_t bytes = span->_n << PAGE_SHIFT; // 大块内存的大小 - 字节数
char* end = start + bytes; // 大块内存的结束地址
// 将获取到的span切成小块内存链接起来
// 先切一块下来做头, 方便尾插
span->_freeList = start;
void* tail = span->_freeList;
start += size;
while (start < end)
{
NextObj(tail) = start;
tail = start;
start += size;
}
list._mtx.lock(); // 加在前面
// 将切好的span插入central cache的SpanList中
list.PushFront(span);
return span;
}
对于上面的代码我们还需要注意的是加锁解锁问题, 了解代码后就会知道, 当 central cache 向 page cache 申请内存时, central cache 对应的哈希桶是处于加锁状态的, 那在访问 page cache 之前我们应不应该把 central cache 对应的桶锁解掉呢?
这里建议在访问 page cache 前, 先把 central cache 对应的桶锁解掉, 虽然此时 central cache 的这个桶当中是没有内存供其他 thread cache 申请的, 但 thread cache 除了在 central cache中申请内存外, 还会释放内存回到 central cache 中, 如果在访问 page cache 前将 central cache 对应的桶锁解掉, 那么此时如果有 thread cache 想要归还内存给 central cache 的这个桶时就不会被阻塞.
还需要注意的是在调用 NewSpan 函数之前, 还需要将 page cache 的大锁加上, 当申请到k页的span后, 我们需要将 page cache 的大锁解掉, 但此时我们不需要立刻获取到 central cache 中对应的桶锁. 因为 central cache 拿到k页的span后还会对其进行切分操作, 因此我们可以在span切好后需要将其挂到 central cache 对应的桶上之前, 获取对应的桶锁.
获取一个k页的span
当我们调用 GetOneSpan 函数从 central cache 的某个哈希桶获取一个非空的span时, 如果遍历哈希桶中的双向链表后发现其中没有span, 或该双向链表中的span都为空, 那么此时 central cache 就需要向 page cache 申请若干页的span.
因为 page cache 是直接按照页数进行映射的, 因此我们要从 page cache 中获取一个k页的span, 就应该直接去找 page cache 的第k号桶, 如果第k号桶中有span, 那我们直接头删一个span返回给 central cache 就行了, 如果没有, 继续向下遍历直至最后一个桶, 在这期间, 如果第n号桶挂有n页的span非空, 将其分裂成一个k页的span和一个n-k页的span.
但是如果遍历到了最后一个桶, 依然没有发现非空的span, 此时需要向系统堆申请一个大页的span, 我们这里申请的是128页的span, 直接使用我们封装的 SystemAlloc 函数即可.
// 获取一个k页的span
Span* PageCache::NewSpan(size_t k)
{
assert(k > 0 && k < NPAGES);
// 判断k号桶是否有非空的span
if (!_spanLists[k].Empty())
{
_spanLists[k].PopFront();
}
// 遍历k+1到最后一个桶
for (size_t i = k + 1; i < NPAGES; i++)
{
if (!_spanLists[i].Empty())
{
Span* kSpan = new Span;
Span* nSpan = _spanLists[i].PopFront();
// 将nSpan头部的k页切下来
kSpan->_pageId = nSpan->_pageId;
kSpan->_n = k;
nSpan->_pageId += k;
nSpan->_n -= k;
// 将nSpan剩下的挂到其映射的位置
_spanLists[nSpan->_n].PushFront(nSpan);
return kSpan;
}
}
// 走到这说明需要向系统堆申请128页的span
Span* 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);
}
转载：https://blog.csdn.net/weixin_57544072/article/details/129142764