本文主要介绍rte_mbuf与rte_mempool数据结构之间的组织关系、以及网卡接收到的数据是如何存储在rte_mbuf中的。
一、rte_mbuf、rte_mempool及网卡收到的数据包在内存中的组织结构
调用rte_mempool_create()函数创建rte_mempool的时候，指定申请多少个rte_mbuff及每个rte_mbuf中elt_size的大小。elt_size是为网卡接收的数据包预先分配的内存的大小，该内存块就是rte_mbuf->pkt.data的实际存储区域。具体如上图所示。
在申请的rte_mempool内存块中，最前面存储struct rte_mempool数据结构，后面紧接着是rte_pktmbuf_pool_private数据，再后面就是N个rte_mbuf内存块。
每个rte_mbuf内存中，最前面同样存储的是struct rte_mbuf数据结果，后面是RTE_PKTMBUF_HEADROOM，最后面就是实际网卡接收到的数据，如下：
struct rte_mbuf *m = _m; uint32_t buf_len = mp->elt_size - sizeof(struct rte_mbuf); RTE_MBUF_ASSERT(mp->elt_size >= sizeof(struct rte_mbuf)); memset(m, 0, mp->elt_size); /* start of buffer is just after mbuf structure */ m->buf_addr = (char *)m + sizeof(struct rte_mbuf); m->buf_physaddr = rte_mempool_virt2phy(mp, m) + sizeof(struct rte_mbuf); m->buf_len = (uint16_t)buf_len; /* keep some headroom between start of buffer and data */ m->pkt.data = (char*) m->buf_addr + RTE_MIN(RTE_PKTMBUF_HEADROOM, m->buf_len); /* init some constant fields */ m->type = RTE_MBUF_PKT; m->pool = mp; m->pkt.nb_segs = 1; m->pkt.in_port = 0xff;
二、网卡接收的数据是如何存储到rte_mbuf中的？
以e1000网卡为例，在网卡初始化的时候，调用eth_igb_rx_init()初始化网卡的收包队列。每个收包队列数据结果如下：
/** * Structure associated with each RX queue. */ struct igb_rx_queue { struct rte_mempool *mb_pool; /**< mbuf pool to populate RX ring. */ volatile union e1000_adv_rx_desc *rx_ring; /**< RX ring virtual address. */ uint64_t rx_ring_phys_addr; /**< RX ring DMA address. */ volatile uint32_t *rdt_reg_addr; /**< RDT register address. */ volatile uint32_t *rdh_reg_addr; /**< RDH register address. */ struct igb_rx_entry *sw_ring; /**< address of RX software ring. */ struct rte_mbuf *pkt_first_seg; /**< First segment of current packet. */ struct rte_mbuf *pkt_last_seg; /**< Last segment of current packet. */ uint16_t nb_rx_desc; /**< number of RX descriptors. */ uint16_t rx_tail; /**< current value of RDT register. */ uint16_t nb_rx_hold; /**< number of held free RX desc. */ uint16_t rx_free_thresh; /**< max free RX desc to hold. */ uint16_t queue_id; /**< RX queue index. */ uint16_t reg_idx; /**< RX queue register index. */ uint8_t port_id; /**< Device port identifier. */ uint8_t pthresh; /**< Prefetch threshold register. */ uint8_t hthresh; /**< Host threshold register. */ uint8_t wthresh; /**< Write-back threshold register. */ uint8_t crc_len; /**< 0 if CRC stripped, 4 otherwise. */ uint8_t drop_en; /**< If not 0, set SRRCTL.Drop_En. */ };
我们只关注其中两个成员变量，rx_ring和sw_ring。rx_ring记录的是union e1000_adv_rx_desc数组，每个union e1000_adv_rx_desc中指定了网卡接收数据的DMA地址，网卡收到数据后，直接往该地址写数据。sw_ring数组记录的是每个具体的rte_mbuf地址，每个rte_mbuf的rte_mbuff->buf_phyaddr + RTE_PKTMBUF_HEADROOM映射后的DMA地址就存储在rx_ring队列的union e1000_adv_rx_desc数据结构中。rte_mbuff->buf_phyaddr + RTE_PKTMBUF_HEADROOM指向的就是rte_mbuf->pkt.data的地址。此时，rte_mbuf、rte_mbuf->pkt.data，已及网卡的收包队列就关联起来了。具体如下：
static int igb_alloc_rx_queue_mbufs(struct igb_rx_queue *rxq) { struct igb_rx_entry *rxe = rxq->sw_ring; uint64_t dma_addr; unsigned i; /* Initialize software ring entries. */ for (i = 0; i < rxq->nb_rx_desc; i++) { volatile union e1000_adv_rx_desc *rxd; struct rte_mbuf *mbuf = rte_rxmbuf_alloc(rxq->mb_pool); if (mbuf == NULL) { PMD_INIT_LOG(ERR, "RX mbuf alloc failed " "queue_id=%hu ", rxq->queue_id); return (-ENOMEM); } dma_addr = rte_cpu_to_le_64(RTE_MBUF_DATA_DMA_ADDR_DEFAULT(mbuf)); rxd = &rxq->rx_ring[i]; rxd->read.hdr_addr = dma_addr; rxd->read.pkt_addr = dma_addr; rxe[i].mbuf = mbuf; } return 0; }
网卡收到数据后，向rx_ring指定的DMA地址上写数据，其实，就是往每个rte_mbuf->pkt.data写数据。应用程序在调用rte_eth_rx_burst()收包时，以e1000网卡为例，最后调用的是eth_igb_recv_pkts()，就是从每个收包队列中，从sw_ring数组中将rte_mbuf取出来，然后重启申请新的rte_mbuf替换到rx_ring中，重新关联rte_mbuf、union e1000_adv_rx_desc、sw_ring以及rte_mbuf->pkt.data的DMA地址。如下简图所示。
