DPDK示例指南（翻译）（ 1～7）

2017-10-30 本文已影响976人半天妖

1. 简介

本文档描述了数据平面开发工具包（DPDK）中包含的示例应用程序。每章描述了一个示例应用程序，它展示了具体的功能，并提供了有关如何编译，运行和使用示例应用程序的说明。

1.1. 文档路线图

以下是一份建议顺序阅读的DPDK参考文档列表:

发行公告：提供特定发行版本的信息，包括支持的特性、限制条件、修复的问题、已知的问题等等。此外，还以FAQ的方式提供了常见问题的解决方法。
入门指南：介绍如何安装及配置DPDK软件，旨在帮助用户快速上手。
FreeBSD 入门指南：DPDK1.6.0发布版本之后添加了FreeBSD* 平台上的入门指南。有关如何在FreeBSD* 上安装配置DPDK，请参阅这个文档。
编程指南（本文档），描述了如下内容：
- 软件架构以及如何使用（示例介绍），特别是在Linux用户环境中的使用
- DPDK的主要内容，系统构建（包括可以在DPDK根目录Makefile中用来构建工具包和应用程序的命令）及应用移植细则。
- 软件中使用的，以及新开发中需要考虑的一些优化。
API参考：提供有关DPDK功能、数据结构和其他编程结构的详细信息。
示例程序用户指南：描述了一组例程。每个章节描述了一个用例，展示了具体的功能，并提供了有关编译、运行和使用的说明。

2. Command Line

本章介绍了作为数据平面开发工具包（DPDK）的一部分的命令行示例应用程序。

2.1. 概述

命令行示例应用程序是提供在DPDK中使用命令行界面的简单应用程序。该应用程序是一种类似于readline的接口，可用于在Linux *应用程序环境中调试DPDK应用程序。

注意：rte_cmdline库不应在生产代码中使用，因为没有按照其他DPDK库的相同标准验证过。
请参阅发行说明的“已知问题”部分中的“rte_cmdline库不应在生产代码中由于有限的测试使用”项目。

命令行示例应用程序支持GNU readline库的一些功能，例如，完成，剪切/粘贴和其他使配置和调试更快更容易的特殊绑定。

该应用程序显示如何扩展rte_cmdline应用程序来处理对象列表。有三个简单的命令：

add obj_name IP: Add a new object with an IP/IPv6 address associated to it
del obj_name: Delete the specified object.
show obj_name: Show the IP associated with the specified object.

注意：终止程序，使用ctrl+d

2.2. 编译程序

export RTE_SDK=/path/to/rte_sdk
cd ${RTE_SDK}/examples/cmdline

设置target，如：

export RTE_TARGET=x86_64-native-linuxapp-gcc

编译

make

2.3. 运行程序

使用以下命令运行应用程序

./build/cmdline -l 0-3 -n 4

可以参考DPDK入门指南获取参数信息及用法。
上面的参数为在逻辑核0～3上运行程序，每个处理器插槽上内存通道数目为4。

注意，必须先分配hugepage才能正确运行程序。

error.png

正确运行.png

2.4. 代码逻辑

第一个任务就是创建环境适配层，初始化环境：

int main(int argc, char **argv)
{
    int ret;
    struct cmdline *cl;

    ret = rte_eal_init(argc, argv);
    if (ret < 0)
        rte_panic("Cannot init EAL\n");

然后，创建一个新的命令行对象并开始通过控制台与用户进行交互：

    cl = cmdline_stdin_new(main_ctx, "example> ");
    if (cl == NULL)
        rte_panic("Cannot create cmdline instance\n");
    cmdline_interact(cl);
    cmdline_stdin_exit(cl);

    return 0;
}

cmd line_interact（）函数在用户键入Ctrl-d时返回，在这种情况下，应用程序退出。

一个命令行上下文就是列在非终止表中的命令，如下所示：

cmdline_parse_ctx_t main_ctx[] = {
   (cmdline_parse_inst_t *)&cmd_obj_del_show,
   (cmdline_parse_inst_t *)&cmd_obj_add,
   (cmdline_parse_inst_t *)&cmd_help,
   NULL,
};

每个命令都是静态定义的。它包含一个指针，指向命令解析时需要执行的函数，一个不透明指针，一个帮助字符串和参数列表。

rte_cmdline应用程序提供了一个预定义的参数类型列表：

String Token: Match a static string, a list of static strings or any string
Number Token: Match a number that can be signed or unsigned, from 8-bit to 32-bit.
IP Address Token: Match an IPv4 or IPv6 address or network.
Ethernet* Address Token: Match a MAC address.

在本例子中，新的对象类型及其实现在parse_obj_list.c及parse_obj_list.c中。

如，cmd_obj_del_show定义如下：

cmdline_parse_inst_t cmd_obj_del_show = {
    .f = cmd_obj_del_show_parsed,  /* function to call */
    .data = NULL,      /* 2nd arg of func */
    .help_str = "Show/del an object",
    .tokens = {        /* token list, NULL terminated */
        (void *)&cmd_obj_action,
        (void *)&cmd_obj_obj,
        NULL,
    },
};

struct cmd_obj_del_show_result {
    cmdline_fixed_string_t action;
    struct object *obj;
};

static void cmd_obj_del_show_parsed(void *parsed_result,
                    struct cmdline *cl,
                    __attribute__((unused)) void *data)
{
  ……
}

该命令有两个输入参数：

第一个表示是del还是show
第二个为obj，表示前面用add命令添加的对象
一旦命令被解析，rte_cmdline应用程序填充一个cmd_obj_del_show_result结构体。指向这个结构体的指针作为回调函数的参数。

3. Ethtool

Ethtool示例程序展示了类似ethtool的API实现，并提供了一个控制台环境，允许查询和更改以太网卡的参数。该示例程序基于简单的L2框架反射器。

3.1. 编译程序

3.2. 运行程序

3.3.代码逻辑

程序使用控制台来驱动，具体控制带命令的实现，可以参考第二章Command Line用例。

void ethapp_main(void)
{
    struct cmdline *ctx_cmdline;

    ctx_cmdline = cmdline_stdin_new(list_prompt_commands, "EthApp> ");
    cmdline_interact(ctx_cmdline);
    cmdline_stdin_exit(ctx_cmdline);
}

支持如下命令：

drvinfo：打印驱动信息
eeprom：将EEPROM输出到文件
link：打印端口link状态
macaddr：获取/设置MAC地址
mtu：设置NIC的MTU
open：开启端口
pause：获取/设置端口pause状态
portstats：打印端口统计信息
regs：打印端口寄存器到文件
ringparam：获取/设置ring参数
rxmode：反转端口RX模式
stop：停止端口
validate：检查给定的MAC地址是否合法单播地址
vlan：添加/移除VLAN ID
quit：退出程序

整个程序分成两部分，一部分是后台程序，packet reflector，运行于slave core；另一部分是前台的ethtool shell程序，运行于master core。

3.3.1. Packet Reflector

后台数据包反射器旨在演示由ethtool控制的基本数据包处理。每个传入的MAC帧被重写，使得它被传回发送端，使用接收端的MAC地址作为源地址，然后在相同的端口上发送出去。

3.3.2. Ethtool Shell

Ethtool示例的前台部分是基于Command Line的界面，可以接收程序中定义的命令。单个回调函数处理与每个命令相关联的详细信息，这些命令利用了ethtool接口中定义的功能和DPDK功能。

3.3.3. Ethtool实现的接口

rte_ethtool_get_drvinfo()
rte_ethtool_get_regs_len()
rte_ethtool_get_regs()
rte_ethtool_get_link()
rte_ethtool_get_eeprom_len()
rte_ethtool_get_eeprom()
rte_ethtool_set_eeprom()
rte_ethtool_get_pauseparam()
rte_ethtool_set_pauseparam()
rte_ethtool_net_open()
rte_ethtool_net_stop()
rte_ethtool_net_get_mac_addr()
rte_ethtool_net_set_mac_addr()
rte_ethtool_net_validate_addr()
rte_ethtool_net_change_mtu()
rte_ethtool_net_get_stats64()
rte_ethtool_net_vlan_rx_add_vid()
rte_ethtool_net_vlan_rx_kill_vid()
rte_ethtool_net_set_rx_mode()
rte_ethtool_get_ringparam()
rte_ethtool_set_ringparam()

4. Exception Path

Exception Path示例旨在展示如何使用DPDK设置数据包穿过Linux内核。这是通过虚拟的TAP网络接口实现的。可以由DPDK应用程序读取和写入，并作为标准网络接口呈现给内核。

4.1. 概述

应用程序为每个网卡接口创建了两个线程，一个线程从接口中读取并写数据到线程指定的TAP接口上。另一个线程从TAP接口读取并写数据到NIC端口上。

报文流的路径如下图所示：

报文路径

为了进行吞吐测量，必须设置内核网桥以实现在网桥之间转发数据。

4.2. 编译程序

4.3. 运行程序

按照如下命令运行程序：

.build/exception_path [EAL options] -- -p PORTMASK -i IN_CORES -o OUT_CORES

其中：
-p PORTMASK：端口掩码，表示使用的端口，十六进制表示
-i IN_CORES：从网卡中读数据的core掩码
-o OUT_CORES：向网卡中写数据的core掩码

注意，-c参数或-l参数指定的程序运行core列表需要包含IN_CORES和OUT_CORES。且IN_CORES和OUT_CORES中不能有相同的core。端口和core之间的亲和性从每个掩码的最低有效位开始设置，也就是说，由PORTMASK中的最低位表示的端口由IN_CORES中的最低位表示的core读取，并由core写入由OUT_CORES中的最低位表示。

示例：

./build/exception_path -l 0-3 -n 4 -- -p 3 -i 3 -o c

当程序运行时，报文收发统计信息可以通过发送SIGUSER1给程序来展示

killall -USR1 exception_path

通过发送SIGUSER2信号来重置统计信息。

4.4. 代码逻辑

4.4.1. 初始化

创建mbuf pool、驱动、队列与L2转发用例一样。此外，需要创建TAP接口。每个lcore上需要创建一个TAP接口。代码如下：

/*
 * Create a tap network interface, or use existing one with same name.
 * If name[0]='\0' then a name is automatically assigned and returned in name.
 */
static int tap_create(char *name)
{
    struct ifreq ifr;
    int fd, ret;

    fd = open("/dev/net/tun", O_RDWR);
    if (fd < 0)
        return fd;

    memset(&ifr, 0, sizeof(ifr));

    /* TAP device without packet information */
    ifr.ifr_flags = IFF_TAP | IFF_NO_PI;

    if (name && *name)
        snprintf(ifr.ifr_name, IFNAMSIZ, "%s", name);

    ret = ioctl(fd, TUNSETIFF, (void *) &ifr);
    if (ret < 0) {
        close(fd);
        return ret;
    }

    if (name)
        snprintf(name, IFNAMSIZ, "%s", ifr.ifr_name);

    return fd;
}

初始化程序的另一个操作是为每个port关联两个core：

一个用于从port读取并写数据到TAP接口
一个用于从TAP接口读取并写数据到port

这是通过port_ids[]数组实现的，该数据由lcore ID作为索引，具体操作代码如下：

/* Record affinities between ports and lcores in global port_ids[] array */
static void
setup_port_lcore_affinities(void)
{
    unsigned long i;
    uint8_t tx_port = 0;
    uint8_t rx_port = 0;

    /* Setup port_ids[] array, and check masks were ok */
    RTE_LCORE_FOREACH(i) {
        if (input_cores_mask & (1ULL << i)) {
            /* Skip ports that are not enabled */
            while ((ports_mask & (1 << rx_port)) == 0) {
                rx_port++;
                if (rx_port > (sizeof(ports_mask) * 8))
                    goto fail; /* not enough ports */
            }

            port_ids[i] = rx_port++;
        } else if (output_cores_mask & (1ULL << (i & 0x3f))) {
            /* Skip ports that are not enabled */
            while ((ports_mask & (1 << tx_port)) == 0) {
                tx_port++;
                if (tx_port > (sizeof(ports_mask) * 8))
                    goto fail; /* not enough ports */
            }

            port_ids[i] = tx_port++;
        }
    }

    if (rx_port != tx_port)
        goto fail; /* uneven number of cores in masks */

    if (ports_mask & (~((1 << rx_port) - 1)))
        goto fail; /* unused ports */

    return;
fail:
    FATAL_ERROR("Invalid core/port masks specified on command line");
}

4.4.2. 报文转发

初始化完成之后，每个lcore都会运行main_loop函数。该函数首先检查lcore_id及用户提供的input_core_mask和output_core_mask，以确定该core是contTAP读取还是想TAP写入数据。

对于从NIC port读取的情况，数据包的接收与L2FWD用例相同。通过使用适当的TAP接口的文件描述符调用write来完成数据包的传输，然后将mbuf释放会池中。

    /* Loop forever reading from NIC and writing to tap */
    for (;;) {
        struct rte_mbuf *pkts_burst[PKT_BURST_SZ];
        unsigned i;
        const unsigned nb_rx =
                rte_eth_rx_burst(port_ids[lcore_id], 0,
                    pkts_burst, PKT_BURST_SZ);
        lcore_stats[lcore_id].rx += nb_rx;
        for (i = 0; likely(i < nb_rx); i++) {
            struct rte_mbuf *m = pkts_burst[i];
            /* Ignore return val from write() */
            int ret = write(tap_fd,
                            rte_pktmbuf_mtod(m, void*),
                            rte_pktmbuf_data_len(m));
            rte_pktmbuf_free(m);
            if (unlikely(ret < 0))
                lcore_stats[lcore_id].dropped++;
            else
                lcore_stats[lcore_id].tx++;
        }
    }

反之，对于从TAP接口读取，并写数据到NIC port的core，通过条用read接口从TAP接口文件描述符中读取报文，填充到mbuf。代码逻辑如下：
/* Loop forever reading from tap and writing to NIC */
for (;;) {
int ret;
struct rte_mbuf *m = rte_pktmbuf_alloc(pktmbuf_pool);
if (m == NULL)
continue;

        ret = read(tap_fd, rte_pktmbuf_mtod(m, void *),
            MAX_PACKET_SZ);
        lcore_stats[lcore_id].rx++;
        if (unlikely(ret < 0)) {
            FATAL_ERROR("Reading from %s interface failed",
                        tap_name);
        }
        m->nb_segs = 1;
        m->next = NULL;
        m->pkt_len = (uint16_t)ret;
        m->data_len = (uint16_t)ret;
        ret = rte_eth_tx_burst(port_ids[lcore_id], 0, &m, 1);
        if (unlikely(ret < 1)) {
            rte_pktmbuf_free(m);
            lcore_stats[lcore_id].dropped++;
        }
        else {
            lcore_stats[lcore_id].tx++;
        }
    }

为了创建环路，测试吞吐量，TAP接口可以通过网桥连接。

4.4.3. 管理TAP接口和桥接

Exception Path用例创建TAP接口，TAP接口名称为tap_dpdk_nn，其中nn为lcore_ID，这些TAP接口需要配置才能使用：

ifconfig tap_dpdk_00 up

在两个接口间创建桥接，那么发往其中一个接口的报文可以从另一个接口上收到。

brctl addbr "br0"
brctl addif br0 tap_dpdk_00
brctl addif br0 tap_dpdk_03
ifconfig br0 up

示例程序创建的TAP接口只有当程序运行是才存在，所以上面的步骤需要在程序运行之后重复执行。为了避免这种情况，TAP接口可以通过openvpn命令来创建：

openvpn --mktun --dev tap_dpdk_00

使用这种方法，上面的步骤只需要执行一次，且TAP接口可以被程序重复利用。如果需要移除TAP接口，可以使用以下命令：

ifconfig br0 down
brctl delbr br0
openvpn --rmtun --dev tap_dpdk_00

5. Hello World

5.1. 编译程序

5.2. 运行程序

5.3. 代码逻辑

初始化eal
一旦eal初始化完成，应用程序就可以在lcore上开始运行。在这个例子中，lcore_hell函数在每个可用的lcore上调用
在每个lcore上执行，使用

/* call lcore_hello() on every slave lcore */
    RTE_LCORE_FOREACH_SLAVE(lcore_id) {
        rte_eal_remote_launch(lcore_hello, NULL, lcore_id);
    }
    /* call it on master lcore too */
    lcore_hello(NULL);

6. Basic Forwarding

Basci Forwarding示例程序是转发应用程序的简单框架示例。它旨在演示DPDK转发应用程序的基本组件，有关更详细的实现，请参阅L2和L3转发示例。

6.1. 编译程序

6.2. 运行程序

6.3. 代码逻辑

6.3.1. 主函数

主函数执行初始化操作，并为每个lcore调用执行线程。

首先是初始化环境失配层EAL，argc和argv参数都是传递给rte_eal_init()函数的。

/* Initialize the Environment Abstraction Layer (EAL). */
int ret = rte_eal_init(argc, argv);
if (ret < 0)
    rte_exit(EXIT_FAILURE, "Error with EAL initialization\n");

Main函数中还申请了内存池以提供程序中使用的mbuf。

/* 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());

if (mbuf_pool == NULL)
    rte_exit(EXIT_FAILURE, "Cannot create mbuf pool\n");

Mbuf是DPDK中所使用的报文缓冲区数据结构。

Main函数中还需要初始化程序要用到的端口：

/* Initialize all ports. */
for (portid = 0; portid < nb_ports; portid++)
    if (port_init(portid, mbuf_pool) != 0)
        rte_exit(EXIT_FAILURE, "Cannot init port %"PRIu8 "\n",
                portid);

一旦这些初始化都完成之后，应用程序就可以在指定的lcore上开始运行了。在这个程序中，只运行于单个lcore。

6.3.2. 端口初始化函数

端口初始化函数代码如下：

/*
 * Initializes a given port using global settings and with the RX buffers
 * coming from the mbuf_pool passed as a parameter.
 */
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;
    int retval;
    uint16_t q;

    if (port >= rte_eth_dev_count())
        return -1;

    /* Configure the Ethernet device. */
    retval = rte_eth_dev_configure(port, rx_rings, tx_rings, &port_conf);
    if (retval != 0)
        return retval;

    /* 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);
        if (retval < 0)
            return retval;
    }

    /* 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);
        if (retval < 0)
            return retval;
    }

    /* Start the Ethernet port. */
    retval = rte_eth_dev_start(port);
    if (retval < 0)
        return retval;

 ……

    /* Enable RX in promiscuous mode for the Ethernet device. */
    rte_eth_promiscuous_enable(port);

    return 0;
}

以太网端口使用rte_eth_dev_configure及port_conf_default进行默认配置。默认配置中只配置了收包长度1518。
在这个例子中，每个端口创建了1个RX队列和1个TX队列。
使用rte_eth_dev_start来启动端口。

6.3.3. 主循环

所有初始化做完之后，在lcore上开启主循环。在本例中，只在一个lcore上运行。

/*
 * The lcore main. This is the main thread that does the work, reading from
 * an input port and writing to an output port.
 */
static __attribute__((noreturn)) void lcore_main(void)
{
    const uint8_t nb_ports = rte_eth_dev_count();
    uint8_t port;

    /*
     * Check that the port is on the same NUMA node as the polling thread
     * for best performance.
     */
    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);

    printf("\nCore %u forwarding packets. [Ctrl+C to quit]\n", rte_lcore_id());

    /* 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 packets, 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]);
            }
        }
    }
}

这里的port^1操作，是将port 0收到的报文发往port 1，port 1收到的报文发往port 0；port 2收到的报文发往port 3，port 3收到的报文发往port 2。

发送函数rte_eth_tx_burst将自动释放报文缓存，如果报文发送失败，则需要手动调用rte_pktmbuf_free来释放报文缓存。

可以使用Ctrl+C来终止程序。

7. RX/RX Callbacks

RX/TX Callback示例是报文转发用例，展示了使用用户自定义的回调函数来接收和发送报文。用例实现了一个简单的时延检查，使用自定义的回调函数，用于确定报文在程序中的耗时。

在示例中，用户定义的回调函数给每个接收报文提供了一个时间戳，在报文传输之前，回调函数对所有的数据包计算CPU周期。

7.1. 编译程序

注意RX/TX Callback示例需要配置CONFIG_RTE_ETHDEV_RXTX_CALLBACKS，该配置在/config/comman_config中，通常情况下，默认是开启这个配置的。

7.2. 运行程序

7.3. 代码逻辑

RX/TX Callback示例是简单的转发用例，基于Basic Forwarding程序。

7.3.1. 主函数

Main函数执行初始化操作，在每个指定的lcore上调用执行线程。具体过程与Basic Forwarding一样。
主循环也基本一样。
主要的差别在于端口初始化部分，在这里我们添加了回调函数。

7.3.2. 端口初始化函数

端口初始化的代码逻辑也可以参考上一个用例，只是在最后，我们添加了以下两行代码：

    rte_eth_add_rx_callback(port, 0, add_timestamps, NULL);
    rte_eth_add_tx_callback(port, 0, calc_latency, NULL);

显然，在每个端口的队列0上添加了RX/TX回调。
注意，可以添加一个或多个回调函数，且可以向回调函数传入void*参数，本例子中传入的是NULL参数。

7.3.3. RX Callback

本例子中，添加了add_timestamps回调，当每个报文收到时，将执行这个函数：

static uint16_t
add_timestamps(uint8_t port __rte_unused, uint16_t qidx __rte_unused,
        struct rte_mbuf **pkts, uint16_t nb_pkts,
        uint16_t max_pkts __rte_unused, void *_ __rte_unused)
{
    unsigned i;
    uint64_t now = rte_rdtsc();

    for (i = 0; i < nb_pkts; i++)
        pkts[i]->udata64 = now;
    return nb_pkts;
}

7.3.4. TX Callback

本例中，添加了TX回调calc_latency，当每个报文发送之前，计算耗时：

static uint16_t
calc_latency(uint8_t port __rte_unused, uint16_t qidx __rte_unused,
        struct rte_mbuf **pkts, uint16_t nb_pkts, void *_ __rte_unused)
{
    uint64_t cycles = 0;
    uint64_t now = rte_rdtsc();
    unsigned i;

    for (i = 0; i < nb_pkts; i++)
        cycles += now - pkts[i]->udata64;
    latency_numbers.total_cycles += cycles;
    latency_numbers.total_pkts += nb_pkts;

    if (latency_numbers.total_pkts > (100 * 1000 * 1000ULL)) {
        printf("Latency = %"PRIu64" cycles\n",
        latency_numbers.total_cycles / latency_numbers.total_pkts);
        latency_numbers.total_cycles = latency_numbers.total_pkts = 0;
    }
    return nb_pkts;
}

当一亿个报文发送时，打印出平均耗时，并重置计数。