DPDK-收包完整过程

本篇博客作为自己了解dpdk收包过程的一个记录。在写时发现已经有很多写DPDK收包过程的博客了，但还是决定自己写一遍。

DPDK收包分为两个阶段，首先是DMA将数据包从网卡搬运到内存，然后是调用dpdk提供的接口rte_eth_rx_burst去取。但是具体是怎么做的呢？简单大致的过程如图1所示。

图1，dpdk收包大致过程

本文将按如下顺序进行展开。首先会简单介绍dma的环形缓存区，接着会介常规linux dma的收发包过程，然后会结合源码介绍dpdk的双环形缓存区设计，最后结合源码给出dpdk收包的过程。

1、DMA搬运数据包过程

网卡DMA控制器通过环形队列与CPU交互，环形队列由一组控制寄存器和一块物理上连续的缓存构成。主要的控制寄存器有Base, Size, Head和Tail。通过设置Base寄存器，可以将分配的一段物理连续的内存地址作为环形队列的起始地址，通告给DMA控制器。同样通过Size寄存器，可以通告内存快的大小。Head寄存器往往对软件只读，它表示硬件当前访问的描述符单元。而Tail寄存器则由软件来填写更新，通知DMA控制器当前已准备好被硬件访问的描述符单元。这一段话来自《深入浅出DPDK》的描述。

图2，网卡的收发描述符

下面我们结合图3先简单看下Linux中DMA收包的过程，后面再给出dpdk的收包过程。

图3，linxu dma收发包过程

详细过程如下：

1、cpu填充地址到接收侧描述符，也就是将存放sk_buff的地址物理地址填写到Rx ring；

2、网卡读取接收侧描述符获取缓冲区地址；

3、网卡收到数据包后，根据2中读到的缓冲区地址，将数据包的内容通过dma写到缓冲区；

4、网卡回写接收侧描述符更新状态，表示数据包已写完；

5、cpu读取接收侧描述符以确定数据包接收完毕；

6、cpu读取包内容做转发判断；

7、cpu填充更改包内容，做发送准备，比如调换ip地址和mac地址；

8、cpu读取发送侧描述符，检查是否有发送完成标志；

9、cpu将准备发送的缓冲区地址填写到发送侧描述符，也就是将sk_buff的地址填写到Tx ring；

10、网卡读取发送侧描述符中的地址；

11、网卡根据描述符中地址，读取缓冲区的数据内容进行发送；

12、网卡写发送侧描述符，更新发送已完成标记。

上述步骤1-6为接收过程，步骤7-12为发送过程。步骤1-12这段描述也copy自《深入浅出DPDK》。

但是，这里我要补充的是，图2中的rxd和txd就是我们图1中介绍的环形缓冲区，分别叫做Rx ring和Tx ring，也统一叫做DMA ring buffer。

DMA ring buffer的创建在驱动初始化阶段，这块内存由网卡与处理器共享，详细描述见《Linux设备驱动程序》第15章内存直接访问。这也就解释了为啥cpu可以写和读Rx/Tx ring中的描述符，网卡也可以。

2、DPDK收包

在上文描述了linux dma收包过程，那么使用dpdk收包会有啥不同呢？要弄清楚dpdk的收包过程，需要先清楚dpdk的双环形缓存区设计。

2.1 DPDK双环形缓冲区结构

dpdk收包会用到两个ring，一个是rx ring，一个是sw ring，rx ring存放的是数据包的dma地址，sw ring存放的是rte_mbuf的地址，rx ring和sw ring是一一映射的。

图4，rx_ring和sw_ring的映射关系

如图4所以，rx_ring里面的每个描述符中的地址填的是rte_mbuf的buf data地址，而rx_ring的每个描述符存放的是rte_mbuf的地址，也就是说当dma拿到dma_addr1后，然后将数据包写到dma_addr1指向的地址，实际是写到了rte_mbuf的data buffer里。

而rx_ring是一个环形，sw_ring也是一个环形，这就是《深入浅出DPDK》第6章中描述的双环形缓冲区，见图5。

图5，内存池的双环形缓存区结构

这里有一个问题，就是rx_ring和sw_ring是啥时候初始化的以及建立映射关系的？

拿ice driver举例，先看一个结构体struct ice_rx_queue，rx_ring和sw_ring成员是我们后面的重点关注对象。

（免费订阅,永久学习）学习地址: Dpdk/网络协议栈/vpp/OvS/DDos/NFV/虚拟化/高性能专家-学习视频教程-腾讯课堂

更多DPDK相关学习资料有需要的可以自行报名学习,免费订阅,永久学习,或点击这里加qun免费
领取,关注我持续更新哦! !

@dpdk-stable-20.11.3/drivers/net/ice/ice_rxtx.h
struct ice_rx_entry {
    struct rte_mbuf *mbuf;
};

struct ice_rx_queue {
    struct rte_mempool *mp; /* mbuf pool to populate RX ring */
    volatile union ice_rx_flex_desc *rx_ring;/* RX ring virtual address */
    rte_iova_t rx_ring_dma; /* RX ring DMA address */
    struct ice_rx_entry *sw_ring; /* address of RX soft ring */
    uint16_t nb_rx_desc; /* number of RX descriptors */
    ...
}

rx_ring和sw_ring是ice_rx_queue结构体成员。rx_ring和sw_ring的初始化都是在dpdk应用程序中调用rte_eth_rx_queue_setup函数时进行的，这个函数最后调用了ice_rx_queue_setup。

@dpdk-stable-20.11.3/drivers/net/ice/ice_rxtx.c
int ice_rx_queue_setup(struct rte_eth_dev *dev,
           uint16_t queue_idx,
           uint16_t nb_desc,
           unsigned int socket_id,
           const struct rte_eth_rxconf *rx_conf,
           struct rte_mempool *mp)
{
    struct ice_pf *pf = ICE_DEV_PRIVATE_TO_PF(dev->data->dev_private);
    struct ice_adapter *ad =
        ICE_DEV_PRIVATE_TO_ADAPTER(dev->data->dev_private);
    struct ice_vsi *vsi = pf->main_vsi;
    struct ice_rx_queue *rxq;
    const struct rte_memzone *rz;
    uint32_t ring_size;
    uint16_t len;
    int use_def_burst_func = 1;

...

    /* Allocate the rx queue data structure */
    rxq = rte_zmalloc_socket(NULL,                //创建rxq，rxq的类型为ice_rx_queue
                 sizeof(struct ice_rx_queue),
                 RTE_CACHE_LINE_SIZE,
                 socket_id);

    ...
    /* Allocate the maximun number of RX ring hardware descriptor. */
    len = ICE_MAX_RING_DESC;

    /**
     * Allocating a little more memory because vectorized/bulk_alloc Rx
     * functions doesn't check boundaries each time.
     */
    len += ICE_RX_MAX_BURST;

    /* Allocate the maximum number of RX ring hardware descriptor. */
    ring_size = sizeof(union ice_rx_flex_desc) * len;
    ring_size = RTE_ALIGN(ring_size, ICE_DMA_MEM_ALIGN);
    /* 创建rx_ring */
    rz = rte_eth_dma_zone_reserve(dev, "rx_ring", queue_idx,
                      ring_size, ICE_RING_BASE_ALIGN,
                      socket_id);
    ...

    /* Zero all the descriptors in the ring. */
    memset(rz->addr, 0, ring_size); 
    /* 设置rx_ring的dma地址和虚拟地址。*/
    rxq->rx_ring_dma = rz->iova;
    rxq->rx_ring = rz->addr;

    /* always reserve more for bulk alloc */
    len = (uint16_t)(nb_desc + ICE_RX_MAX_BURST);

    /* Allocate the software ring. */
    rxq->sw_ring = rte_zmalloc_socket(NULL,              //创建sw_ring
                      sizeof(struct ice_rx_entry) * len,
                      RTE_CACHE_LINE_SIZE,
                      socket_id);

    ...
}

根据上面的分析，我们知道了rx_ring和sw_ring是在调用rte_eth_rx_queue_setup进而调用ice_rx_queue_setup时创建的，但是我们不知道这两个ring是啥时候建立的映射。

rx_ring和sw_ring映射关系的建立发生在rte_eth_dev_start调用栈里，具体为调用关系为rte_eth_dev_start-->ice_dcf_dev_start-->ice_dcf_start_queues-->ice_dcf_rx_queue_start-->alloc_rxq_mbufs，看看源码吧。

@dpdk-stable-20.11.3/drivers/net/ice/ice_dcf_ethdev.c
static int
alloc_rxq_mbufs(struct ice_rx_queue *rxq)
{
    volatile union ice_rx_flex_desc *rxd;
    struct rte_mbuf *mbuf = NULL;
    uint64_t dma_addr;
    uint16_t i;

    for (i = 0; i < rxq->nb_rx_desc; i++) {
        mbuf = rte_mbuf_raw_alloc(rxq->mp);     //申请一个mbuf
        if (unlikely(!mbuf)) {
            PMD_DRV_LOG(ERR, "Failed to allocate mbuf for RX");
            return -ENOMEM;
        }

        rte_mbuf_refcnt_set(mbuf, 1);
        mbuf->next = NULL;
        mbuf->data_off = RTE_PKTMBUF_HEADROOM;
        mbuf->nb_segs = 1;
        mbuf->port = rxq->port_id;
        
        //获取rte_mbuf data buff的dma地址
        dma_addr =
            rte_cpu_to_le_64(rte_mbuf_data_iova_default(mbuf));

        rxd = &rxq->rx_ring[i];
        //填充rx_ring的描述符，其实就是rte_mbuf data buff的dma地址，当网卡收到数据包后直接通过
        //dma往这个地址写，所以数据包就存放rte_mbuf里面了。
        rxd->read.pkt_addr = dma_addr;
        rxd->read.hdr_addr = 0;
#ifndef RTE_LIBRTE_ICE_16BYTE_RX_DESC
        rxd->read.rsvd1 = 0;
        rxd->read.rsvd2 = 0;
#endif
        //填充sw_ring，结合图4，这样sw_ring和rx_ring就建立起了映射关系
        rxq->sw_ring[i].mbuf = (void *)mbuf;
    }

    return 0;
}

简单的说就是rte_eth_rx_queue_setup函数创建rx_ring和sw_ring，然后rte_eth_dev_start建立rx_ring和sw_ring之间的映射关系。

2.2 dpdk收包过程

花了很大篇幅来说明网卡的收包原理，以及解释dpdk底层相关的知识，现在终于可以说说dpdk是怎么收包的了。

dpdk提供的收包接口为rte_eth_rx_burst。这个函数最后其实调用了ice_recv_pkts函数。

@dpdk-stable-20.11.3/drivers/net/ice/ice_rxtx.c

uint16_t ice_recv_pkts(void *rx_queue, struct rte_mbuf **rx_pkts, uint16_t nb_pkts)
{
    struct ice_rx_queue *rxq = rx_queue;
    volatile union ice_rx_flex_desc *rx_ring = rxq->rx_ring;   //重点关注
    volatile union ice_rx_flex_desc *rxdp;
    union ice_rx_flex_desc rxd;
    struct ice_rx_entry *sw_ring = rxq->sw_ring;    //重点关注
    struct ice_rx_entry *rxe;
    struct rte_mbuf *nmb; /* new allocated mbuf */
    struct rte_mbuf *rxm; /* pointer to store old mbuf in SW ring */
    uint16_t rx_id = rxq->rx_tail;
    uint16_t nb_rx = 0;
    uint16_t nb_hold = 0;
    uint16_t rx_packet_len;
    uint16_t rx_stat_err0;
    uint64_t dma_addr;
    uint64_t pkt_flags;
    uint32_t *ptype_tbl = rxq->vsi->adapter->ptype_tbl;
    struct rte_eth_dev *dev;

    while (nb_rx < nb_pkts) {         //期望接收nb_pkts个数据包
        rxdp = &rx_ring[rx_id];   //rx_id是rx_ring新包的起始id，因为rx_ring是一个环
        rx_stat_err0 = rte_le_to_cpu_16(rxdp->wb.status_error0);

        /* Check the DD bit first */
        if (!(rx_stat_err0 & (1 << ICE_RX_FLEX_DESC_STATUS0_DD_S)))
            break;

        /* allocate mbuf */   
        nmb = rte_mbuf_raw_alloc(rxq->mp);  //申请一个新的rte_mbuf，用于回填rx_ring和sw_ring
        if (unlikely(!nmb)) {
            dev = ICE_VSI_TO_ETH_DEV(rxq->vsi);
            dev->data->rx_mbuf_alloc_failed++;
            break;
        }
        //rxdp指向一个rx_ring，现在rxd也指向了这个rx_ring
        rxd = *rxdp; /* copy descriptor in ring to temp variable*/
        
        nb_hold++;
        //rx_id是sw_ring新包的起始id，因为sw_ring是一个环
        rxe = &sw_ring[rx_id]; /* get corresponding mbuf in SW ring */
        rx_id++;  //rx_ring和sw_ring的下一个地址
        if (unlikely(rx_id == rxq->nb_rx_desc))
            rx_id = 0;   //表示到了ring的尾部，重新开始从ring的头部开始取rte_mbuf
        //保存sw_ring里面的rte_mbuf指针，这个rte_mbuf里面是有数据的
        rxm = rxe->mbuf;
        //给sw_ring里面回填新的描述符，也就是新申请的rte_mbuf指针，用于下一轮收包
        rxe->mbuf = nmb;
        //计算新申请的rte_mbuf data buff的dma地址
        dma_addr =
            rte_cpu_to_le_64(rte_mbuf_data_iova_default(nmb));

        /**
         * fill the read format of descriptor with physic address in
         * new allocated mbuf: nmb
         */
        rxdp->read.hdr_addr = 0;
        //更新rx_ring里面的描述符地址
        rxdp->read.pkt_addr = dma_addr;

        /* calculate rx_packet_len of the received pkt */
        rx_packet_len = (rte_le_to_cpu_16(rxd.wb.pkt_len) &
                 ICE_RX_FLX_DESC_PKT_LEN_M) - rxq->crc_len;

        //填充收到的rte_mbuf里面的相关字段，这个应该很熟悉了
        /* fill old mbuf with received descriptor: rxd */
        rxm->data_off = RTE_PKTMBUF_HEADROOM;
        rte_prefetch0(RTE_PTR_ADD(rxm->buf_addr, RTE_PKTMBUF_HEADROOM));
        rxm->nb_segs = 1;
        rxm->next = NULL;
        rxm->pkt_len = rx_packet_len;
        rxm->data_len = rx_packet_len;
        rxm->port = rxq->port_id;
        rxm->packet_type = ptype_tbl[ICE_RX_FLEX_DESC_PTYPE_M &
            rte_le_to_cpu_16(rxd.wb.ptype_flex_flags0)];
        ice_rxd_to_vlan_tci(rxm, &rxd);
        rxq->rxd_to_pkt_fields(rxq, rxm, &rxd);
        pkt_flags = ice_rxd_error_to_pkt_flags(rx_stat_err0);
        rxm->ol_flags |= pkt_flags;
        /* copy old mbuf to rx_pkts */
        rx_pkts[nb_rx++] = rxm;
    }
    rxq->rx_tail = rx_id;
    /**
     * If the number of free RX descriptors is greater than the RX free
     * threshold of the queue, advance the receive tail register of queue.
     * Update that register with the value of the last processed RX
     * descriptor minus 1.
     */
    nb_hold = (uint16_t)(nb_hold + rxq->nb_rx_hold);
    if (nb_hold > rxq->rx_free_thresh) {
        rx_id = (uint16_t)(rx_id == 0 ?
                   (rxq->nb_rx_desc - 1) : (rx_id - 1));
        /* write TAIL register */
        ICE_PCI_REG_WC_WRITE(rxq->qrx_tail, rx_id);
        nb_hold = 0;
    }
    rxq->nb_rx_hold = nb_hold;

    /* return received packet in the burst */
    return nb_rx;
}