【DPDK学习路径】六、申请缓冲区内存池

        节5中展示了如何使用DPDK提供的运行时接口创建线程并绑定核心，创建线程是为了执行确定的任务，对于DPDK而言，最重要的任务就是处理网卡接收到的数据包报文。

        Linux 内核协议栈与网卡之间的工作是非常经典的生产者-消费者模型，在接收报文的情况下，网卡总是生产者，而协议栈总是消费者，发送的时候则是相反的。DPDK 想要取代内核协议栈，就必须要完成对这一生产-消费关系的处理，具体来讲，DPDK 使用了无锁循环队列来作为数据包缓冲区。

        队列的组织依赖于存储数据报文的结构体，在内核协议栈中，这一结构体是sk_buff，而在DPDK中，这一结构体是 mbuf。简单来说，内核使用 sk_buff 保存接收的报文，通过多个 sk_buff 组合成为接收队列，而在 DPDK 中，使用 mbuf替代 sk_buff。

        DPDK 提供了一次性创建多个 mbuf的接口 rte_pktmbuf_pool_create，这样一次性创建出来的多个 m_buff 称为一个缓冲区内存池。对上层而言，直接调用此接口创建内存池，使用内存池创建一个接收队列，将队列绑定网卡，即可轮询从队列中取得数据包并保存于内存池之中。

        rte_pktmbuf_pool_create 含有六个参数，分别是内存池名称、mbuf数量、缓存大小、私有空间大小、单个 mbuf 大小、socket id。

        这里的 socket id 是对 NUMA 的支持，与 TCP/IP 的 socket 无关，在后面的章节将详细介绍 NUMA 支持。

        下面给出 rte_pktmbuf_pool_create 的具体实例：

#include 
#include 
#include 
int main(int argc, char *argv[]) {
    char buf_name[64] = {0};
    struct rte_mempool *mbuf_pool;
    int socketid = 0;
    
    if(rte_eal_init(argc, argv) name);
    // 创建及修改mbuf
    struct rte_mbuf *mbuf;
    char *pkt_start;
    char data[64] = "hello world!";
    int data_len = strlen(data);
    
    mbuf = rte_pktmbuf_alloc(mbuf_pool);
    if (mbuf == NULL) {
        rte_exit(EXIT_FAILURE, "Cannot allocate mbuf\n");
    }
    // mbuf->buf_addr指向了实际的内存缓冲区，mbuf->data_off则表示缓冲区的偏移量
    // 因此，缓冲区指针+偏移量，指向了数据包真实的起点。
    pkt_start = (char *)(mbuf->buf_addr) + mbuf->data_off;
    rte_memcpy(pkt_start, data, data_len);
    printf("%s\n", pkt_start);
    // free
    rte_pktmbuf_free(mbuf);
    rte_mempool_free(mbuf_pool);
    rte_eal_cleanup();
    
    return 0;
}

        在上述实例中我们默认 socket id 是 0，这种假设在 non-NUMA 架构的 CPU 中是合适的，但是，在 NUMA 架构中，实际会导致多核性能下降，这是因为 DPDK 支持 NUMA 架构中的核心优先访问本地内存，而如果默认 socket id 为 0，那么，使用其他核心执行任务时也只能访问 core 0 的内存，也就是访问了远程内存，这会导致访存时间边长。

        因此，在 NUMA 架构下有必要充分使用 DPDK 对 NUMA 的支持创建内存池，下面给出具体的实例：

#define NB_SOCKETS 10
#define MEMPOOL_CACHE_SIZE 250
static int numa_on = 0;
static struct rte_mempool * pktmbuf_pool[NB_SOCKETS];
static int
init_mem(unsigned nb_mbuf){
    unsigned lcore_id;
    int socketid;
    char mbuf_pool_name[64];
    for(lcore_id = 0; lcore_id = NB_SOCKETS){
            rte_exit(EXIT_FAILURE, "Socket %d of lcore %u is out of range %d\n",
                socketid, lcore_id, NB_SOCKETS);
        }
        if(pktmbuf_pool[socketid] == NULL){
            snprintf(mbuf_pool_name, sizeof(mbuf_pool_name), "mbuf_pool_%d", socketid);
            pktmbuf_pool[socketid] = rte_pktmbuf_pool_create(mbuf_pool_name, nb_mbuf,
                MEMPOOL_CACHE_SIZE, 0, RTE_MBUF_DEFAULT_BUF_SIZE, socketid);
            if(pktmbuf_pool[socketid] == NULL){
                rte_exit(EXIT_FAILURE, "Cannot init mbuf pool on socket %d\n", socketid);
            }
            else{
                printf("Allocated mbuf pool on socket %d\n", socketid);
            }
        }
    }
    return 0;
}

        尝试将此函数添加到你的 DPDK 程序中，为各个 socket 创建相应的内存池吧。（对于 non-NUMA 架构，只能为 socket 0 创建内存池）

        下一节我们将讲述如何使用内存池缓冲区创建网卡的 接收/发送队列。