发布于 由admin

C++ 与 NUMA 拓扑：在分布式存储系统中实现数据本地化访问的调度算法

C++ 与 NUMA 拓扑：在分布式存储系统中实现数据本地化访问的调度算法

各位专家、开发者们，大家好！

在现代高性能计算领域，CPU 的核心数量与日俱增，内存容量也越来越大。然而，这种硬件的飞速发展也带来了一个隐而未现但影响深远的性能瓶颈——内存访问。随着处理器集成度的提高，传统的统一内存访问（UMA）架构已经无法满足多核处理器对内存带宽和延迟的需求。取而代之的是非统一内存访问（Non-Uniform Memory Access, NUMA）架构。

NUMA 架构的出现，使得内存不再是“一视同仁”的资源。访问本地 NUMA 节点的内存通常比访问远端 NUMA 节点的内存快得多。在分布式存储系统中，数据本地化访问是提升性能的关键。一个存储节点内部的 NUMA 拓扑如果被忽视，即使网络和磁盘 I/O 已经优化到极致，也可能因为频繁的远端内存访问而导致性能瓶颈。

今天，我们将深入探讨 NUMA 拓扑如何影响 C++ 应用程序，特别是在分布式存储系统中，以及我们如何设计和实现智能调度算法，利用 C++ 的强大能力和操作系统提供的底层接口，实现数据和计算的本地化，从而榨取硬件的极致性能。

开篇：NUMA拓扑与现代计算的挑战

在过去，CPU 的频率和指令级并行是性能提升的主要驱动力。然而，随着摩尔定律的减缓以及功耗墙的限制，CPU 制造商转向了多核架构。现在，一个典型的服务器处理器可能拥有几十个甚至上百个逻辑核心。这些核心需要访问大量内存，而传统的单内存控制器设计，使得所有核心共享同一组内存通道，很快就达到了带宽极限。

为了解决这个问题，NUMA 架构应运而生。它将系统内存划分为多个独立的“NUMA 节点”，每个节点通常包含一个或多个 CPU 插槽上的处理器以及与这些处理器直接相连的本地内存。这些 NUMA 节点通过高速互联网络（如 Intel 的 QPI/UPI，AMD 的 Infinity Fabric）相互连接。

NUMA 的核心思想在于：

  1. 内存局部性：每个 CPU 核心访问其本地 NUMA 节点上的内存速度最快，延迟最低。
  2. 远端访问成本：访问其他 NUMA 节点上的内存需要通过互联网络，这会引入显著的延迟和带宽开销。

在单体应用中，如果一个线程在某个 NUMA 节点上运行，但它频繁访问的内存数据却分配在另一个 NUMA 节点上，那么每次数据访问都会产生额外的延迟，严重影响性能。在分布式存储系统中，这个问题被放大。一个存储节点通常承载了大量的数据块和并发请求。如果数据块被随机地分配在不同的 NUMA 节点上，或者处理请求的线程没有与数据所在的 NUMA 节点对齐，那么即使数据已经在内存中（例如，作为缓存），也可能因为 NUMA 效应而导致性能下降，甚至远超预期。

因此，理解 NUMA 架构，并利用 C++ 及其相关的操作系统接口进行精细的内存和线程调度，是构建高性能分布式存储系统的必由之路。

理解NUMA架构：内存访问的层次与成本

为了更好地进行 NUMA 优化，我们首先需要深入理解其架构细节和内存访问成本。

一个典型的 NUMA 系统由多个 NUMA 节点组成。每个节点可以看作是一个独立的微型计算机，拥有自己的处理器核心、L3 缓存以及本地 DRAM。节点之间通过高速互联总线通信，以维持缓存一致性并允许跨节点访问内存。

内存访问的层次与成本大致如下：

  1. CPU 寄存器：最快，纳秒级甚至更低。
  2. L1/L2 缓存：CPU 核心私有或共享，几纳秒。
  3. L3 缓存：通常是 NUMA 节点内所有核心共享，几十纳秒。
  4. 本地 DRAM：与当前 CPU 核心在同一个 NUMA 节点上的物理内存，几十到一百多纳秒。
  5. 远端 DRAM：位于其他 NUMA 节点上的物理内存，需要通过互联总线访问，可能高达几百纳秒，甚至更多。

表格：NUMA 节点间内存访问的相对成本示例（具体数值因硬件而异）

访问类型 相对延迟（倍数，以L1 Cache为基准） 典型延迟范围（纳秒） 备注
L1 Cache 1 ~0.5 – 1 CPU核心私有
L2 Cache ~3 – 4 ~3 – 4 CPU核心私有
L3 Cache ~10 – 20 ~10 – 20 NUMA节点内共享
本地 DRAM ~100 – 200 ~60 – 150 通过本地内存控制器直接访问
远端 DRAM ~300 – 600 ~200 – 400+ 需跨QPI/UPI等互联总线访问，延迟显著增加

我们可以使用 numactl --hardware 命令来查看系统的 NUMA 拓扑结构：

$ numactl --hardware
available: 2 nodes (0-1)
node 0 cpus: 0 1 2 3 4 5 6 7 8 9 10 11
node 0 size: 64446 MB
node 0 free: 59723 MB
node 1 cpus: 12 13 14 15 16 17 18 19 20 21 22 23
node 1 size: 64500 MB
node 1 free: 60012 MB
node distances:
node   0   1
  0:  10  21
  1:  21  10

上述输出表明系统有两个 NUMA 节点（0 和 1）。节点 0 包含了 CPU 0-11，节点 1 包含了 CPU 12-23。node distances 表明了节点间访问的相对成本，通常 10 代表本地访问，20+ 代表远端访问，数字越大，成本越高。这为我们进行 NUMA 优化提供了直接的依据。

C++与NUMA内存管理：OS层面的接口与实践

C++ 标准库提供的 new 和 malloc 等内存分配函数通常不具备 NUMA 感知能力。它们的默认行为通常遵循所谓的“first-touch”原则：当一个页面首次被写入时，操作系统会将其分配到执行写入操作的线程所在的 NUMA 节点上。

“first-touch”原则虽然简单，但在 NUMA 环境下却可能导致问题。例如，如果一个线程在 NUMA 节点 0 上分配了一块内存，但实际写入和大量访问这块内存的线程却在 NUMA 节点 1 上，那么这块内存就会被分配到节点 0，而节点 1 的线程每次访问都会产生远端访问开销。这被称为“内存错位”。

为了解决这个问题，我们需要借助操作系统提供的底层 API 来精确控制内存和线程的 NUMA 亲和性。在 Linux 系统上，主要通过 libnuma 库和一些系统调用来实现。

主要的 NUMA 相关 API：

  1. libnuma 库：提供了一系列 C 函数来查询 NUMA 拓扑、分配内存和设置进程/线程亲和性。

    • numa_available()：检查系统是否支持 NUMA。
    • numa_node_of_cpu(cpu_id)：获取指定 CPU 所属的 NUMA 节点 ID。
    • numa_num_configured_nodes()：获取配置的 NUMA 节点数量。
    • numa_alloc_onnode(size, node_id)：在指定 NUMA 节点上分配内存。
    • numa_alloc_local(size)：在当前线程所在的 NUMA 节点上分配内存。
    • numa_alloc_interleave(size, nodemask)：将内存交错分配到 nodemask 指定的节点上。
    • numa_set_preferred(node_id)：设置进程/线程的首选 NUMA 节点，后续的 malloc 或 new 会倾向于在此节点分配。
    • numa_run_on_node(node_id) / numa_run_on_node_mask(nodemask)：将当前线程绑定到指定 NUMA 节点或节点集合。
    • numa_set_localalloc()：将内存策略设置为本地分配。
    • numa_set_membind(nodemask)：将内存策略设置为严格绑定到 nodemask 指定的节点。

  2. 系统调用：

    • mbind(addr, len, policy, nodemask, max_node, flags)：将一个内存区域绑定到一个或一组 NUMA 节点上。这是 libnuma 内部实现 numa_alloc_onnode 等函数的基础。
    • set_mempolicy(policy, nodemask, max_node)：设置进程或线程的默认内存分配策略。
    • sched_setaffinity(pid, cpusetsize, cpuset)：将进程或线程绑定到指定的 CPU 核心集合上。pthread_setaffinity_np() 是针对线程的 POSIX 封装。

C++ 代码示例：使用 libnuma 进行内存分配和线程绑定

#include <iostream>
#include <vector>
#include <numeric>
#include <thread>
#include <chrono>
#include <stdexcept>
#include <cstring> // For memset
#include <numa.h>
#include <numaif.h> // For mbind/set_mempolicy definitions
#include <sched.h>  // For sched_setaffinity

// 检查NUMA支持
void check_numa_support() {
    if (numa_available() == -1) {
        std::cerr << "NUMA support not available on this system." << std::endl;
        exit(1);
    }
    std::cout << "NUMA support is available. Number of nodes: " 
              << numa_num_configured_nodes() << std::endl;
}

// 在指定NUMA节点上分配内存
void* allocate_on_numa_node(size_t size, int node_id) {
    void* ptr = numa_alloc_onnode(size, node_id);
    if (ptr == nullptr) {
        std::cerr << "Failed to allocate memory on NUMA node " << node_id << std::endl;
        throw std::bad_alloc();
    }
    std::cout << "Allocated " << size << " bytes on NUMA node " << node_id 
              << " at address " << ptr << std::endl;
    // 第一次写入，确保内存真正被分配到指定节点
    memset(ptr, 0, size); 
    return ptr;
}

// 释放NUMA内存
void free_numa_memory(void* ptr, size_t size) {
    numa_free(ptr, size);
    std::cout << "Freed memory at address " << ptr << std::endl;
}

// 将当前线程绑定到指定CPU核心
void bind_thread_to_cpu(int cpu_id) {
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(cpu_id, &cpuset);

    if (pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset) != 0) {
        std::cerr << "Error binding thread to CPU " << cpu_id << ": " << strerror(errno) << std::endl;
    } else {
        std::cout << "Thread " << std::this_thread::get_id() << " bound to CPU " << cpu_id << std::endl;
    }
}

// 获取当前线程所在的NUMA节点ID
int get_current_numa_node() {
    int node = numa_node_of_cpu(sched_getcpu());
    if (node == -1) {
        std::cerr << "Error getting current NUMA node." << std::endl;
        return -1;
    }
    return node;
}

// 模拟工作函数，访问数据并检查内存所在节点
void worker_function(int thread_id, int cpu_id, int* data, size_t data_size, int expected_node_id) {
    bind_thread_to_cpu(cpu_id);
    int current_node = get_current_numa_node();
    std::cout << "Worker " << thread_id << " (on CPU " << cpu_id << ", NUMA node " << current_node 
              << ") starting to process data." << std::endl;

    // 访问数据
    long long sum = 0;
    for (size_t i = 0; i < data_size / sizeof(int); ++i) {
        sum += data[i];
    }
    std::cout << "Worker " << thread_id << " finished. Sum: " << sum << std::endl;

    // 验证数据所在内存节点 (通过get_mempolicy/mbind获取，这里简化为检查期望值)
    // 实际检查需要更复杂的mbind/get_mempolicy调用
    // long pagesize = sysconf(_SC_PAGESIZE);
    // int mem_node = -1;
    // mbind(data, pagesize, MPOL_PREFERRED, NULL, 0, MPOL_F_MEMBIND); 
    // get_mempolicy(&mem_node, NULL, 0, (void*)data, MPOL_F_ADDR);
    // std::cout << "Data at " << data << " is actually on NUMA node " << mem_node << std::endl;
}

int main() {
    check_numa_support();

    const size_t data_size = 1024 * 1024 * 100; // 100MB
    int* data_node0 = nullptr;
    int* data_node1 = nullptr;

    try {
        // 在NUMA节点0上分配数据
        data_node0 = static_cast<int*>(allocate_on_numa_node(data_size, 0));
        std::iota(data_node0, data_node0 + (data_size / sizeof(int)), 1); // 填充数据

        // 在NUMA节点1上分配数据
        data_node1 = static_cast<int*>(allocate_on_numa_node(data_size, 1));
        std::iota(data_node1, data_node1 + (data_size / sizeof(int)), 1000); // 填充数据

        // 启动两个线程，分别绑定到不同的NUMA节点上的CPU
        // 假设CPU 0在NUMA节点0，CPU 12在NUMA节点1 (根据numactl --hardware输出)
        std::thread t0(worker_function, 0, 0, data_node0, data_size, 0);
        std::thread t1(worker_function, 1, 12, data_node1, data_size, 1);
        std::thread t2(worker_function, 2, 0, data_node1, data_size, 1); // 故意让节点0的线程访问节点1的数据

        t0.join();
        t1.join();
        t2.join();

    } catch (const std::bad_alloc& e) {
        std::cerr << "Memory allocation error: " << e.what() << std::endl;
    } catch (const std::exception& e) {
        std::cerr << "An error occurred: " << e.what() << std::endl;
    }

    if (data_node0) free_numa_memory(data_node0, data_size);
    if (data_node1) free_numa_memory(data_node1, data_size);

    return 0;
}

编译指令:
g++ -std=c++17 -o numa_example numa_example.cpp -lnuma -lpthread
运行指令:
numactl --interleave=all ./numa_example (或者直接 ./numa_example 观察默认行为)

上述代码展示了如何使用 libnuma 在指定 NUMA 节点上分配内存，以及如何使用 pthread_setaffinity_np 将线程绑定到特定的 CPU 核心。通过运行 t2 线程，我们可以观察到当线程访问远端 NUMA 内存时，其性能可能不如 t0 和 t1 线程。

分布式存储系统中的数据本地化：为何NUMA至关重要？

在分布式存储系统中，每个存储节点（或存储服务器）通常管理着大量的数据块（chunks、blocks、objects）。这些数据块可能以文件、数据库页或自定义格式存储在本地磁盘上，并可能被缓存到内存中以加速访问。

传统分布式存储系统面临的挑战：

  • 网络延迟：数据在不同存储节点之间传输的延迟。
  • 磁盘 I/O 瓶颈：数据从磁盘加载到内存的速度。

在 NUMA 架构下，即使解决了上述挑战，新的瓶颈也可能出现：

  • 内存缓存命中，但 NUMA 错位：一个数据块被成功加载到内存缓存中，但它被分配到了 NUMA 节点 A。如果处理该数据块的请求线程在 NUMA 节点 B 上运行，那么每次访问缓存数据都会成为远端 NUMA 访问，导致延迟显著增加，带宽受限。
  • 元数据服务：分布式存储系统通常有元数据服务（例如，管理文件路径、数据块位置、权限等）。这些元数据结构可能非常庞大且访问频繁。如果元数据结构横跨多个 NUMA 节点，或元数据访问线程与元数据存储节点不一致，性能会急剧下降。
  • 计算密集型任务：一些存储操作可能涉及数据处理（如压缩、加密、校验）。如果这些计算任务的数据位于远端 NUMA 节点，那么计算线程在加载数据时会遇到性能瓶颈。

分布式存储系统中的 NUMA 优化目标：

  • 数据与计算对齐：将处理特定数据块的线程（或线程池）与该数据块所在的 NUMA 节点尽可能地对齐。
  • 缓存本地化：确保热点数据缓存尽可能地位于访问它们的 CPU 核心的本地 NUMA 节点上。
  • 元数据本地化：关键元数据结构应被妥善分配，并由本地 NUMA 节点上的线程访问。

通过实现这些目标，我们可以最大限度地减少远端 NUMA 内存访问，提高内存带宽利用率，降低访问延迟，从而显著提升整个分布式存储系统的吞吐量和响应速度。

调度算法核心原则：实现数据本地化访问

设计 NUMA 感知调度算法需要遵循一些核心原则，以确保数据和计算的有效协同：

  1. 亲和性（Affinity）：这是 NUMA 优化的核心。目标是将数据、处理数据的线程以及用于缓存数据的内存区域尽可能地绑定到同一个 NUMA 节点上。
    • 内存亲和性：在分配内存时指定 NUMA 节点。
    • 线程亲和性：将线程绑定到特定的 CPU 核心或 NUMA 节点。
  2. 可见性（Awareness）：系统必须能够感知到 NUMA 拓扑结构，包括有多少个 NUMA 节点、每个节点有多少 CPU 核心、内存大小以及节点间的距离。此外，系统还需要知道数据块当前位于哪个 NUMA 节点。
  3. 迁移避免（Migration Avoidance）：频繁地在 NUMA 节点之间迁移数据或线程会引入额外的开销。理想情况下，一旦数据或线程被放置在某个 NUMA 节点上，就应尽量保持其位置。
  4. 负载均衡（Load Balancing）：虽然本地化很重要，但也不能忽视负载均衡。过度地将所有工作集中在一个 NUMA 节点上，可能导致该节点过载，而其他节点空闲。因此，需要在本地化和整体系统负载均衡之间找到一个最佳平衡点。
  5. 预测性（Predictability）：在可能的情况下，通过分析历史访问模式或预先定义的策略，预测哪些数据将是热点数据，并将其预先放置在合适的 NUMA 节点上。

基于NUMA的调度算法设计与实现

在分布式存储系统中，我们可以将 NUMA 感知调度算法分为几种策略，它们可以单独使用，也可以组合使用。

预备知识：系统NUMA拓扑获取

在 C++ 程序中，获取 NUMA 拓扑信息通常通过 libnuma 库实现。

#include <numa.h>
#include <vector>
#include <iostream>

struct NumaNodeInfo {
    int node_id;
    std::vector<int> cpus;
    long long total_memory_mb;
    long long free_memory_mb;
};

std::vector<NumaNodeInfo> get_numa_topology() {
    std::vector<NumaNodeInfo> topology;
    if (numa_available() == -1) {
        std::cerr << "NUMA not available." << std::endl;
        return topology;
    }

    int num_nodes = numa_num_configured_nodes();
    for (int i = 0; i < num_nodes; ++i) {
        NumaNodeInfo info;
        info.node_id = i;

        // 获取CPU列表
        struct bitmask *cpu_mask = numa_allocate_cpumask();
        if (numa_node_to_cpus(i, cpu_mask) == 0) {
            for (int cpu_id = 0; cpu_id < cpu_mask->size; ++cpu_id) {
                if (numa_bitmask_isbitset(cpu_mask, cpu_id)) {
                    info.cpus.push_back(cpu_id);
                }
            }
        }
        numa_free_cpumask(cpu_mask);

        // 获取内存信息 (libnuma直接获取free memory可能不准确，通常需要/proc/meminfo或numastat)
        // 这里只是一个简化示例，实际生产环境可能需要更精确的OS统计
        info.total_memory_mb = numa_node_size(i, NULL) / (1024 * 1024);
        info.free_memory_mb = numa_node_size(i, &info.total_memory_mb) / (1024 * 1024); // 获取free memory

        topology.push_back(info);
    }
    return topology;
}

// 示例调用
/*
int main() {
    auto topology = get_numa_topology();
    for (const auto& node : topology) {
        std::cout << "Node " << node.node_id << ": " << std::endl;
        std::cout << "  CPUs: ";
        for (int cpu : node.cpus) {
            std::cout << cpu << " ";
        }
        std::cout << std::endl;
        std::cout << "  Total Memory: " << node.total_memory_mb << " MB" << std::endl;
        std::cout << "  Free Memory: " << node.free_memory_mb << " MB" << std::endl;
    }
    return 0;
}
*/

策略一：静态数据与任务绑定

概念：在数据块被创建、加载或首次缓存时，就将其显式地分配到特定的 NUMA 节点。同时，所有需要访问该数据块的计算任务（线程）也被调度或绑定到同一个 NUMA 节点上的 CPU 核心。这种策略适用于数据访问模式相对稳定、读密集型的场景。

适用场景：

  • 存储系统中的不可变数据块。
  • 静态配置文件或元数据。
  • 持久化缓存，数据一旦加载就长时间驻留。

实现细节：

  1. 数据结构设计：为每个数据块（例如 Chunk 或 Block）添加一个字段来记录其所在的 NUMA 节点 ID。
  2. 内存分配：在数据块加载到内存或创建时，使用 numa_alloc_onnode() 在指定的 NUMA 节点上分配内存。
  3. 线程绑定：为每个 NUMA 节点维护一个独立的工作线程池。当需要处理某个数据块时，将请求分发给该数据块所在 NUMA 节点的工作线程池。线程在启动时或任务执行前，通过 pthread_setaffinity_np() 绑定到该节点的 CPU 核心。

C++ 代码示例：NUMA 感知的数据块管理器

#include <iostream>
#include <vector>
#include <numeric>
#include <thread>
#include <chrono>
#include <stdexcept>
#include <cstring>
#include <numa.h>
#include <sched.h>
#include <atomic>
#include <future>
#include <queue>
#include <mutex>
#include <condition_variable>

// (前述的 check_numa_support, bind_thread_to_cpu, get_current_numa_node, get_numa_topology 函数略，假设已包含)

// 简单的 Chunk 结构
struct DataChunk {
    int chunk_id;
    int numa_node_id;
    size_t size_bytes;
    int* data; // 实际数据指针

    DataChunk(int id, int node_id, size_t s_bytes) 
        : chunk_id(id), numa_node_id(node_id), size_bytes(s_bytes), data(nullptr) {}

    // 构造函数，在指定NUMA节点分配内存
    void allocate() {
        data = static_cast<int*>(numa_alloc_onnode(size_bytes, numa_node_id));
        if (data == nullptr) {
            throw std::bad_alloc();
        }
        memset(data, 0, size_bytes); // 首次写入，确保分配到指定节点
        std::cout << "Chunk " << chunk_id << " allocated " << size_bytes 
                  << " bytes on NUMA node " << numa_node_id << " at " << data << std::endl;
        std::iota(data, data + (size_bytes / sizeof(int)), chunk_id * 1000); // 填充数据
    }

    // 析构函数，释放内存
    ~DataChunk() {
        if (data) {
            numa_free(data, size_bytes);
            std::cout << "Chunk " << chunk_id << " freed memory at " << data << std::endl;
        }
    }

    // 禁止拷贝，因为管理的是原始指针
    DataChunk(const DataChunk&) = delete;
    DataChunk& operator=(const DataChunk&) = delete;
    // 允许移动
    DataChunk(DataChunk&& other) noexcept
        : chunk_id(other.chunk_id), numa_node_id(other.numa_node_id),
          size_bytes(other.size_bytes), data(other.data) {
        other.data = nullptr; // 防止二次释放
    }
    DataChunk& operator=(DataChunk&& other) noexcept {
        if (this != &other) {
            if (data) numa_free(data, size_bytes);
            chunk_id = other.chunk_id;
            numa_node_id = other.numa_node_id;
            size_bytes = other.size_bytes;
            data = other.data;
            other.data = nullptr;
        }
        return *this;
    }
};

// NUMA感知的线程池，每个NUMA节点一个线程池
class NumaThreadPool {
public:
    NumaThreadPool(int node_id, const std::vector<int>& cpus, int num_threads = 4)
        : node_id_(node_id), cpus_(cpus), stop_(false) {
        if (cpus_.empty()) {
            std::cerr << "Warning: NUMA node " << node_id << " has no CPUs assigned. Using default." << std::endl;
            // 如果没有指定CPU，这里可以根据node_id获取
            // 简单起见，这里假设至少有一个CPU
            if (num_threads > 0) {
                 workers_.emplace_back([this, cpu_id = node_id * 12]() { // 假设每个node有12个cpu, cpu_id是node_id*12的倍数
                    this->worker_loop(cpu_id);
                });
            }
        } else {
            for (int i = 0; i < num_threads; ++i) {
                // 轮流绑定到该NUMA节点下的CPU
                int cpu_to_bind = cpus_[i % cpus_.size()];
                workers_.emplace_back([this, cpu_to_bind]() {
                    this->worker_loop(cpu_to_bind);
                });
            }
        }
        std::cout << "NumaThreadPool for node " << node_id << " created with " 
                  << workers_.size() << " threads." << std::endl;
    }

    ~NumaThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
            stop_ = true;
        }
        condition_.notify_all();
        for (std::thread& worker : workers_) {
            worker.join();
        }
        std::cout << "NumaThreadPool for node " << node_id_ << " destroyed." << std::endl;
    }

    template<class F, class... Args>
    auto enqueue(F&& f, Args&&... args) 
        -> std::future<typename std::result_of<F(Args...)>::type> {
        using return_type = typename std::result_of<F(Args...)>::type;

        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );

        std::future<return_type> res = task->get_future();
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
            if (stop_)
                throw std::runtime_error("enqueue on stopped NumaThreadPool");
            tasks_.emplace([task]() { (*task)(); });
        }
        condition_.notify_one();
        return res;
    }

private:
    void worker_loop(int cpu_id) {
        bind_thread_to_cpu(cpu_id);
        int current_node = get_current_numa_node();
        std::cout << "Worker thread " << std::this_thread::get_id() << " on CPU " << cpu_id 
                  << " (NUMA node " << current_node << ") is ready." << std::endl;

        while (true) {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lock(queue_mutex_);
                condition_.wait(lock, [this]{ return stop_ || !tasks_.empty(); });
                if (stop_ && tasks_.empty())
                    return;
                task = std::move(tasks_.front());
                tasks_.pop();
            }
            task();
        }
    }

    int node_id_;
    std::vector<int> cpus_;
    std::vector<std::thread> workers_;
    std::queue<std::function<void()>> tasks_;
    std::mutex queue_mutex_;
    std::condition_variable condition_;
    bool stop_;
};

// 模拟存储服务，管理数据块和NUMA线程池
class NumaAwareStorageService {
public:
    NumaAwareStorageService() {
        check_numa_support();
        auto topology = get_numa_topology();
        for (const auto& node_info : topology) {
            thread_pools_.emplace(node_info.node_id, 
                                  std::make_unique<NumaThreadPool>(node_info.node_id, node_info.cpus));
        }
    }

    // 创建并分配一个数据块
    std::shared_ptr<DataChunk> create_chunk(int chunk_id, size_t size_bytes, int target_node_id) {
        if (thread_pools_.find(target_node_id) == thread_pools_.end()) {
            std::cerr << "Error: Target NUMA node " << target_node_id << " not found." << std::endl;
            return nullptr;
        }
        auto chunk = std::make_shared<DataChunk>(chunk_id, target_node_id, size_bytes);
        chunk->allocate();
        chunks_[chunk_id] = chunk;
        return chunk;
    }

    // 模拟数据处理请求
    std::future<long long> process_chunk(int chunk_id) {
        auto it = chunks_.find(chunk_id);
        if (it == chunks_.end()) {
            throw std::runtime_error("Chunk not found: " + std::to_string(chunk_id));
        }
        std::shared_ptr<DataChunk> chunk = it->second;
        int target_node_id = chunk->numa_node_id;

        NumaThreadPool* pool = thread_pools_.at(target_node_id).get();
        return pool->enqueue([chunk]() {
            long long sum = 0;
            size_t num_elements = chunk->size_bytes / sizeof(int);
            for (size_t i = 0; i < num_elements; ++i) {
                sum += chunk->data[i];
            }
            std::cout << "Processing chunk " << chunk->chunk_id << " on NUMA node " 
                      << get_current_numa_node() << ". Sum: " << sum << std::endl;
            return sum;
        });
    }

private:
    std::map<int, std::unique_ptr<NumaThreadPool>> thread_pools_;
    std::map<int, std::shared_ptr<DataChunk>> chunks_; // 管理已分配的数据块
};

int main() {
    NumaAwareStorageService service;
    const size_t chunk_size = 1024 * 1024 * 10; // 10MB

    // 创建两个数据块，分别在不同NUMA节点
    auto chunk0 = service.create_chunk(100, chunk_size, 0);
    auto chunk1 = service.create_chunk(101, chunk_size, 1);

    // 提交处理请求，请求会被路由到数据块所在的NUMA节点线程池
    std::future<long long> res0 = service.process_chunk(100);
    std::future<long long> res1 = service.process_chunk(101);

    // 提交一个“错误”请求，让节点0的线程处理节点1的数据（模拟非NUMA感知）
    // 这需要修改enqueue的逻辑，或者直接创建另一个线程来模拟
    // 简化起见，这里仍然通过正确的路由来演示
    // 实际可以通过一个专门的“跨节点”线程池来模拟性能下降。

    res0.get(); // 等待并获取结果
    res1.get();

    std::cout << "All chunk processing finished." << std::endl;

    return 0;
}

编译指令:
g++ -std=c++17 -o numa_storage_service numa_storage_service.cpp -lnuma -lpthread
运行指令:
./numa_storage_service

这个示例展示了如何创建 NUMA 感知的数据块，并在其指定的 NUMA 节点上分配内存。然后，它维护了一个按 NUMA 节点划分的线程池集合，当请求处理某个数据块时，请求会被路由到该数据块所在 NUMA 节点上的线程池进行处理，确保了计算和数据的本地化。

策略二：动态请求路由与负载均衡

概念：对于更动态的分布式存储系统，数据块可能在不同的 NUMA 节点上被创建、修改或缓存。当一个外部请求（例如，读取某个文件的一部分）到达存储节点时，系统需要动态地判断所需数据块的 NUMA 节点位置，并将该请求路由到对应的 NUMA 节点上的工作线程进行处理。同时，为了避免某个 NUMA 节点过载，还需要考虑负载均衡。

适用场景：

  • 高并发读写混合工作负载。
  • 数据块位置可能动态变化的缓存系统。
  • 需要快速响应客户端请求的场景。

实现细节：

  1. 全局请求分发器：存储节点接收所有传入的请求。
  2. 数据块元数据：维护一个映射，记录每个数据块的 ID 及其当前所在的 NUMA 节点 ID。这个元数据需要高效查询。
  3. NUMA 节点工作队列：每个 NUMA 节点维护一个独立的请求队列和工作线程池。
  4. 路由逻辑：当请求到达分发器时，它会查询元数据获取数据块的 NUMA 节点 ID，然后将请求放入对应 NUMA 节点的工作队列。
  5. 负载均衡：如果某个 NUMA 节点上的请求队列过长或 CPU 利用率过高，分发器可以考虑将新请求（如果数据可以复制或迁移）路由到负载较低的 NUMA 节点，或者在允许的情况下，暂时将请求放到一个通用的“溢出”队列，由任何空闲线程处理（但会有远端访问开销）。

C++ 伪代码或设计思路：

// 假设有DataChunk结构，包含numa_node_id

// 全局请求分发器 (单例或通过DI注入)
class RequestDispatcher {
public:
    // ... 构造函数，初始化NumaThreadPools ...
    RequestDispatcher(const std::map<int, std::unique_ptr<NumaThreadPool>>& pools) : thread_pools_(pools) {}

    // 模拟数据块元数据存储
    std::map<int, int> chunk_numa_mapping; // chunk_id -> numa_node_id

    void register_chunk_location(int chunk_id, int numa_node_id) {
        chunk_numa_mapping[chunk_id] = numa_node_id;
    }

    // 处理传入的请求
    std::future<long long> handle_request(int chunk_id) {
        auto it = chunk_numa_mapping.find(chunk_id);
        if (it == chunk_numa_mapping.end()) {
            throw std::runtime_error("Chunk ID not found in mapping.");
        }
        int target_node_id = it->second;

        // 获取目标NUMA节点的线程池
        auto pool_it = thread_pools_.find(target_node_id);
        if (pool_it == thread_pools_.end()) {
            throw std::runtime_error("No thread pool for NUMA node " + std::to_string(target_node_id));
        }
        NumaThreadPool* target_pool = pool_it->second.get();

        // 将任务加入目标线程池
        // 假设我们有一个获取DataChunk实例的函数
        auto get_chunk_func = [this, chunk_id]() { return chunks_[chunk_id]; }; // 简化，实际需要安全获取

        return target_pool->enqueue([chunk_id, get_chunk_func]() {
            auto chunk = get_chunk_func(); // 获取DataChunk实例
            long long sum = 0;
            size_t num_elements = chunk->size_bytes / sizeof(int);
            for (size_t i = 0; i < num_elements; ++i) {
                sum += chunk->data[i];
            }
            std::cout << "Request for chunk " << chunk_id << " processed on NUMA node " 
                      << get_current_numa_node() << ". Result: " << sum << std::endl;
            return sum;
        });
    }

private:
    std::map<int, std::shared_ptr<DataChunk>> chunks_; // 简化，实际可能通过服务获取
    const std::map<int, std::unique_ptr<NumaThreadPool>>& thread_pools_; // 引用外部线程池
};

// Main函数中，NumaAwareStorageService可以内部包含RequestDispatcher
// 并负责管理chunks_和thread_pools_的生命周期。
// 外部请求通过调用 service.handle_request(chunk_id) 来触发

策略三：热点数据复制与缓存

概念：对于那些被频繁访问的“热点”数据块，即使它们最初被分配在某个特定的 NUMA 节点，也可以考虑将其复制到其他同样频繁访问它们的 NUMA 节点上的本地缓存中。这以内存消耗为代价，换取更低的访问延迟和更高的吞吐量。

适用场景：

  • 读多写少的场景。
  • 部分数据块访问频率远高于其他数据块。
  • 多 NUMA 节点上的不同工作负载需要访问相同热点数据。

实现细节：

  1. 热度追踪：实现一个监控机制，追踪每个数据块的访问频率、访问者（哪个 NUMA 节点上的线程）以及访问模式。这可能涉及到计数器、时间戳等。
  2. 动态复制策略：
    • 阈值判断：当一个数据块的访问频率超过某个阈值，并且有来自其他 NUMA 节点的频繁访问时，考虑将其标记为热点。
    • 副本创建：在需要访问该热点数据的其他 NUMA 节点上，使用 numa_alloc_onnode() 分配内存并创建数据块的副本。
    • 缓存替换策略：如果 NUMA 节点的本地缓存空间有限，需要实现 LRU、LFU 等缓存替换策略来管理副本。
  3. 缓存一致性：这是最复杂的环节。当原始数据块被修改时，其所有副本都必须被更新或失效。这需要引入分布式缓存一致性协议，例如基于版本号（MVCC）、两阶段提交（2PC）或消息队列通知。

C++ 伪代码或设计思路：

#include <map>
#include <mutex>
#include <atomic>
#include <list> // For LRU cache

// (假设 DataChunk 结构和 NumaAwareStorageService 已经存在)

struct CachedChunkInfo {
    std::shared_ptr<DataChunk> chunk;
    std::chrono::steady_clock::time_point last_access;
    std::atomic<long> access_count;
    // ... 其他元数据，如版本号 ...
};

// 每个NUMA节点一个本地缓存
class NumaLocalCache {
public:
    NumaLocalCache(int node_id, size_t max_size_bytes) 
        : node_id_(node_id), max_size_bytes_(max_size_bytes), current_size_bytes_(0) {}

    // 尝试从本地缓存获取数据块
    std::shared_ptr<DataChunk> get(int chunk_id) {
        std::lock_guard<std::mutex> lock(cache_mutex_);
        auto it = cache_map_.find(chunk_id);
        if (it != cache_map_.end()) {
            // 命中，更新访问信息 (LRU/LFU)
            it->second.access_count++;
            it->second.last_access = std::chrono::steady_clock::now();
            // 针对LRU，需要将元素移到列表前端
            // cache_list_.erase(it->second.list_it);
            // cache_list_.push_front(chunk_id);
            // it->second.list_it = cache_list_.begin();
            std::cout << "Chunk " << chunk_id << " HIT in NUMA node " << node_id_ << " local cache." << std::endl;
            return it->second.chunk;
        }
        std::cout << "Chunk " << chunk_id << " MISS in NUMA node " << node_id_ << " local cache." << std::endl;
        return nullptr;
    }

    // 将数据块添加到本地缓存 (通常是复制操作)
    void put(std::shared_ptr<DataChunk> original_chunk) {
        std::lock_guard<std::mutex> lock(cache_mutex_);
        if (cache_map_.count(original_chunk->chunk_id)) {
            return; // 已经存在
        }

        // 检查空间，如果不足则淘汰旧数据
        if (current_size_bytes_ + original_chunk->size_bytes > max_size_bytes_) {
            evict_oldest(); // 假设LRU淘汰
        }

        // 创建副本并分配到本地NUMA节点
        auto new_chunk = std::make_shared<DataChunk>(original_chunk->chunk_id, node_id_, original_chunk->size_bytes);
        new_chunk->allocate(); // 在当前NUMA节点分配
        memcpy(new_chunk->data, original_chunk->data, original_chunk->size_bytes); // 复制数据

        CachedChunkInfo info;
        info.chunk = new_chunk;
        info.last_access = std::chrono::steady_clock::now();
        info.access_count = 1;

        cache_map_[original_chunk->chunk_id] = info;
        current_size_bytes_ += original_chunk->size_bytes;
        std::cout << "Chunk " << original_chunk->chunk_id << " replicated to NUMA node " 
                  << node_id_ << " local cache." << std::endl;
    }

    // 淘汰最旧的/最不常用的数据 (LRU/LFU实现细节)
    void evict_oldest() {
        // 实际LRU需要list来追踪顺序，这里简化
        if (cache_map_.empty()) return;
        int chunk_to_evict = -1;
        std::chrono::steady_clock::time_point oldest_time = std::chrono::steady_clock::now();

        for (auto const& [id, info] : cache_map_) {
            if (info.last_access < oldest_time) {
                oldest_time = info.last_access;
                chunk_to_evict = id;
            }
        }

        if (chunk_to_evict != -1) {
            current_size_bytes_ -= cache_map_.at(chunk_to_evict).chunk->size_bytes;
            cache_map_.erase(chunk_to_evict);
            std::cout << "Evicted chunk " << chunk_to_evict << " from NUMA node " << node_id_ << " cache." << std::endl;
        }
    }

    // 通知缓存，某个原始数据块已被修改，需要失效或更新副本
    void invalidate_chunk(int chunk_id) {
        std::lock_guard<std::mutex> lock(cache_mutex_);
        auto it = cache_map_.find(chunk_id);
        if (it != cache_map_.end()) {
            current_size_bytes_ -= it->second.chunk->size_bytes;
            cache_map_.erase(it);
            std::cout << "Invalidated chunk " << chunk_id << " in NUMA node " << node_id_ << " local cache due to update." << std::endl;
        }
    }

private:
    int node_id_;
    size_t max_size_bytes_;
    size_t current_size_bytes_;
    std::map<int, CachedChunkInfo> cache_map_;
    std::mutex cache_mutex_;
    // std::list<int> cache_list_; // for LRU order
};

// NumaAwareStorageService中集成缓存管理
/*
class NumaAwareStorageService {
    // ... 其他成员 ...
    std::map<int, std::unique_ptr<NumaLocalCache>> local_caches_;

    void init_caches() {
        auto topology = get_numa_topology();
        for (const auto& node_info : topology) {
            local_caches_.emplace(node_info.node_id, 
                                  std::make_unique<NumaLocalCache>(node_info.node_id, node_info.total_memory_mb * 0.1 * 1024 * 1024)); // 10%内存作为缓存
        }
    }

    std::future<long long> process_chunk(int chunk_id) {
        auto it = chunks_.find(chunk_id);
        if (it == chunks_.end()) {
            throw std::runtime_error("Chunk not found: " + std::to_string(chunk_id));
        }
        std::shared_ptr<DataChunk> original_chunk = it->second;

        // 获取当前线程所在的NUMA节点
        int current_thread_node = get_current_numa_node();

        // 尝试从本地缓存获取
        std::shared_ptr<DataChunk> chunk_to_process = nullptr;
        if (local_caches_.count(current_thread_node)) {
            chunk_to_process = local_caches_.at(current_thread_node)->get(chunk_id);
        }

        if (!chunk_to_process) {
            // 如果本地缓存没有，则使用原始数据，并考虑是否复制到本地缓存
            chunk_to_process = original_chunk;
            // 热点检测和复制逻辑可以在这里或后台线程执行
            // 例如：如果original_chunk被current_thread_node频繁访问，则进行复制
            if (local_caches_.count(current_thread_node)) {
                // 异步或在条件允许时复制
                // local_caches_.at(current_thread_node)->put(original_chunk); 
            }
        }

        // 提交任务到处理数据的NUMA节点（本地缓存的节点或原始数据节点）
        int target_node_id = chunk_to_process->numa_node_id;
        NumaThreadPool* pool = thread_pools_.at(target_node_id).get();
        return pool->enqueue([chunk_to_process]() {
            long long sum = 0;
            size_t num_elements = chunk_to_process->size_bytes / sizeof(int);
            for (size_t i = 0; i < num_elements; ++i) {
                sum += chunk_to_process->data[i];
            }
            return sum;
        });
    }

    // 当原始数据块被修改时，通知所有本地缓存失效
    void update_chunk(int chunk_id, const void* new_data, size_t data_size) {
        // ... 更新原始数据 ...
        for (auto const& [node_id, cache] : local_caches_) {
            cache->invalidate_chunk(chunk_id);
        }
    }
};
*/

这种策略引入了更高的复杂性，特别是在维护缓存一致性方面。然而，对于某些工作负载，其性能收益可能非常显著。

挑战、权衡与未来展望

NUMA 感知编程并非没有代价，它带来了额外的复杂性和挑战：

  1. 复杂性增加：手动管理内存分配和线程亲和性，需要更深入地理解系统架构，增加了代码的复杂性和维护成本。
  2. 操作系统调度器干扰：操作系统本身的调度器可能不会总是完全遵守应用程序设定的亲和性，尤其是在系统负载很高时。
  3. 动态负载适应：静态绑定在负载模式动态变化的场景下可能不是最优解。需要动态调整策略，这进一步增加了复杂性。
  4. 伪共享（False Sharing）：如果不同 NUMA 节点上的线程访问位于同一缓存行但属于不同逻辑数据的数据，会导致缓存一致性协议频繁地在节点间同步缓存行，从而降低性能。需要精心设计数据结构以避免此问题。
  5. 缓存一致性开销：即使没有伪共享，跨 NUMA 节点共享数据仍然需要复杂的缓存一致性协议来维护数据视图的一致性，这会引入额外的总线流量和延迟。
  6. 内存碎片：在特定 NUMA 节点上进行频繁的内存分配和释放可能导致内存碎片，影响长期运行的稳定性。

权衡：
实现 NUMA 优化需要在性能提升、代码复杂度和资源消耗之间进行权衡。对于性能要求不高的系统，默认的“first-touch”行为可能已足够。但对于高性能分布式存储系统，NUMA 优化是不可或缺的。

未来展望：

  • 硬件发展：CXL (Compute Express Link) 等新型互联技术旨在提供更灵活、更高效的内存共享和设备互联方式，可能在未来简化 NUMA 编程。
  • 智能 OS 调度：操作系统调度器可能会变得更加 NUMA 感知，能够更智能地将进程和内存分配到合适的 NUMA 节点。
  • AI 辅助优化：通过机器学习技术分析程序的访问模式和 NUMA 拓扑，自动进行内存和线程的动态调度。

性能度量与调优

NUMA 优化效果的验证离不开精确的性能度量：

关键指标：

  • 吞吐量：每秒完成的请求数量。
  • 延迟：请求从发出到完成的时间。
  • CPU 利用率：各 NUMA 节点上 CPU 的利用情况，关注是否存在热点或空闲节点。
  • 内存带宽利用率：各 NUMA 节点内存控制器的带宽使用情况。
  • NUMA 远程访问计数：操作系统提供的统计数据，记录远程内存访问的次数。

常用工具：

  • numastat：查看每个 NUMA 节点的内存使用统计，包括本地和远程页面分配、页面迁移等。
  • perf：Linux 下强大的性能分析工具，可以跟踪 CPU 事件、缓存命中/未命中、NUMA 事件。例如，perf stat -e numa_hit,numa_miss ...。
  • sar -B：报告系统页错误和 NUMA 相关活动。
  • mpstat -N ALL：查看各个 NUMA 节点上 CPU 的利用率。
  • 自定义基准测试：模拟真实工作负载，对优化前后的系统进行对比测试。

调优策略：

  1. 逐步引入：不要一次性进行所有 NUMA 优化。从小范围开始，逐步引入并测量效果。
  2. 持续监控：在生产环境中持续监控 NUMA 相关的性能指标，及时发现并解决问题。
  3. 调整参数：根据实际负载调整线程池大小、缓存策略、数据分发粒度等参数。
  4. 避免过度优化：有时，为了微小的性能提升而引入巨大的复杂性是不值得的。找到最佳的平衡点。

迈向高效能分布式存储

通过对 NUMA 拓扑的深刻理解和 C++ 语言的精细控制能力，我们能够在分布式存储系统的单个节点内部实现数据与计算的紧密对齐，从而显著降低内存访问延迟，提高内存带宽利用率，并最终提升整个系统的吞吐量和响应速度。虽然 NUMA 感知编程增加了系统设计的复杂性，但对于追求极致性能的分布式存储系统而言，它无疑是一项值得投入的策略。通过精心设计的调度算法和底层优化，我们可以构建出更高效、更具竞争力的存储解决方案。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注