节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 <string.h>#include <rte_eal.h>
#include <rte_mbuf.h>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) < 0){rte_exit(EXIT_FAILURE, "Error with eal init\n");}snprintf(buf_name, sizeof(buf_name)-1, "buf_1");// 创建 mbuf poolmbuf_pool = rte_pktmbuf_pool_create(buf_name, 10, 250, 0, 2048+128, socketid);if (mbuf_pool == NULL){rte_exit(EXIT_FAILURE, "Cannot init mbuf pool on socket %d\n", socketid);}else{printf("Allocated mbuf pool on socket %d\n", socketid);}printf("alloc mempool name : [%s]\n", mbuf_pool->name);// 创建及修改mbufstruct 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);// freerte_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 < RTE_MAX_LCORE; lcore_id++){if (rte_lcore_is_enabled(lcore_id) == 0){continue;}if(numa_on){socketid = rte_lcore_to_socket_id(lcore_id);}else{socketid = 0;}if(socketid >= 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 创建内存池）

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