Trading Systems: Designing C++ memory pools to avoid dynamic memory allocations

这是学习 《Building Low Latency Applications With C++: Develop a Complete Low Latency Trading Ecosystem From Scratch Using Modern C++》 的读书笔记。BTW，这本书真是一本好书，是市面上为数不多讲解交易系统架构的书籍，逐层讲解，深入浅出👍

要解决的问题

在一般的做法当中，一个非 static 的对象的生命周期与其内存生命周期是一致的，但是这样就要求动态分配内存，在 runtime 时在 heap 上给这个对象分配内存空间。但是动态内存分配的 overhead 很大。所以在 low latency applications 当中我们希望不要有动态内存分配，于是我们把 “对象生命周期” 与 “内存生命周期” 进行解耦：在启动时就申请足够的内存块，在对象需要的时候直接写入内存块，无需等待再一次内存块分配。

常见实现

1. 使用两个 vector 分别保存内存块本身和表示内存块是否空闲的变量
2. 可以使用同一个 vector 保存内存块和表示其是否空闲的变量（需要用到结构体） 我们后续的演示会采用第二种实现

分配与归还

由于已经提前申请好内存块了，因此分配与归还就是标记问题而已。

分配：

1. 找空槽位
2. 在槽位地址上构造 T
3. 标记槽位为 occupied
4. 返回 T*

归还：

1. 根据 T* 找回所述槽位
2. 标记槽位为空
3. 以后可以复用这个地址

内存池涉及的变量

• store_：停车场本身
• ObjectBlock：一个停车位
• object_：停在车位里的对象
• is_free_：车位是否空闲
• next_free_index_：下一个可能空闲的车位
• allocate()：占用车位并构造对象
• deallocate()：释放车位供复用

初始化内存池

public:
    explicit MemPool(std::size_t num_elem)
        : store_(num_elem, ObjectBlock{T(), true}) {
        ASSERT(
            reinterpret_cast<const ObjectBlock*>(
                &store_[0].object_
            ) == &store_[0],
            "T object should be first member of ObjectBlock."
        );
    }

    MemPool() = delete;
    MemPool(const MemPool&) = delete;
    MemPool(const MemPool&) = delete;
    MemPool& operator=(const MemPool&) = delete;
    MemPool& operator=(const MemPool&) = delete; 

这一段代码要素过多，以下逐个拆解

1. 使用 explicit，是为了如下的禁止隐式转换

MemPool<int> pool = 100;

假设没有 explicit，隐式转换按照如下逻辑展开：

• 目标类型已经确定为 MemPool
• 初始值是 int 类型的 100
• 查找能把 100 转换成 MemPool 的非 explicit 构造函数
• 找到 MemPool(std::size_t num_elem)
• 先把 100 从 int 转换成 std::size_t
• 调用构造函数创建 pool 在低延迟系统当中，隐式转换带来以下问题：
• 可能在关键路径出现隐藏的动态分配
• 隐式转换陈胜的 MemPool 可能只是临时对象，会按调用临时创建和销毁（违背使用内存池的初衷）
• 数值转换可能出现错误（比如把 -1 转成一个极大的 size_t 正数）

2. store_(num_elem, ObjectBlock{T(), true}) 这里是在初始化 store_ 这个对象，需要结合 MemPool 的私有成员来理解

private:
    struct ObjectBlock { 
        T object_;
        bool is_free_ = true;
        };
    std::vector<ObjectBlock> store_;
    size_t next_free_index_ = 0;

store_ 是真正用来存放内存块（又叫槽位）的 vector。这里初始化这个 vector 的时候给出了两个参数 第一个参数 num_elem 表示为 vector 分配多少个元素的空间（内存池的预先分配就是这么实现的）； 第二个参数 ObjectBlock{T(), true} 表示把每个元素初始化为什么，这里调用了 ObjectBlock 对象的构造函数

因此 store_(num_elem, ObjectBlock{T(), true}) 表示初始化时为 num_elem 个元素申请了槽位，每个槽位当中能够容纳一个 ObjectBlock，每个 ObjectBlock 都被初始化为 { T 对象, true }

这里的 true 表示该槽位当前空闲，可以容纳一个 T 对象。

3. assert

ASSERT(
    reinterpret_cast<const ObjectBlock*>(
        &store_[0].object_
    ) == &store_[0],
    "T object should be first member of ObjectBlock."
);

这里确保第一个 ObjectBlock 的地址等于其中的 T 类型 object_ 的地址 这件事情看起来像多此一举，但是是为了确保内存池的 deallocate() 使用指针访问 ObjectBlock 的时候可以确保正确操作 is_free_ 变量，不会误伤其它变量

4. 确保其它构造函数都 =delete 确保它们不会在我们未知的情况下被使用，不会造成隐式转换（与上述函数使用 explicit 的原因一致）

MemPool() = delete;
MemPool(const MemPool&) = delete;
MemPool(const MemPool&) = delete;
MemPool& operator=(const MemPool&) = delete;
MemPool& operator=(const MemPool&) = delete; 

allocate() 为对象分配空槽位

template<typename... Args>
T* allocate(Args... args) noexcept {
    auto obj_block = &(store_[next_free_index_]);

    ASSERT(obj_block->is_free_, ...);

    T* ret = &(obj_block->object_);
    ret = new(ret) T(args...);

    obj_block->is_free_ = false;
    updateNextFreeIndex();

    return ret;
}

1. next_free_index_ 始终指向一个空闲槽位

• 所以 auto obj_block = &(store_[next_free_index_]); 无需搜索，可以直接分配内存地址

2. 验证不变量： ASSERT(obj_block->is_free_, ...)

• 这些 ASSERT 在实际生产构建中会被关闭，因此只是在开发过程中用到
• 整个 MemPool 程序中这么多断言就是为了尽早发现错误

3. 获取槽位中 T 的地址

T* ret = &(obj_block->object_)

得到一个指向 T 类型的 object_ 的地址

ObjectBlock
┌──────────────────────┐
│ object_ 的存储空间    │ ← ret
├──────────────────────┤
│ is_free_             │
└──────────────────────┘

4. 使用 placement new 构造对象

ret = new(ret) T(args...)

placement new 和 普通 new 的区别在于：

• 普通 new 完成 【申请内存】+【构造对象】两件事情
• placement new 只做第二件：【在指定地址上构造对象】

5. 标记槽位已经占用

obj_block->is_free_ = false;

6. 提前寻找下一个空槽位

updateNextFreeIndex();

见下

updateNextFreeIndex() 寻找下个空槽位

private:
    auto updateNextFreeIndex() noexcept {
    const auto initial_free_index = next_free_index_;
    while (!store_[next_free_index_].is_free_) {
        ++next_free_index_;

        if (UNLIKELY(next_free_index_ == store_.size())) {
            // hardware branch predictor should almost always predict this to be false any ways.
            next_free_index_ = 0;
        }

        if (UNLIKELY(initial_free_index == next_free_index_)) {
            ASSERT(initial_free_index != next_free_index_, "Memory Pool out of space.");
        }
    }
} 

从【本次分配的槽位】开始依次向后扫描 if (UNLIKELY(next_free_index_ == store_.size())) 是处理绕到 vector 尾部，需要从 vector 头部继续扫描的情况 if (UNLIKELY(initial_free_index == next_free_index_)) 表明如果扫描回到了【本次分配的槽位】，说明内存池已经满了

deallocate() 把槽位标记为空闲

auto deallocate(const T* elem) noexcept {
    const auto elem_index = reinterpret_cast<const ObjectBlock*>(elem) - &store_[0];

    ASSERT(
        elem_index >= 0 &&
        static_cast<size_t>(elem_index) < store_.size(),
        "Element does not belong to this pool."
    );

    ASSERT(!store_[elem_index].is_free_);

    store_[elem_index].object_.~T();
    store_[elem_index].is_free_ = true;
}

T* elem 是要释放的对象地址

const auto elem_index = reinterpret_cast<const ObjectBlock*>(elem) - &store_[0];

• elem 要从 T* 转换成 ObjectBlock* ，因为内存池当中的槽位其实是一个个 ObjectBlock，T 只是 ObjectBlock 的一部分，要释放的是整个 ObjectBlock。
• 这里减去 &store_[0] 是为了得到该 ObjectBlock 在整个 vector 当中的索引 elem_index

store_[elem_index].object_.~T();

调用 object_ （T 类型）的析构函数

store_[elem_index].is_free_ = true;

• 然后这里就用这个索引 elem_index，在 vector 当中把对应的 ObjectBlock 的 is_free 标为 true

使用内存池

#include "mem_pool.h"
struct MyStruct {
    int d_[3];
};
int main(int, char **) {
    using namespace Common;
    MemPool<double> prim_pool(50);
    MemPool<MyStruct> struct_pool(50);
    for(auto i = 0; i < 50; ++i) {
        auto p_ret = prim_pool.allocate(i);
        auto s_ret = struct_pool.allocate(MyStruct{i, i+1, i+2});
        std::cout << "prim elem:" << *p_ret << " allocated at:" << p_ret << std::endl;
        std::cout << "struct elem:" << s_ret->d_[0] << "," << s_ret->d_[1] << "," << s_ret->d_[2] << " allocated at:" << s_ret << std::endl;
        if(i % 5 == 0) {
            std::cout << "deallocating prim elem:" << *p_ret << " from:" << p_ret << std::endl;
            std::cout << "deallocating struct elem:" << s_ret ->d_[0] << "," << s_ret->d_[1] << "," << s_ret->d_[2] << " from:" << s_ret << std::endl;
            prim_pool.deallocate(p_ret);
            struct_pool.deallocate(s_ret);
        }
    }
    return 0;
} 

运行这段代码会看到的结果是：槽位被分配与回收。回收后的槽位并未归还系统，而是被标记为空闲并可能被再次分配给其它内存。