本文整理下之前的学习笔记，基于DPDK17.11版本源码，主要分析一下收发包流程。
使用DPDK的APP收发报文流程如下
main
//环境抽象层初始化，比如网卡，cpu，内存等
rte_eal_init(argc, argv);
//为rx和tx队列分配内存，将用户指定的配置信息dev_conf保存到dev
rte_eth_dev_configure(portid, 1, 1, &port_conf);
//分配网卡接收队列结构体，接收ring硬件描述符和软件ring等内存
rte_eth_rx_queue_setup(portid, 0, nb_rxd,
rte_eth_dev_socket_id(portid),
NULL,
l2fwd_pktmbuf_pool);
//分配网卡发送队列结构体，发送ring硬件描述符等内存
rte_eth_tx_queue_setup(portid, 0, nb_txd,
rte_eth_dev_socket_id(portid),
NULL);
//启动网卡，设置网卡寄存器，将网卡和系统内存关联起来
rte_eth_dev_start(portid);
while (1) {
//接收报文
rte_eth_rx_burst(portid, 0, pkts_burst, MAX_PKT_BURST);
//处理报文
//发送报文，此函数只是将报文放到一个buffer中，满32个后才调用rte_eth_tx_burst真正发送
rte_eth_tx_buffer(dst_port, 0, buffer, m);
}
以ixgbe驱动为例，相关的数据结构如下
image.png
收包流程
我们都知道网卡会通过DMA将报文放在系统内存中，那网卡如何知道应该放在哪里呢？如何将网卡和系统内存关联起来？这需要用到网卡的几个寄存器:
RDBAL(Receive Descriptor Base Address Low)，
RDBAH(Receive Descriptor Base Address High)
RDLEN(Receive Descriptor Length)
驱动初始化时会分配一块内存，将这块内存的起始物理地址(64位)写到寄存器RDBAL(保存物理地址的低32位)和RDBAH(保存物理地址的高32位)，然后将这块内存的大小写到寄存器RDLEN中。这块内存称为硬件描述符，大小为接收队列硬件描述符个数乘接收队列硬件描述符大小。
一个接收队列硬件描述符大小为16字节，有两种格式: 读格式和回写格式。
读格式是从网卡角度来说的，由驱动将mbuf的物理地址写到packet buffer address字段，网卡读取此字段获取内存物理地址，收到的报文就可以存到此内存。
image.png
回写格式也是从网卡角度来说，网卡将报文写到指定的内存后，就会以下面的格式将报文的相关信息回写到描述符中，最后设置DD位(第二个8字节的最低位)，驱动通过判断DD位是否为1来接收报文。
image.png
总结一下接收队列硬件描述符就是一块内存，网卡先以读格式获取内存的物理地址，将报文写到内存后，就以回写格式将报文额外信息写到描述符中，驱动可以以回写格式读取描述符，获取报文的长度，类型等信息。
网卡和内存关联起来后，就可以收取报文了，此时又用到两个寄存器: RDH(Receive Descriptor Head)和RDT(Receive Descriptor Tail)。
RDH为头指针，指向第一个可用描述符，网卡收取报文并回写成功后，由网卡来移动RDH到下一个可用描述符。
RDT为尾指针，指向最后一个可用描述符，RDH和RDT之间的描述符为网卡可用描述符，RDT由驱动来移动，驱动从第一个描述符开始，轮询DD位是否为1，为1就认为此描述符对应的mbuf有报文，此时会申请新的mbuf，将新mbuf物理地址写到此描述符的pkt_addr，并将DD位置0，这样的话此描述符就又可用被网卡使用了，同时将老的有报文的mbuf返回给用户。描述符再次可用后，驱动就可以更新RDT指向此描述符，为了性能考虑不会每次都会更新RDT，而是等可用描述符超过一定阈值(rx_free_thresh)才更新一次。
image.png
如下为接收描述符的格式，是union类型，可同时有读和回写两种格式。
/* Receive Descriptor - Advanced */
union ixgbe_adv_rx_desc {
struct {
__le64 pkt_addr; /* Packet buffer address */
__le64 hdr_addr; /* Header buffer address */
} read;
struct {
struct {
union {
__le32 data;
struct {
__le16 pkt_info; /* RSS, Pkt type */
__le16 hdr_info; /* Splithdr, hdrlen */
} hs_rss;
} lo_dword;
union {
__le32 rss; /* RSS Hash */
struct {
__le16 ip_id; /* IP id */
__le16 csum; /* Packet Checksum */
} csum_ip;
} hi_dword;
} lower;
struct {
__le32 status_error; /* ext status/error */
__le16 length; /* Packet length */
__le16 vlan; /* VLAN tag */
} upper;
} wb; /* writeback */
};
了解网卡接收原理后，下面从代码角度看一下实现，大概分为如下几步：
a. 分配接收队列硬件描述符rx_ring，分配软件ring sw_ring
b. 将接收队列硬件描述符的物理地址和长度写到寄存器
c. 分配mbuf，将mbuf接收报文的物理地址赋给接收队列硬件描述符 rx_ring->pkt_addr，虚拟地址赋给 sw_ring
d. 设置头尾寄存器，头指针寄存器RDH为0，指向第一个可用描述符，尾指针寄存器RDT指向最后一个可用描述符
a. rte_eth_rx_queue_setup
接收队列设置
- 分配队列结构体 struct ixgbe_rx_queue
- 分配接收ring硬件描述符(一般为4096)，每个描述符16字节，保存到 rxq->rx_ring
- 分配软件ring，用来保存mbuf，保存到 rxq->sw_ring
rte_eth_rx_queue_setup -> ixgbe_dev_rx_queue_setup
int __attribute__((cold))
ixgbe_dev_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)
const struct rte_memzone *rz;
struct ixgbe_rx_queue *rxq;
struct ixgbe_hw *hw;
uint16_t len;
struct ixgbe_adapter *adapter = (struct ixgbe_adapter *)dev->data->dev_private;
hw = IXGBE_DEV_PRIVATE_TO_HW(dev->data->dev_private);
/* First allocate the rx queue data structure */
rxq = rte_zmalloc_socket("ethdev RX queue", sizeof(struct ixgbe_rx_queue),
RTE_CACHE_LINE_SIZE, socket_id);
rxq->mb_pool = mp;
rxq->nb_rx_desc = nb_desc;
rxq->rx_free_thresh = rx_conf->rx_free_thresh;
rxq->queue_id = queue_idx;
rxq->reg_idx = (uint16_t)((RTE_ETH_DEV_SRIOV(dev).active == 0) ?
queue_idx : RTE_ETH_DEV_SRIOV(dev).def_pool_q_idx + queue_idx);
rxq->port_id = dev->data->port_id;
rxq->crc_len = (uint8_t) ((dev->data->dev_conf.rxmode.hw_strip_crc) ? 0 : ETHER_CRC_LEN);
rxq->drop_en = rx_conf->rx_drop_en;
rxq->rx_deferred_start = rx_conf->rx_deferred_start;
#define IXGBE_MAX_RING_DESC 4096 /* replicate define from rxtx */
#define RTE_PMD_IXGBE_RX_MAX_BURST 32
#define RX_RING_SZ ((IXGBE_MAX_RING_DESC + RTE_PMD_IXGBE_RX_MAX_BURST) * \
sizeof(union ixgbe_adv_rx_desc))
/*
* Allocate RX ring hardware descriptors. A memzone large enough to
* handle the maximum ring size is allocated in order to allow for
* resizing in later calls to the queue setup function.
*/
//分配接收队列硬件描述符内存，注意这里是按最大值分配。
//注意要128字节对齐，因为82599网卡芯片手册规则物理地址必须是128字节对齐
rz = rte_eth_dma_zone_reserve(dev, "rx_ring", queue_idx,
RX_RING_SZ, IXGBE_ALIGN, socket_id);
/*
* Zero init all the descriptors in the ring.
*/
memset(rz->addr, 0, RX_RING_SZ);
rxq->rdt_reg_addr =
IXGBE_PCI_REG_ADDR(hw, IXGBE_RDT(rxq->reg_idx));
rxq->rdh_reg_addr =
IXGBE_PCI_REG_ADDR(hw, IXGBE_RDH(rxq->reg_idx));
//保存接收队列硬件描述符的物理地址
rxq->rx_ring_phys_addr = rz->iova;
//保存接收队列硬件描述符的虚拟地址
rxq->rx_ring = (union ixgbe_adv_rx_desc *) rz->addr;
/*
* Allocate software ring. Allow for space at the end of the
* S/W ring to make sure look-ahead logic in bulk alloc Rx burst
* function does not access an invalid memory region.
*/
len = nb_desc;
if (adapter->rx_bulk_alloc_allowed)
len += RTE_PMD_IXGBE_RX_MAX_BURST;
//分配软件ring内存，这里的大小为参数指定的描述符个数 nb_desc
rxq->sw_ring = rte_zmalloc_socket("rxq->sw_ring",
sizeof(struct ixgbe_rx_entry) * len,
RTE_CACHE_LINE_SIZE, socket_id);
//将接收队列结构保存到对应位置
dev->data->rx_queues[queue_idx] = rxq;
b. ixgbe_dev_rx_init
将接收队列硬件描述符的物理地址写到网卡寄存器RDBAL和RDBAH，将接收队列硬件描述符的长度写到网卡寄存器RDLEN。
rte_eth_dev_start -> ixgbe_dev_start -> ixgbe_dev_rx_init
接收队列初始化
/*
* Initializes Receive Unit.
*/
int __attribute__((cold))
ixgbe_dev_rx_init(struct rte_eth_dev *dev)
{
struct ixgbe_hw *hw;
struct ixgbe_rx_queue *rxq;
uint64_t bus_addr;
uint32_t rxctrl;
uint32_t fctrl;
uint32_t hlreg0;
uint16_t i;
struct rte_eth_rxmode *rx_conf = &dev->data->dev_conf.rxmode;
int rc;
hw = IXGBE_DEV_PRIVATE_TO_HW(dev->data->dev_private);
/*
* Make sure receives are disabled while setting
* up the RX context (registers, descriptor rings, etc.).
*/
//确保网卡的接收功能是关闭的
rxctrl = IXGBE_READ_REG(hw, IXGBE_RXCTRL);
IXGBE_WRITE_REG(hw, IXGBE_RXCTRL, rxctrl & ~IXGBE_RXCTRL_RXEN);
//使能接收广播，丢弃pause报文
/* Enable receipt of broadcasted frames */
fctrl = IXGBE_READ_REG(hw, IXGBE_FCTRL);
fctrl |= IXGBE_FCTRL_BAM; /* Broadcast Accept Mode */
fctrl |= IXGBE_FCTRL_DPF; /* Discard Pause Frame */
fctrl |= IXGBE_FCTRL_PMCF; /* Pass MAC Control Frames */
IXGBE_WRITE_REG(hw, IXGBE_FCTRL, fctrl);
/*
* Configure CRC stripping, if any.
*/
//设置硬件自动去掉crc
hlreg0 = IXGBE_READ_REG(hw, IXGBE_HLREG0);
if (rx_conf->hw_strip_crc)
hlreg0 |= IXGBE_HLREG0_RXCRCSTRP;
else
hlreg0 &= ~IXGBE_HLREG0_RXCRCSTRP;
/*
* Configure jumbo frame support, if any.
*/
//使能接收巨帧
if (rx_conf->jumbo_frame == 1) {
hlreg0 |= IXGBE_HLREG0_JUMBOEN;
maxfrs = IXGBE_READ_REG(hw, IXGBE_MAXFRS);
maxfrs &= 0x0000FFFF;
maxfrs |= (rx_conf->max_rx_pkt_len << 16);
IXGBE_WRITE_REG(hw, IXGBE_MAXFRS, maxfrs);
} else
hlreg0 &= ~IXGBE_HLREG0_JUMBOEN;
IXGBE_WRITE_REG(hw, IXGBE_HLREG0, hlreg0);
/* Setup RX queues */
for (i = 0; i < dev->data->nb_rx_queues; i++) {
rxq = dev->data->rx_queues[i];
//将接收队列硬件描述符的物理地址写到网卡接收描述符寄存器中
/* Setup the Base and Length of the Rx Descriptor Rings */
bus_addr = rxq->rx_ring_phys_addr;
IXGBE_WRITE_REG(hw, IXGBE_RDBAL(rxq->reg_idx), (uint32_t)(bus_addr & 0x00000000ffffffffULL));
IXGBE_WRITE_REG(hw, IXGBE_RDBAH(rxq->reg_idx), (uint32_t)(bus_addr >> 32));
//将用户请求的nb_tx_desc个数的接收队列硬件描述符长度写到寄存器
IXGBE_WRITE_REG(hw, IXGBE_RDLEN(rxq->reg_idx), rxq->nb_rx_desc * sizeof(union ixgbe_adv_rx_desc));
//头尾指针先设置为0
IXGBE_WRITE_REG(hw, IXGBE_RDH(rxq->reg_idx), 0);
IXGBE_WRITE_REG(hw, IXGBE_RDT(rxq->reg_idx), 0);
}
//根据设置选择不同的接收函数，后面会以 ixgbe_recv_pkts 为例说明
ixgbe_set_rx_function(dev);
...
return 0;
}
c. ixgbe_dev_rx_queue_start
申请mbuf，将mbuf存放报文的物理地址设置到接收队列硬件描述符的pkt_addr字段，这样网卡就知道收到报文后将报文放在哪里了。
rte_eth_dev_start -> ixgbe_dev_start -> ixgbe_dev_rxtx_start -> ixgbe_dev_rx_queue_start
/*
* Start Receive Units for specified queue.
*/
int __attribute__((cold))
ixgbe_dev_rx_queue_start(struct rte_eth_dev *dev, uint16_t rx_queue_id)
{
struct ixgbe_hw *hw;
struct ixgbe_rx_queue *rxq;
uint32_t rxdctl;
int poll_ms;
hw = IXGBE_DEV_PRIVATE_TO_HW(dev->data->dev_private);
if (rx_queue_id < dev->data->nb_rx_queues) {
rxq = dev->data->rx_queues[rx_queue_id];
//分配mbuf，填充到 rxq->sw_ring 中
/* Allocate buffers for descriptor rings */
if (ixgbe_alloc_rx_queue_mbufs(rxq) != 0) {
PMD_INIT_LOG(ERR, "Could not alloc mbuf for queue:%d",
rx_queue_id);
return -1;
}
...
//头指针为0，指向第一个可用描述符
IXGBE_WRITE_REG(hw, IXGBE_RDH(rxq->reg_idx), 0);
//尾指针为最大描述符，指向最后一个可用描述符
IXGBE_WRITE_REG(hw, IXGBE_RDT(rxq->reg_idx), rxq->nb_rx_desc - 1);
dev->data->rx_queue_state[rx_queue_id] = RTE_ETH_QUEUE_STATE_STARTED;
}
return 0;
}
static int __attribute__((cold))
ixgbe_alloc_rx_queue_mbufs(struct ixgbe_rx_queue *rxq)
{
struct ixgbe_rx_entry *rxe = rxq->sw_ring;
uint64_t dma_addr;
unsigned int i;
/* Initialize software ring entries */
for (i = 0; i < rxq->nb_rx_desc; i++) {
volatile union ixgbe_adv_rx_desc *rxd;
//分配mbuf
struct rte_mbuf *mbuf = rte_mbuf_raw_alloc(rxq->mb_pool);
mbuf->data_off = RTE_PKTMBUF_HEADROOM;
mbuf->port = rxq->port_id;
//获取mbuf存放报文的物理地址，注意不是mbuf的首地址
dma_addr =
rte_cpu_to_le_64(rte_mbuf_data_iova_default(mbuf));
rxd = &rxq->rx_ring[i];
//清空接收描述符的DD位
rxd->read.hdr_addr = 0;
//将mbuf接收报文的物理地址赋给描述符
rxd->read.pkt_addr = dma_addr;
rxe[i].mbuf = mbuf;
}
return 0;
}
最后使能网卡的接收功能 hw->mac.ops.enable_rx_dma(hw, rxctrl);
下面是正式收包流程，还以ixgbe驱动为例 rte_eth_rx_burst -> ixgbe_recv_pkts
uint16_t
ixgbe_recv_pkts(void *rx_queue, struct rte_mbuf **rx_pkts, uint16_t nb_pkts)
struct ixgbe_rx_queue *rxq;
volatile union ixgbe_adv_rx_desc *rx_ring;
volatile union ixgbe_adv_rx_desc *rxdp;
struct ixgbe_rx_entry *sw_ring;
struct ixgbe_rx_entry *rxe;
struct rte_mbuf *rxm;
struct rte_mbuf *nmb;
union ixgbe_adv_rx_desc rxd;
uint64_t dma_addr;
uint32_t staterr;
uint32_t pkt_info;
uint16_t pkt_len;
uint16_t rx_id;
uint16_t nb_rx;
uint16_t nb_hold;
uint64_t pkt_flags;
uint64_t vlan_flags;
nb_rx = 0;
nb_hold = 0;
rxq = rx_queue;
rx_id = rxq->rx_tail;
rx_ring = rxq->rx_ring;
sw_ring = rxq->sw_ring;
vlan_flags = rxq->vlan_flags;
while (nb_rx < nb_pkts) {
/*
* The order of operations here is important as the DD status
* bit must not be read after any other descriptor fields.
* rx_ring and rxdp are pointing to volatile data so the order
* of accesses cannot be reordered by the compiler. If they were
* not volatile, they could be reordered which could lead to
* using invalid descriptor fields when read from rxd.
*/
//获取硬件描述符
rxdp = &rx_ring[rx_id];
//获取硬件描述符的 status_error
staterr = rxdp->wb.upper.status_error;
//判断DD位是否被硬件置1，为1说明有报文，不是1就break
if (!(staterr & rte_cpu_to_le_32(IXGBE_RXDADV_STAT_DD)))
break;
rxd = *rxdp;
//分配一个新的mbuf
nmb = rte_mbuf_raw_alloc(rxq->mb_pool);
nb_hold++;
//获取软件ring的当前元素
rxe = &sw_ring[rx_id];
//尾指针加1
rx_id++;
//如果达到最大值，则翻转为0，相当于环形效果
if (rx_id == rxq->nb_rx_desc)
rx_id = 0;
//从rxe->mbuf取出mbuf地址，此mbuf已经有报文内容
rxm = rxe->mbuf;
//rxe->mbuf被赋予一个新的mbuf
rxe->mbuf = nmb;
//获取新mbuf的物理地址
dma_addr =
rte_cpu_to_le_64(rte_mbuf_data_iova_default(nmb));
//hdr_addr清0，就会将DD位也清0，否则下次循环到此描述符就会错误的认为有报文
rxdp->read.hdr_addr = 0;
//将mbuf的物理地址赋给描述符，网卡就可以把新报文写到新mbuf中
rxdp->read.pkt_addr = dma_addr;
//从描述符的wb字段获取报文相关的信息，包括长度，vlanid等，并填到mbuf中
pkt_len = (uint16_t) (rte_le_to_cpu_16(rxd.wb.upper.length) - rxq->crc_len);
rxm->data_off = RTE_PKTMBUF_HEADROOM;
rte_packet_prefetch((char *)rxm->buf_addr + rxm->data_off);
rxm->nb_segs = 1;
rxm->next = NULL;
rxm->pkt_len = pkt_len;
rxm->data_len = pkt_len;
rxm->port = rxq->port_id;
pkt_info = rte_le_to_cpu_32(rxd.wb.lower.lo_dword.data);
/* Only valid if PKT_RX_VLAN set in pkt_flags */
rxm->vlan_tci = rte_le_to_cpu_16(rxd.wb.upper.vlan);
...
/*
* Store the mbuf address into the next entry of the array
* of returned packets.
*/
//将已经有报文的mbuf返回给调用者
rx_pkts[nb_rx++] = rxm;
}
//更新尾指针
rxq->rx_tail = rx_id;
//nb_hold表示本次调用成功读取的报文个数，也同时意味着本次调用重新可用mbuf的个数，
//因为读取一次报文，就会分配新的mbuf，并赋给描述符，这个描述符就可以被网卡再次使用。
//rxq->nb_rx_hold是累计可用的描述符个数。
nb_hold = (uint16_t) (nb_hold + rxq->nb_rx_hold);
//如果累计的可用描述符个数超过了阈值，就要更新网卡能看到的描述符尾指针了。
//如果不更新尾指针，随着收包头指针一直增加，和尾指针重合时，就没有可用描述符了。
if (nb_hold > rxq->rx_free_thresh) {
PMD_RX_LOG(DEBUG, "port_id=%u queue_id=%u rx_tail=%u "
"nb_hold=%u nb_rx=%u",
(unsigned) rxq->port_id, (unsigned) rxq->queue_id,
(unsigned) rx_id, (unsigned) nb_hold,
(unsigned) nb_rx);
rx_id = (uint16_t) ((rx_id == 0) ?
(rxq->nb_rx_desc - 1) : (rx_id - 1));
IXGBE_PCI_REG_WRITE(rxq->rdt_reg_addr, rx_id);
//清空计数
nb_hold = 0;
}
//更新nb_rx_hold
rxq->nb_rx_hold = nb_hold;
return nb_rx;
发包流程
发送报文时也需要将网卡和内存关联起来，即将要发送的报文地址告诉网卡，这也是通过硬件描述符来实现的。
发送队列硬件描述符格式如下，也分为读和回写两种格式，都从网卡的角度来说。
对于读格式，驱动将报文的物理地址设置到第一个8字节的address字段，网卡读取此字段就能获取发送报文的物理地址，同时驱动也会设置第二个8字节的相关字段，比如报文长度，是否是最后一个报文段，何时回写等，网卡根据这些信息正确的将报文发送出去。
对于回写格式，只有一个字段有效，第二个8字节的第32位，此位代表DD(Descriptor Done)位，网卡完成报文发送后，并且此描述符设置了RS标志位，则会将此DD位设置为1，驱动读取此位就知道此描述符及它之前的描述符都可以被驱动使用。
image.png
DCMD字段中的RS(report status)位用来控制网卡何时回写DD位。注意和接收方向的区别，在接收方向网卡每收到一个报文就会回写一次接收描述符，将报文长度等信息填写到接收描述符，这是必须的，否则驱动怎么知道接收的报文多长呢，但是发送方向网卡不需要每发送一个报文就回写一次，并且每个报文回写会影响性能，驱动只关心报文是否发送成功，对应的发送描述符是否可用，可以通过参数tx_rs_thresh设置网卡多久回写一次，如果发送报文个数超过tx_rs_thresh，就会设置DCMD的RS位。
发送方向代码流程和接收方向大体相似，不再赘述。
总结
在pmd中，对于接收方向(从网卡收数据)来说，初始状态head指针指向base，tail指向指向base+len。网卡是生产者，通过移动head指针将数据放在mbuf中，驱动是消费者，将接收ring中buf_addr换成新mbuf的地址，旧的mbuf可以返回给应用程序来处理。驱动通过移动tail指针，将接收描述符还给网卡，但是并没有每次收包都更新收包队列尾部索引寄存器，而是在可释放的收包描述符数量达到一个阈值（rx_free_thresh）的时候才真正更新收包队列尾部索引寄存器。设置合适的可释放描述符数量阈值，可以减少没有必要的过多的收包队列尾部索引寄存器的访问，改善收包的性能。
对于发送方向来说，初始状态head和tail都指向base。驱动是生产者，发包时，先将发送数据的物理地址赋值给发送描述符的txd->read.buffer_addr，最后通过移动tail指针通知网卡有数据要发送。网卡是消费者，当获知tail指针移动就会发送数据，网卡发送完数据，会移动head指针。
Q && A
a. pmd发包时，如何通知网卡有新数据需要发送？
更新tail指针时就会触发网卡发送数据。比如在ixgbe_xmit_pkts函数最后，都会更新tail指针: IXGBE_PCI_REG_WRITE_RELAXED(txq->tdt_reg_addr, tx_id);
从网卡datasheet也能看到相关说明：
image.png
b. 网卡发送成功后，驱动怎么知道描述符可用？
从datasheet看到，有四种方法，默认采用第三种，即通过DD标志位获取
image.png
c. 网卡驱动发送方向，mbuf什么时候释放？
许多驱动程序并没有在数据包传输后立即将mbuf释放回到mempool或本地缓存中。相反，他们将mbuf留在Tx环中，当需要在Tx环中插入，或者 tx_rs_thresh 已经超过时，执行批量释放。
网友评论