实例

在对DPDK的原理和代码展开进一步解析之前，先看一些小而简单的例子，建立一个形象上的认知。

1. helloworld
   • 特点：最基础的入门程序，代码简短简单。
   • 功能：多核（线程）环境下每个线程打印
   • 启动基础运行环境，DPDK构建了一个基于操作系统的，但适合包处理的软件运行环境，你可以认为这是个mini-OS。
     最早期DPDK，可以完全运行在没有操作系统的物理核（bare-metal）上，这部分代码现在不在主流的开源包中。
2. skeleton
   • 特点：一个最简单的单核报文收发示例
   • 功能：对收入报文不做任何处理直接发送
   • 也许这是当前世界上运行最快的报文进出测试程序。
3. l3fwd
   • 特点：DPDK中最流行的例子，也是发布DPDK性能测试的例子
   • 功能：三层转发是DPDK用于发布性能测试指标的主要应用

HelloWorld

功能

DPDK里的HelloWorld是最基础的入门程序，代码简短，功能也不复杂。

它建立了一个多核（线程）运行的基础环境，每个线程会打印 hello from core#，core#是由操作系统管理的。
如无特别说明，本文里的DPDK线程与硬件线程是一一对应的关系。

代码

main()

int main(int argc, char **argv)
{
    unsigned lcore_id;
    
    // 初始化
    int ret = rte_eal_init(argc, argv);
    if (ret < 0)
        rte_panic("Cannot init EAL\n");
	}

    // 在每个从lcore上调用lcore hello()
    // call lcore_hello() on every slave lcore
    RTE_LCORE_FOREACH_SLAVE(lcore_id) {	
        rte_eal_remote_launch(lcore_hello, NULL, lcore_id);
    }
    
    // 调用主lcore
    // call it on master lcore too
    lcore_hello(NULL);
    rte_eal_mp_wait_lcore();
    return 0;
}

代码说明

• eal：environment abstraction layer，环境抽象层
• rte：runtime environment，运行时环境
  • DPDK的主要对外函数接口都以rte_作为前缀
• lcore：logi core，逻辑核

抽象化函数接口是典型软件设计思路，可以帮助DPDK运行在多个操作系统上，DPDK官方支持Linux与FreeBSD。和多数并行处理系统类似，DPDK也有主线程、从线程的差异。

初始化基础运行环境 rte_eal_init

主线程运行入口是main函数，调用了rte_eal_init入口函数，启动基础运行环境。

int rte_eal_init(int argc, char **argv);

入口参数是启动DPDK的命令行，可以是长长的一串很复杂的设置，需要深入了解的读者可以查看DPDK相关的文档与源代码\lib\librte_eal\common\eal_common_options.c。

对于HelloWorld这个实例，最需要的参数是 -c<core mask>，线程掩码（core mask）指定了需要参与运行的线程（核）集合。

rte_eal_init 本身所完成的工作很复杂，它读取入口参数，解析并保存作为DPDK运行的系统信息，依赖这些信息，构建一个针对包处理设计的运行环境。

主要动作分解如下：

1. 配置初始化
2. 内存初始化
3. 内存池初始化
4. 队列初始化
5. 告警初始化
6. 中断初始化
7. PCI初始化
8. 定时器初始化
9. 检测内存本地化（NUMA）
10. 插件初始化
11. 主线程初始化
12. 轮询设备初始化
13. 建立主从线程通道
14. 将从线程设置在等待模式
15. PCI设备的探测与初始化

对于DPDK库的使用者，这些操作已经被EAL封装起来，接口清晰。如果需要对DPDK进行深度定制，二次开发，需要仔细研究内部操作，这里不做详解。

rte_eal_remote_launch() 设置亲核性

多核运行初始化

DPDK面向多核设计，程序会试图独占运行在逻辑核（lcore）上。main函数里重要的是启动多核运行环境，RTE_LCORE_FOREACH_SLAVE(lcore_id) 如名所示，遍历所有EAL指定可以使用的lcore，然后通过rte_eal_remote_launch在每个lcore上，启动被指定的线程。

int rte_eal_remote_launch(
    int (*f)(void *),
    void *arg,
    unsigned slave_id
);

// 参数lcore_id指定了从线程ID，运行入口函数lcore_hello。
/**
 * 第一个参数是从线程，是被征召的线程；
 * 第二个参数是传给从线程的参数；
 * 第三个参数是指定的逻辑核，从线程会执行在这个core上。
 */
int rte_eal_remote_launch(lcore_hello, NULL, lcore_id);

lcore_hello() 线程函数

运行函数 lcore_hello，它读取自己的逻辑核编号（lcore_id），打印出 “hello from core#”

static int lcore_hello(__attribute__((unused)) void *arg)
{
    unsigned lcore_id;
    lcore_id = rte_lcore_id();
    printf("hello from core %u\n", lcore_id);
    return 0;
}

这是个简单示例，从线程很快就完成了指定工作，在更真实的场景里，这个从线程会是一个循环运行的处理过程。

Skeleton

功能

DPDK为多核设计，但这是单核实例，设计初衷是实现一个最简单的报文收发示例，对收入报文不做任何处理直接发送。整个代码非常精简，可以用于平台的单核报文出入性能测试。

代码

main()

主要处理函数main的处理逻辑如下（伪码），调用 rte_eal_init 初始化运行环境，检查网络接口数，据此分配内存池 rte_pktmbuf_pool_create，入口参数是指定 rte_socket_id()，考虑了本地内存使用的范例。调用 port_init(portid, mbuf_pool) 初始化网口的配置，最后调用 lcore_main() 进行主处理流程。

int main(int argc, char *argv[])
{
    struct rte_mempool *mbuf_pool; 	// 内存池
    unsigned nb_ports;				// 端口数
    
    // 初始化环境抽象层(EAL)
    // Initialize the Environment Abstraction Layer (EAL)
    int ret = rte_eal_init(argc, argv);
    
    // 检查是否有偶数个端口可以发送/接收
    // Check that there is an even number of ports t send/receive on
    nb_ports = rte_eth_dev_count();
    if (nb_ports < 2 || (nb_ports & 1))
        rte_exit(EXIT_FAILURE, "Error: number of ports must be even\n");
    
    // 在内存中创建一个新的内存池来保存mbufs
    // Creates a new mempool in memory to hold the mbufs
    mbuf_pool = rte_pktmbuf_pool_create("MBUF_POOL", NUM_MBUFS * nb_ports,
        MBUF_CACHE_SIZE, 0, RTE_MBUF_DEFAULT_BUF_SIZE, rte_socket_id());
    
    // 初始化所有端口
    // Initialize all ports
    uint8_t portid;
    for (portid = 0; portid < nb_ports; portid++)
        if (port_init(portid, mbuf_pool) != 0)
            rte_exit(EXIT_FALURE, "Cannot init port %"PRIu8 "\n",
                     portid);
    
    // 只在主核上调用lcore main
    // Call lcore_main on the master core only
    lcore_main();
    return 0;
}

// 网口初始化流程
port_init(uint8_t port, struct rte_mempool *mbuf_pool) {...}

首先对指定端口设置队列数，基于简单原则，本例只指定单队列。在收发两个方向上，基于端口与队列进行配置设置，缓冲区进行关联设置。如不指定配置信息，则使用默认配置。

port_init() 网口初始化

网口设置：对指定端口设置接收、发送方向的队列数目，依据配置信息来指定端口功能

int rte_eth_dev_configure(
    uint8_t port_id,
    uint16_t nb_rx_q, 
    uint16_t nb_tx_q, 
    const struct rte_eth_conf *dev_conf
)

队列初始化：对指定端口的某个队列，指定内存、描述符数量、报文缓冲区，并且对队列进行配置

int rte_eth_rx_queue_setup(uint8_t port_id, uint16_t rx_queue_id,
              uint16_t nb_rx_desc, unsigned int socket_id,
              const struct rte_eth_rxconf *rx_conf,
              struct rte_mempool *mp);

int rte_eth_tx_queue_setup(uint8_t port_id, uint16_t tx_queue_id,
              uint16_t nb_tx_desc, unsigned int socket_id,
              const struct rte_eth_txconf *tx_conf);

网口设置：初始化配置结束后，启动端口int rte_eth_dev_start（uint8_t port_id）；

完成后，读取MAC地址，打开网卡的混杂模式设置，允许所有报文进入。

static inline int port_init(uint8_t port, struct rte_mempool *mbuf_pool)
{
    struct rte_eth_conf port_conf = port_conf_default;
    const uint16_t rx_rings = 1, tx_rings = 1;
    // Configure the Ethernet device
    retval = rte_eth_dev_configure(port, rx_rings, tx_rings, &port_conf);
    // Allocate and set up 1 RX queue per Ethernet port
    for (q = 0; q < rx_rings; q++) {
        retval = rte_eth_rx_queue_setup(port, q, RX_RING_SIZE,
                rte_eth_dev_socket_id(port), NULL, mbuf_pool);
    }
    // Allocate and set up 1 TX queue per Ethernet port
    for (q = 0; q < tx_rings; q++) {
        retval = rte_eth_tx_queue_setup(port, q, TX_RING_SIZE,
                rte_eth_dev_socket_id(port), NULL);
    }
    // Start the Ethernet port
    retval = rte_eth_dev_start(port);
    // Display the port MAC address
    struct ether_addr addr;
    rte_eth_macaddr_get(port, &addr);
    // Enable RX in promiscuous mode for the Ethernet device
    rte_eth_promiscuous_enable(port);
    return 0;
}

网口收发报文循环收发在lcore_main中有个简单实现，因为是示例，为保证性能，首先检测CPU与网卡的Socket是否最优适配，建议使用本地CPU就近操作网卡，后续章节有详细说明。数据收发循环非常简单，为高速报文进出定义了burst的收发函数如下，4个参数意义非常直观：端口，队列，报文缓冲区以及收发包数。

rte_eth_[rx/tx]_burst() 收发函数

基于端口队列的报文收发函数：

static inline uint16_t rte_eth_rx_burst(
    uint8_t port_id, uint16_t queue_id, 
	struct rte_mbuf **rx_pkts, const uint16_t nb_pkts
)

static inline uint16_t rte_eth_tx_burst(
    uint8_t port_id, uint16_t queue_id, 
	struct rte_mbuf **tx_pkts, uint16_t nb_pkts
)

这就构成了最基本的DPDK报文收发展示。可以看到，此处不涉及任何具体网卡形态，软件接口对硬件没有依赖。

static __attribute__((noreturn)) void lcore_main(void)
{
    const uint8_t nb_ports = rte_eth_dev_count();
    uint8_t port;
    for (port = 0; port < nb_ports; port++)
        if (rte_eth_dev_socket_id(port) > 0 &&
                rte_eth_dev_socket_id(port) !=
                        (int)rte_socket_id())
            printf("WARNING, port %u is on remote NUMA node to "
                    "polling thread.\n\tPerformance will "
                    "not be optimal.\n", port);
    // Run until the application is quit or killed
    for (;;) {
        /*
         * Receive packets on a port and forward them on the paired
         * port. The mapping is 0 -> 1, 1 -> 0, 2 -> 3, 3 -> 2, etc.
         */
        for (port = 0; port < nb_ports; port++) {
            // Get burst of RX packes, from first port of pair
            struct rte_mbuf *bufs[BURST_SIZE];
            const uint16_t nb_rx = rte_eth_rx_burst(port, 0,
                    bufs, BURST_SIZE);
            if (unlikely(nb_rx == 0))
                continue;
            // Send burst of TX packets, to second port of pair
            const uint16_t nb_tx = rte_eth_tx_burst(port ^ 1, 0,
                    bufs, nb_rx);
            // Free any unsent packets
            if (unlikely(nb_tx < nb_rx)) {
                uint16_t buf;
                for (buf = nb_tx; buf < nb_rx; buf++)
                    rte_pktmbuf_free(bufs[buf]);   
            }
        }
    }
}

L3fwd

这是DPDK中最流行的例子，也是发布DPDK性能测试的例子。

如果将PCIE插槽上填满高速网卡，将网口与大流量测试仪表连接，它能展示在双路服务器平台具备 200Gbit/s 的转发能力。数据包被收入系统后，会查询IP报文头部，依据目标地址进行路由查找，发现目的端口，修改IP头部后，将报文从目的端口送出。路由查找有两种方式，一种方式是基于目标IP地址的完全匹配（exact match），另一种方式是基于路由表的最长掩码匹配（Longest Prefix Match，LPM）。三层转发的实例代码文件有2700多行（含空行与注释行），整体逻辑其实很简单，是前续 HelloWorld 与 Skeleton 的结合体。

启动参数

启动这个例子，指定命令参数格式如下：

./build/l3fwd [EAL options] -- -p PORTMASK [-P] 

--config(port,queue,lcore)[,(port,queue,lcore)]

命令参数分为 两个部分，以 “--” 为分界线：

• 分界线左边：是DPDK的EAL Options
  • [EAL Options] 是DPDK运行环境的输入配置选项，输入命令会交给 rte_eal_init 处理；
• 分界线右边：是三层转发的私有命令选项
  • PORTMASK 依据掩码选择端口，DPDK启动时会搜索系统认识的PCIe设备，依据黑白名单原则来决定是否接管，早期版本可能会接管所有端口，断开网络连接。现在可以通过脚本绑定端口，具体可以参见 http://www.dpdk.org/browse/dpdk/tree/tools/dpdk_nic_bind.py
  • config 选项指定（port，queue，lcore），用指定线程处理对应的端口的队列。要实现200Gbit/s的转发，需要大量线程（核）参与，并行转发

先来看主线程流程main的处理流程，因为和HelloWorld与Skeleton类似，不详细叙述。

// 初始化运行环境
rte_eal_init(argc, argv);

// 分析入参
parse_args(argc, argv);

// 初始化 lcore 与 port 配置端口与队列初始化，类似 Skeleton 处理端口启动，使能混杂模式启动从线程，令其运行
main_loop()

从线程执行main_loop（）的主要步骤如下：

// 读取自己的 lcore 信息完成配置;
// 读取关联的接收与发送队列信息;
// 进入循环处理
{
    // 向指定队列批量发送报文;
    // 从指定队列批量接收报文;
    // 批量转发接收到报文;
}

向指定队列批量发送报文，从指定队列批量接收报文，此前已经介绍了DPDK的收发函数。批量转发接收到的报文是处理的主体，提供了基于Hash的完全匹配转发，也可以基于最长匹配原则（LPM）进行转发。转发路由查找方式可以由编译配置选择。除了路由转发算法的差异，下面的例子还包括基于 multi buffer 原理的代码实现。在 #if(ENABLE_MULTI_BUFFER_OPTIMIZE==1) 的路径下，一次处理8个报文。

和普通的软件编程不同，初次见到的程序员会觉得奇怪。它的实现有效利用了处理器内部的乱序执行和并行处理能力，能显著提高转发性能。

for (j = 0; j < n; j += 8) {
    uint32_t pkt_type =
        pkts_burst[j]->packet_type &
        pkts_burst[j+1]->packet_type &
        pkts_burst[j+2]->packet_type &
        pkts_burst[j+3]->packet_type &
        pkts_burst[j+4]->packet_type &
        pkts_burst[j+5]->packet_type &
        pkts_burst[j+6]->packet_type &
        pkts_burst[j+7]->packet_type;
    
    if (pkt_type & RTE_PTYPE_L3_IPV4) {
        simple_ipv4_fwd_8pkts(&pkts_burst[], portid, qconf);
 	} else if (pkt_type & RTE_PTYPE_L3_IPV6) {
        simple_ipv6_fwd_8pkts(&pkts_burst[j], portid, qconf);
    } else {
        l3fwd_simple_forward(pkts_burst[j],portid, qconf);
        l3fwd_simple_forward(pkts_burst[j+1],portid, qconf);
        l3fwd_simple_forward(pkts_burst[j+2],portid, qconf);
        l3fwd_simple_forward(pkts_burst[j+3],portid, qconf);
        l3fwd_simple_forward(pkts_burst[j+4],portid, qconf);
        l3fwd_simple_forward(pkts_burst[j+5],portid, qconf);
        l3fwd_simple_forward(pkts_burst[j+6],portid, qconf);
        l3fwd_simple_forward(pkts_burst[j+7],portid, qconf);
        }
    }

    for (; j < nb_rx ; j++) {
        l3fwd_simple_forward(pkts_burst[j],portid, qconf);
    }
}

依据IP头部的五元组信息，利用rte_hash_lookup来查询目标端口。

mask0 = _mm_set_epi32(ALL_32_BITS, ALL_32_BITS, ALL_32_BITS, BIT_8_TO_15);

ipv4_hdr = (uint8_t *)ipv4_hdr + offsetof(struct ipv4_hdr, time_to_live);

__m128i data = _mm_loadu_si128((__m128i*)(ipv4_hdr));

// 获取5元组:dst端口，src端口，dst IP地址，src IP地址和协议
// Get 5 tuple: dst port, src port, dst IP address, src IP address and protocol
key.xmm = _mm_and_si128(data, mask0);

// 查找目的端口
// Find destination port
ret = rte_hash_lookup(ipv4_l3fwd_lookup_struct, (const void *)&key);

return (uint8_t)((ret < 0) portid : ipv4_l3fwd_out_if[ret]);

这段代码在读取报文头部信息时，将整个头部导入了基于SSE的矢量寄存器（128位宽），并对内部进行了掩码mask0运算，得到key，然后把key作为入口参数送入rte_hash_lookup运算。同样的操作运算还展示在对IPv6的处理上，可以在代码中参考。

本书偏重原理而非代码

我们并不计划在本节将读者带入代码陷阱中，实际上本书总体上也没有偏重代码讲解，而是在原理上进行解析。如果读者希望了解详细完整的编程指南，可以参考DPDK的网站。