C++ 高并发内存池 ThreadCache 设计与实现

ThreadCache 设计

大致思路：这里对于线程申请内存就往 threadcache 要，然后它就会向底层的 freelist 要，如果没有就会向 Centralcache 索取一块给对应的线程用，然后剩下的就挂到对应的 freelist 里面方便下次使用，回收内存就挂到对应 freelist 里面 (可能多了填充到对应的 CentralCache 里面，这里先暂不做处理)。

容量设计：

static const int MAX_BYTES = 256 * 1024; // 线程缓存最大 256kb
static const int FREELIST_NUMS = 208;    // 哈希桶个数

整体大致设计：

class ThreadCache {
public:
    void* Allocate(size_t size);      // 指针以及对应空间实际长度
    void Deallocate(void* ptr, size_t size); // ThreadCache 没了就向 CentralCache 索取
    void* FetchFromCentralCache(size_t blocksize, size_t index);

private:
    FreeList _freelists[FREELIST_NUMS]; // 哈希桶数组
};

首先先看下对应的块的大小及分布的 freelist 的下标位置：

• [1, 128] 8byte 对齐 freelist[0,16)
• [128+1, 1024] 16byte 对齐 freelist[16,72)
• [1024+1, 8*1024] 128byte 对齐 freelist[72,128)
• [81024+1, 641024] 1024byte 对齐 freelist[128,184)
• [641024+1, 2561024] 8*1024byte 对齐 freelist[184,208)

这里在一定范围内就按照指定对齐数向上取整得到对应大小内存块，我们为了拥有更大的利用率，因此进行这样设计！

其次就是这里还会出现两个问题：

1. 申请内存块的时候大小不一定，如何找到对应 freelist 中真正内存块大小呢？

上面我们说了，每次申请内存并不会直接给你对应大小的，比如你申请 15；它会给你一个 16 的内存块，依此类推，可以理解成有个补齐的原则。

判断对齐数是多少：

// 判断要开辟的真实字节数（也就是哪一个哈希桶）
static size_t Real_Size(size_t size) {
    if (size <= 128) {
        return _Real_Size(size, 8);
    } else if (size <= 1024) {
        return _Real_Size(size, 16);
    } else if (size <= 8 * 1024) {
        return _Real_Size(size, 128);
    } else if (size <= 64 * 1024) {
        return _Real_Size(size, 1024);
    } else if (size <= 256 * 1024) {
        return _Real_Size(size, 8 * 1024);
    } else {
        assert(false);
        return -1;
    }
}

获得对应的真实字节数：

static size_t _Real_Size(size_t size, size_t alignnum) {
    size_t rsize;
    if (size % alignnum == 0)
        rsize = size;
    else
        rsize = (size / alignnum + 1) * alignnum;
    return rsize;
}

但是这种方法因为用到了除法，乘法等，对于要求非常高的时候，效率会是十分低的，因此改进成对应的位操作：

static inline size_t _Real_Size(size_t bytes, size_t alignNum) {
    return ((bytes + alignNum - 1) & ~(alignNum - 1));
}

解释下：

比如对齐数是 8，拿到的是 1-7，此时给它 +7 就是 8-14；此时 ~(alignNum - 1) 后最低位是 8 的位是 1；此时 8-14 对应 8 的那位为 1 后面无论是几&后都是 0；因此 real_size 就是 8。

一句话总结：

(bytes + alignNum - 1) 完成对不足对齐数倍数进行进位，此时进了一位或者一位多一点 & ~(alignNum - 1) 干掉不足对齐数的那些位。

2. 如何根据想要申请的字节数找到对应真实内存块的 freelist 的下标呢？

可以判断这个大小字节落在哪个对齐数的桶集内，然后通过求出它落在这个这个桶集的第几个 freelist 里面，然后再加上之前的那些桶的下标即可。

获得落在哪个桶集，并累加之前桶集的总数：

static size_t Index(size_t size) {
    // 相同对齐数的哈希桶分一组（前四种）：
    int groups[4] = {16, 56, 56, 56};
    if (size <= 128) {
        return _Index(size, 8);
    } else if (size <= 1024) {
        return groups[0] + _Index(size - 128, 16);
    } else if (size <= 8 * 1024) {
        return groups[0] + groups[1] + _Index(size - 1024, 128);
    } else if (size <= 64 * 1024) {
        return groups[0] + groups[1] + groups[2] + _Index(size - 64 * 1024, 1024);
    } else if (size <= 256 * 1024) {
        return groups[0] + groups[1] + groups[2] + groups[3] + _Index(size - 256 * 1024, 8 * 1024);
    } else {
        assert(false);
        return -1;
    }
}

求出占了当前桶集的多少 freelist：

// 计算落在对齐数为 alignnum 的那些哈希桶中具体的哈希桶即 freelist 下标：
static size_t _Index(size_t size, size_t alignnum) {
    size_t index;
    if (size % alignnum == 0)
        index = size / alignnum - 1;
    else
        index = size / alignnum;
    return index;
}

当然这里也是有优化的（按位操作）:

static inline size_t _Index(size_t bytes, size_t alignnum) {
    size_t _2_top = static_cast<size_t>(std::log2(alignnum)); // 求 alignnum 是 2 的哪个指数次结果
    return ((bytes + (1 << _2_top) - 1) >> _2_top) - 1;
}

解释下：

bytes 是落在某个对齐数这个区间内的多余的字节（假设对齐数就是 16），那么此时 bytes（假设是 16-31）+15 一定落在 16 32 48 两个区间内，其中要么 bytes 为 16 落在 16 -32 剩下的就是在 32-48 区间内，此时，进行除 对齐数，此时 -1 后，当 bytes 为 16 就是 0，其余就是 1；注意下标：因为第一种桶已经计算了是 16 个，这里 -1 就是多少个桶与下标之间的对应关系映射好！

一句话总结：

(bytes + (1 << _2_top) - 1) 为了给不足一个对齐数倍数的进位；>>_2_top 获得真正是哪个桶 - 1 完成是 多少个桶与下标之间的对应关系 映射。

Allocate 实现

通过对应的 size 获取实际内存块大小，并定位到哪个 freelsit 然后进行 pop；如果 freelist 为空，那就去 CentralCache 对应大小块里面去申请，取出一个使用，剩下的挂回去（之后会结合 CentralCache 实现下，这里先不实现了）。

void* ThreadCache::Allocate(size_t size) {
    assert(size <= MAX_BYTES); // 小于 256kb
    size_t index = Calculate::Index(size); // 工具类里的函数必须设置静态才能访问否则由于 this 指针不能直接访问
    if (!_freelists[index].empty()) {
        return _freelists[index].Pop();
    } else {
        // 一定字节内存块的哈希桶空了，向中心缓存申请：
        size_t block_size = Calculate::Real_Size(size);
        return FetchFromCentralCache(block_size, index);
    }
}

FetchFromCentralCache：从中心缓存获取内存块，这里暂不实现了，之后结合 CentralCache 原理实现！

Deallocate 实现

直接拿到这块不用的内存块，挂回对应的 freelist（记住千万不要释放掉，后续还可能用到，它也没用这个权利）

void ThreadCache::Deallocate(void* ptr, size_t size) {
    assert(ptr);
    assert(size);
    size_t index = Calculate::Index(size);
    _freelists[index].Push(ptr);
}

基于多线程私有 ThreadCache 设计

首先，后期我们使用这个高并发内存池的时候，是多线程使用的，因此每个线程都要独立拥有一个 ThreadCache，它们互相在这里面索取内存块，不用加锁，互不干扰。

因此我们进行这样的操作：

__declspec(thread) ThreadCache* TLSThreadCache = nullptr;

解释下：

其次就是 _declspec(thread) win 专属的，修饰的类型的对象是线程本地储存，也就是线程看到这个对象（被修饰的）就会把它搬到线程局部存储每个线程就不在共享这个全局变量而是私自占有不同的—>就产生了 线程缓存（无锁）。

但是这里还会出现个问题：重定义问题！

解决方法：

• 这里需要搞成 static；否则多个 cpp 包含这个头文件链接 obj 的时候每个 cpp 都有这个定义，互相能看到就重定义了，因此 static 一下只在 本 cpp 有效 互相不能看到了，避免重定义！

static __declspec(thread) ThreadCache* TLSThreadCache = nullptr;

代码编写注意事项

• 之前基于 Linux，我们采取的是 hpp（直接把 h 和 cpp 合并了，故类内直接写函数声明与定义即可）格式完成对类的分块处理，这里采取的是 h 和 cpp 的形式（h 里声明，cpp 里定义）：因此可以考虑在 cpp 写定义的时候需要声明这个函数的类名，剩下的就想象成在这个类内定义都是一样的！
• 其次就是把多个模块经常用到的函数等封装成一个类（可以叫工具类），然后放在一个公共头文件里（common 头文件）；这样就大大简化了麻烦，直接包含这个文件，调用这个类函数就行；我们直接把它搞成静态成员函数，这样每次使用都不需要定义对象，直接声明类，使用这个函数即可。静态成员函数属于类本身，而不是类的某个特定对象。因此，你可以通过类名直接调用静态成员函数，而不需要创建类的实例。

三、ThreadCache 的多线程测试

下面我们简单搞出两个线程来测试下，看是否它们都能拿到各自的 ThreadCache，也就是看每个线程的 TLSThreadCache 是否是不同的即可（如果是不同的也就是每个线程都拿到了不同的 ThreadCache 这个类对象的地址，此时也就保证了它们各自有各自的线程缓存空间了，达成了互不干扰）。

搞出两个线程走从自己的 ThreadCache 申请内存:

void alloc1() {
    for (int i = 0; i < 3; i++) {
        void* ptr = ConcurrentAlloc(7);
    }
}

void alloc2() {
    for (int i = 0; i < 3; i++) {
        void* ptr = ConcurrentAlloc(10);
    }
}

int main() {
    std::thread t1(alloc1);
    t1.join();
    std::thread t2(alloc2);
    t2.join();
}

这两个线程都会走 ConcurrentAlloc 这个函数，但是由于 TLSThreadCache 是线程局部存储，每个线程看到的都是自己的，当访问 Allocate 的时候，也是访问线程独有的（这里可以认为 new 的时候把这个函数已经保存在线程自己的局部储存里面了），因此此种情况下，Allocate 是个可重入函数，多线程访问不会出现问题：

void* ConcurrentAlloc(size_t size) {
    if (TLSThreadCache == nullptr)
        TLSThreadCache = new ThreadCache;
    // TLSThreadCache 为线程局部自己的，因此每个线程都不一样，每次线程拿到的都是自己局部储存里面的 TLSThreadCache，
    // 实现每个线程管理自己的 ThreadCache，非共享的，无锁化
    cout << std::this_thread::get_id() << ":" << TLSThreadCache << endl;
    return TLSThreadCache->Allocate(size);
}

测试结果

第一步先打断点在进行调试：

首先进行调试：

打开并行监视对 TLSThreadCache 指针进行并行监视：

然后观察对应的线程是否可以拿到只属于自己的 threadcache 对象指针，也就是验证是否每个线程有自己独立的 threadcache

第一个线程走第二次申请内存块：

发现获取的和第一次创建的地址一样！

下面换到第二个线程，对应的线程 tid 是不同的，拿到的对应的 threadcache 对象地址也不同：

之后当第二个线程下一次再去 threadcache 申请内存拿到的仍是创建的那个对象地址，同上面，这里就不演示了！

看一下最后的效果：两个不同线程对应两个不同 threadcache 指针，因此验证了每个线程有自己独立的 threadcache！

本次测试成功证明了每个线程都可以拿到自己独立的 ThreadCache，之后我们就可以直接给他们分配内存块，实现多线程各自向自己的线程缓存去申请空间使用。

总结

通过本篇，学习到这个高并发内存池的整体框架，以及之间的相互联系，然后简单的实现了线程独有的 ThreadCache 这个模块的一小部分（后续还需要继续补充，这里便对它有了个认识），之后会更新关于 CentralCache，PageCache 的设计实现等等。