DPDK收發包全景分析---以ixgbe分析

DPDK收發包全景分析---以ixgbe

前言：DPDK收發包是基礎核心模塊，從網卡收到包到驅動把包拷貝到系統內存中，再到系統對這塊數據包的內存管理，因爲在處理過程當中實現了零拷貝，數據包從接收到發送始終只有一份。這篇主要介紹收發包的過程。

1、收發包分解

收發包過程大體能夠分爲2個部分
1.收發包的配置和初始化，主要是配置收發隊列等。
2.數據包的獲取和發送，主要是從隊列中獲取到數據包或者把數據包放到隊列中。

2、收發包的配置和初始化

收發包的配置

收發包的配置最主要的工做就是配置網卡的收發隊列，設置DMA拷貝數據包的地址等，配置好地址後，網卡收到數據包後會經過DMA控制器直接把數據包拷貝到指定的內存地址。咱們使用數據包時，只要去對應隊列取出指定地址的數據便可。
收發包的配置是從rte_eth_dev_configure() 開始的，這裏根據參數會配置隊列的個數，以及接口的配置信息，如隊列的使用模式，多隊列的方式等。
前面會先進行一些各項檢查，若是設備已經啓動，就得先停下來才能配置（這時應該叫再配置吧）。而後把傳進去的配置參數拷貝到設備的數據區。

memcpy(&dev->data->dev_conf, dev_conf, sizeof(dev->data->dev_conf));

以後獲取設備的信息，主要也是爲了後面的檢查使用：

(*dev->dev_ops->dev_infos_get)(dev, &dev_info);

這裏的dev_infos_get是在驅動初始化過程當中設備初始化時配置的(eth_ixgbe_dev_init())

eth_dev->dev_ops = &ixgbe_eth_dev_ops;

重要的信息檢查事後，下面就是對發送和接收隊列進行配置
先看接收隊列的配置，接收隊列是從rte_eth_dev_tx_queue_config()開始的

在接收配置中，考慮的是有兩種狀況，一種是第一次配置；另外一種是從新配置。因此，代碼中都作了區分。
(1)若是是第一次配置，那麼就爲每一個隊列分配一個指針。
(2)若是是從新配置，配置的queue數量不爲0，那麼就取消以前的配置，從新配置。
(3)若是是從新配置，但要求的queue爲0，那麼釋放已有的配置。

發送的配置也是一樣的，在rte_eth_dev_tx_queue_config()。
當收發隊列配置完成後，就調用設備的配置函數，進行最後的配置。

(*dev->dev_ops->dev_configure)(dev)

咱們找到對應的配置函數(以ixgbe爲例)，進入

ixgbe_dev_configure()

來分析其過程,其實這個函數並無作太多的事。
在函數中，先調用了ixgbe_check_mq_mode()來檢查隊列的模式。而後設置容許接收批量和向量的模式

adapter->rx_bulk_alloc_allowed = true;
adapter->rx_vec_allowed = true;

接下來就是收發隊列的初始化，很是關鍵的一部份內容，這部份內容按照收發分別介紹：

接收隊列的初始化

接收隊列的初始化是從rte_eth_rx_queue_setup()開始的，這裏的參數須要指定要初始化的port_id,queue_id,以及描述符的個數，還能夠指定接收的配置，如釋放和回寫的閾值等。

依然如其餘函數的套路同樣，先進行各類檢查，如初始化的隊列號是否合法有效，設備若是已經啓動，就不能繼續初始化了。檢查函數指針是否有效等。檢查mbuf的數據大小是否知足默認的設備信息裏的配置。

rte_eth_dev_info_get(port_id, &dev_info);

這裏獲取了設備的配置信息，若是調用初始化函數時沒有指定rx_conf配置，就會設備配置信息裏的默認值

dev_info->default_rxconf = (struct rte_eth_rxconf) {
        .rx_thresh = {
            .pthresh = IXGBE_DEFAULT_RX_PTHRESH,
            .hthresh = IXGBE_DEFAULT_RX_HTHRESH,
            .wthresh = IXGBE_DEFAULT_RX_WTHRESH,
        },
        .rx_free_thresh = IXGBE_DEFAULT_RX_FREE_THRESH,
        .rx_drop_en = 0,
    };

還檢查了要初始化的隊列號對應的隊列指針是否爲空，若是不爲空，則說明這個隊列已經初始化過了，就釋放這個隊列。

rxq = dev->data->rx_queues;
    if (rxq[rx_queue_id]) {
        RTE_FUNC_PTR_OR_ERR_RET(*dev->dev_ops->rx_queue_release,
                    -ENOTSUP);
        (*dev->dev_ops->rx_queue_release)(rxq[rx_queue_id]);
        rxq[rx_queue_id] = NULL;
    }

最後，調用到隊列的setup函數作最後的初始化。

ret = (*dev->dev_ops->rx_queue_setup)(dev, rx_queue_id, nb_rx_desc,
                          socket_id, rx_conf, mp);

對於ixgbe設備，rx_queue_setup就是函數ixgbe_dev_rx_queue_setup()，這裏就是隊列最終的初始化咯

依然是先檢查，檢查描述符的數量最大不能大於IXGBE_MAX_RING_DESC個，最小不能小於IXGBE_MIN_RING_DESC個。

接下來的都是重點咯：
<1>.分配隊列結構體，並填充結構

rxq = rte_zmalloc_socket("ethdev RX queue", sizeof(struct ixgbe_rx_queue),
                 RTE_CACHE_LINE_SIZE, socket_id);

填充結構體的所屬內存池，描述符個數，隊列號，隊列所屬接口號等成員。

<2>.分配描述符隊列的空間，按照最大的描述符個數進行分配

rz = rte_eth_dma_zone_reserve(dev, "rx_ring", queue_idx,
                      RX_RING_SZ, IXGBE_ALIGN, socket_id);

接着獲取描述符隊列的頭和尾寄存器的地址，在收發包後，軟件要對這個寄存器進行處理。

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));

設置隊列的接收描述符ring的物理地址和虛擬地址。

rxq->rx_ring_phys_addr = rte_mem_phy2mch(rz->memseg_id, rz->phys_addr);
    rxq->rx_ring = (union ixgbe_adv_rx_desc *) rz->addr;

<3>分配sw_ring，這個ring中存儲的對象是struct ixgbe_rx_entry，其實裏面就是數據包mbuf的指針。

rxq->sw_ring = rte_zmalloc_socket("rxq->sw_ring",
                      sizeof(struct ixgbe_rx_entry) * len,
                      RTE_CACHE_LINE_SIZE, socket_id);

以上三步作完之後，新分配的隊列結構體重要的部分就已經填充完了，下面須要重置一下其餘成員

ixgbe_reset_rx_queue()

先把分配的描述符隊列清空，其實清空在分配的時候就已經作了，不必重複作

for (i = 0; i < len; i++) {
        rxq->rx_ring[i] = zeroed_desc;
    }

而後初始化隊列中一下其餘成員

rxq->rx_nb_avail = 0;
rxq->rx_next_avail = 0;
rxq->rx_free_trigger = (uint16_t)(rxq->rx_free_thresh - 1);
rxq->rx_tail = 0;
rxq->nb_rx_hold = 0;
rxq->pkt_first_seg = NULL;
rxq->pkt_last_seg = NULL;

這樣，接收隊列就初始化完了。

發送隊列的初始化

發送隊列的初始化在前面的檢查基本和接收隊列同樣，只有些許區別在於setup環節，咱們就從這個函數提及：ixgbe_dev_tx_queue_setup()
在發送隊列配置中，重點設置了tx_rs_thresh和tx_free_thresh的值。
而後分配了一個發送隊列結構txq,以後分配發送隊列ring的空間，並填充txq的結構體

txq->tx_ring_phys_addr = rte_mem_phy2mch(tz->memseg_id, tz->phys_addr);
    txq->tx_ring = (union ixgbe_adv_tx_desc *) tz->addr;

而後，分配隊列的sw_ring,也掛載隊列上。
重置發送隊列

ixgbe_reset_tx_queue()

和接收隊列同樣，也是要把隊列ring（描述符ring）清空，設置發送隊列sw_ring,設置其餘參數，隊尾位置設置爲0

txq->tx_next_dd = (uint16_t)(txq->tx_rs_thresh - 1);
txq->tx_next_rs = (uint16_t)(txq->tx_rs_thresh - 1);

txq->tx_tail = 0;
txq->nb_tx_used = 0;
/*
 * Always allow 1 descriptor to be un-allocated to avoid
 * a H/W race condition
 */
txq->last_desc_cleaned = (uint16_t)(txq->nb_tx_desc - 1);
txq->nb_tx_free = (uint16_t)(txq->nb_tx_desc - 1);
txq->ctx_curr = 0;

發送隊列的初始化就完成了。

設備的啓動

通過上面的隊列初始化，隊列的ring和sw_ring都分配了，可是發現木有，DMA仍然還不知道要把數據包拷貝到哪裏，咱們說過，DPDK是零拷貝的，那麼咱們分配的mempool中的對象怎麼和隊列以及驅動聯繫起來呢？接下來就是最精彩的時刻了----創建mempool、queue、DMA、ring之間的關係。話說，這個爲何不是在隊列的初始化中就作呢？
設備的啓動是從 rte_eth_dev_start() 中開始的

diag = (*dev->dev_ops->dev_start)(dev);

進而，找到設備啓動的真正啓動函數：ixgbe_dev_start()
先檢查設備的鏈路設置，暫時不支持半雙工和固定速率的模式。看來是暫時只有自適應模式咯。

而後把中斷禁掉，同時，停掉適配器

ixgbe_stop_adapter(hw);

在其中，就是調用了ixgbe_stop_adapter_generic();，主要的工做就是中止發送和接收單元。這是直接寫寄存器來完成的。

而後重啓硬件，ixgbe_pf_reset_hw()->ixgbe_reset_hw()->ixgbe_reset_hw_82599(),最終都是設置寄存器，這裏就不細究了。以後，就啓動了硬件。

再而後是初始化接收單元：ixgbe_dev_rx_init()

在這個函數中，主要就是設置各種寄存器，好比配置CRC校驗，若是支持巨幀，配置對應的寄存器。還有若是配置了loopback模式，也要配置寄存器。

接下來最重要的就是爲每一個隊列設置DMA寄存器，標識每一個隊列的描述符ring的地址，長度，頭，尾等。

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));
IXGBE_WRITE_REG(hw, IXGBE_RDLEN(rxq->reg_idx),
        rxq->nb_rx_desc * sizeof(union ixgbe_adv_rx_desc));
IXGBE_WRITE_REG(hw, IXGBE_RDH(rxq->reg_idx), 0);
IXGBE_WRITE_REG(hw, IXGBE_RDT(rxq->reg_idx), 0);

這裏能夠看到把描述符ring的物理地址寫入了寄存器，還寫入了描述符ring的長度。

下面還計算了數據包數據的長度，寫入到寄存器中.而後對於網卡的多隊列設置,也進行了配置

ixgbe_dev_mq_rx_configure()

同時若是設置了接收校驗和，還對校驗和進行了寄存器設置。

最後，調用ixgbe_set_rx_function()對接收函數再進行設置，主要是針對支持LRO，vector,bulk等處理方法。
這樣，接收單元的初始化就完成了。

接下來再初始化發送單元：ixgbe_dev_tx_init()
發送單元的的初始化和接收單元的初始化基本操做是同樣的，都是填充寄存器的值，重點是設置描述符隊列的基地址和長度。

bus_addr = txq->tx_ring_phys_addr;
IXGBE_WRITE_REG(hw, IXGBE_TDBAL(txq->reg_idx),
        (uint32_t)(bus_addr & 0x00000000ffffffffULL));
IXGBE_WRITE_REG(hw, IXGBE_TDBAH(txq->reg_idx),
        (uint32_t)(bus_addr 右移 32));
IXGBE_WRITE_REG(hw, IXGBE_TDLEN(txq->reg_idx),
        txq->nb_tx_desc * sizeof(union ixgbe_adv_tx_desc));
/* Setup the HW Tx Head and TX Tail descriptor pointers */
IXGBE_WRITE_REG(hw, IXGBE_TDH(txq->reg_idx), 0);
IXGBE_WRITE_REG(hw, IXGBE_TDT(txq->reg_idx), 0);

最後配置一下多隊列使用相關的寄存器：ixgbe_dev_mq_tx_configure()
如此，發送單元的初始化就完成了。
收發單元初始化完畢後，就能夠啓動設備的收發單元咯：ixgbe_dev_rxtx_start()
先對每一個發送隊列的threshold相關寄存器進行設置，這是發送時的閾值參數，這個東西在發送部分有說明。

而後就是依次啓動每一個接收隊列啦！

ixgbe_dev_rx_queue_start()

先檢查，若是要啓動的隊列是合法的，那麼就爲這個接收隊列分配存放mbuf的實際空間，

if (ixgbe_alloc_rx_queue_mbufs(rxq) != 0) 
{
    PMD_INIT_LOG(ERR, "Could not alloc mbuf for queue:%d",
             rx_queue_id);
    return -1;
}

在這裏，你將找到終極答案--mempool、ring、queue ring、queue sw_ring的關係！

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;
        struct rte_mbuf *mbuf = rte_mbuf_raw_alloc(rxq->mb_pool);

        if (mbuf == NULL) {
            PMD_INIT_LOG(ERR, "RX mbuf alloc failed queue_id=%u",
                     (unsigned) rxq->queue_id);
            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;

        dma_addr =
            rte_cpu_to_le_64(rte_mbuf_data_dma_addr_default(mbuf));
        rxd = &rxq->rx_ring[i];
        rxd->read.hdr_addr = 0;
        rxd->read.pkt_addr = dma_addr;
        rxe[i].mbuf = mbuf;
    }

    return 0;
}

看啊，真理就這麼赤果果的在眼前啦，我都不知道該說些什麼了！但仍是得說點什麼呀，否則就能夠結束本文啦！
咱們看到，從隊列所屬內存池的ring中循環取出了nb_rx_desc個mbuf指針，也就是爲了填充rxq->sw_ring。每一個指針都指向內存池裏的一個數據包空間。

而後就先填充了新分配的mbuf結構，最最重要的是填充計算了dma_addr

dma_addr = rte_cpu_to_le_64(rte_mbuf_data_dma_addr_default(mbuf));

而後初始化queue ring，即rxd的信息，標明瞭驅動把數據包放在dma_addr處。最後一句，把分配的mbuf「放入」queue 的sw_ring中，這樣，驅動收過來的包，就直接放在了sw_ring中。

以上最重要的工做就完成了，下面就可使能DMA引擎啦，準備收包。

hw->mac.ops.enable_rx_dma(hw, rxctrl);

而後再設置一下隊列ring的頭尾寄存器的值，這也是很重要的一點！頭設置爲0，尾設置爲描述符個數減1，就是描述符填滿整個ring。

IXGBE_WRITE_REG(hw, IXGBE_RDH(rxq->reg_idx), 0);
IXGBE_WRITE_REG(hw, IXGBE_RDT(rxq->reg_idx), rxq->nb_rx_desc - 1);

隨着這步作完，剩餘的就沒有什麼重要的事啦，就此打住！

接着依次啓動每一個發送隊列：

發送隊列的啓動比接收隊列的啓動要簡單，只是配置了txdctl寄存器，延時等待TX使能完成，最後，設置隊列的頭和尾位置都爲0。

txdctl = IXGBE_READ_REG(hw, IXGBE_TXDCTL(txq->reg_idx));
txdctl |= IXGBE_TXDCTL_ENABLE;
IXGBE_WRITE_REG(hw, IXGBE_TXDCTL(txq->reg_idx), txdctl);

IXGBE_WRITE_REG(hw, IXGBE_TDH(txq->reg_idx), 0);
IXGBE_WRITE_REG(hw, IXGBE_TDT(txq->reg_idx), 0);

發送隊列就啓動完成了。

3、數據包的獲取和發送

數據包的獲取是指驅動把數據包放入了內存中，上層應用從隊列中去取出這些數據包；發送是指把要發送的數據包放入到發送隊列中，爲實際發送作準備。

數據包的獲取

業務層面獲取數據包是從rte_eth_rx_burst()開始的

int16_t nb_rx = (*dev->rx_pkt_burst)(dev->data->rx_queues[queue_id],
            rx_pkts, nb_pkts);

這裏的dev->rx_pkt_burst在驅動初始化的時候已經註冊過了，對於ixgbe設備，就是ixgbe_recv_pkts()函數。
在說收包以前，先了解網卡的DD標誌，這個標誌標識着一個描述符是否可用的狀況：網卡在使用這個描述符前，先檢查DD位是否爲0，若是爲0，那麼就可使用描述符，把數據拷貝到描述符指定的地址，以後把DD標誌位置爲1，不然表示不能使用這個描述符。而對於驅動而言，偏偏相反，在讀取數據包時，先檢查DD位是否爲1，若是爲1，表示網卡已經把數據放到了內存中，能夠讀取，讀取完後，再把DD位設置爲0，不然，就表示沒有數據包可讀。

就重點從這個函數看看，數據包是怎麼被取出來的。
首先，取值rx_id = rxq->rx_tail,這個值初始化時爲0，用來標識當前ring的尾。而後循環讀取請求數量的描述符，這時候第一步判斷就是這個描述符是否可用

staterr = rxdp->wb.upper.status_error;
if (!(staterr & rte_cpu_to_le_32(IXGBE_RXDADV_STAT_DD)))
    break;

若是描述符的DD位不爲1，則代表這個描述符網卡尚未準備好，也就是沒有包！沒有包，就跳出循環。

若是描述符準備好了，就取出對應的描述符，由於網卡已經把一些信息存到了描述符裏，能夠後面把這些信息填充到新分配的數據包裏。

下面就是一個狸貓換太子的事了，先從mempool的ring中分配一個新的「狸貓」---mbuf

nmb = rte_mbuf_raw_alloc(rxq->mb_pool);

而後找到當前描述符對應的「太子」---ixgbe_rx_entry *rxe

rxe = &sw_ring[rx_id];

中間略掉關於預取的操做代碼，以後，就要用這個狸貓換個太子

rxm = rxe->mbuf;
rxe->mbuf = nmb;

這樣換出來的太子rxm就是咱們要取出來的數據包指針，在下面填充一些必要的信息，就能夠把包返給接收的用戶了

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_PKT set in pkt_flags */
rxm->vlan_tci = rte_le_to_cpu_16(rxd.wb.upper.vlan);

pkt_flags = rx_desc_status_to_pkt_flags(staterr, vlan_flags);
pkt_flags = pkt_flags | rx_desc_error_to_pkt_flags(staterr);
pkt_flags = pkt_flags |
    ixgbe_rxd_pkt_info_to_pkt_flags((uint16_t)pkt_info);
rxm->ol_flags = pkt_flags;
rxm->packet_type =
    ixgbe_rxd_pkt_info_to_pkt_type(pkt_info,
                       rxq->pkt_type_mask);

if (likely(pkt_flags & PKT_RX_RSS_HASH))
    rxm->hash.rss = rte_le_to_cpu_32(
                rxd.wb.lower.hi_dword.rss);
else if (pkt_flags & PKT_RX_FDIR) {
    rxm->hash.fdir.hash = rte_le_to_cpu_16(
            rxd.wb.lower.hi_dword.csum_ip.csum) &
            IXGBE_ATR_HASH_MASK;
    rxm->hash.fdir.id = rte_le_to_cpu_16(
            rxd.wb.lower.hi_dword.csum_ip.ip_id);
}
/*
 * Store the mbuf address into the next entry of the array
 * of returned packets.
 */
rx_pkts[nb_rx++] = rxm;

注意最後一句話，就是把包的指針返回給用戶。

其實在換太子中間過程當中，還有一件很是重要的事要作，就是開頭說的，在驅動讀取完數據包後，要把描述符的DD標誌位置爲0，同時設置新的DMA地址指向新的mbuf空間，這麼描述符就能夠再次被網卡硬件使用，拷貝數據到mbuf空間了。

dma_addr = rte_cpu_to_le_64(rte_mbuf_data_dma_addr_default(nmb));
rxdp->read.hdr_addr = 0;
rxdp->read.pkt_addr = dma_addr;

rxdp->read.hdr_addr = 0;
```一句中，就包含了設置DD位爲0。
最後，就是檢查空餘可用描述符數量是否小於閥值，若是小於閥值，進行處理。不詳細說了。

這樣事後，收取數據包就完成啦！Done！

### 數據包的發送

在說發送以前，先說一下描述符的回寫（write-back），回寫是指把用事後的描述符，恢復其從新使用的過程。在接收數據包過程當中，回寫是立馬執行的，也就是DMA使用描述符標識包可讀取，而後驅動程序讀取數據包，讀取以後，就會把DD位置0，同時進行回寫操做，這個描述符也就能夠再次被網卡硬件使用了。
可是發送過程當中，回寫卻不是馬上完成的。發送有兩種方式進行回寫：

1.Updating by writing back into the Tx descriptor
2.Update by writing to the head pointer in system memory
第二種回寫方式貌似針對的網卡比較老，對於82599，使用第一種回寫方式。在下面三種狀況下，才能進行回寫操做：
1.TXDCTL[n].WTHRESH = 0 and a descriptor that has RS set is ready to be written
back.
2.TXDCTL[n].WTHRESH > 0 and TXDCTL[n].WTHRESH descriptors have accumulated.
3.TXDCTL[n].WTHRESH > 0 and the corresponding EITR counter has reached zero. The
timer expiration flushes any accumulated descriptors and sets an interrupt event(TXDW).
而在代碼中，發送隊列的初始化的時候，`ixgbe_dev_tx_queue_setup()`中

txq->pthresh = tx_conf->tx_thresh.pthresh;
txq->hthresh = tx_conf->tx_thresh.hthresh;
txq->wthresh = tx_conf->tx_thresh.wthresh;

pthresh,hthresh,wthresh的值，都是從tx_conf中配置的默認值，而tx_conf若是在咱們的應用中沒有賦值的話，就是採用的默認值：

dev_info->default_txconf = (struct rte_eth_txconf) {
.tx_thresh = {
.pthresh = IXGBE_DEFAULT_TX_PTHRESH,
.hthresh = IXGBE_DEFAULT_TX_HTHRESH,
.wthresh = IXGBE_DEFAULT_TX_WTHRESH,
},
.tx_free_thresh = IXGBE_DEFAULT_TX_FREE_THRESH,
.tx_rs_thresh = IXGBE_DEFAULT_TX_RSBIT_THRESH,
.txq_flags = ETH_TXQ_FLAGS_NOMULTSEGS |
ETH_TXQ_FLAGS_NOOFFLOADS,
};

其中的wthresh就是0，其他兩個是32.也就是說這種設置下，回寫取決於RS標誌位。RS標誌位主要就是爲了標識已經積累了必定數量的描述符，要進行回寫了。

瞭解了這個，就來看看代碼吧，從`ixgbe_xmit_pkts()`開始，爲了看主要的框架，咱們忽略掉網卡卸載等相關的功能的代碼，主要看發送和回寫
先檢查剩餘的描述符是否已經小於閾值，若是小於閾值，那麼就先清理回收一下描述符

if (txq->nb_tx_free < txq->tx_free_thresh)
ixgbe_xmit_cleanup(txq);

這是一個重要的操做，進去看看是怎麼清理回收的：`ixgbe_xmit_cleanup(txq)`
取出上次清理的描述符位置，很明顯，此次清理就接着上次的位置開始。因此，根據上次的位置，加上`txq->tx_rs_thresh`個描述符，就是標記有RS的描述符的位置，由於，tx_rs_thresh就是表示這麼多個描述符後，設置RS（report state）位，進行回寫。因此，從上次清理的位置跳過tx_rs_thresh個描述符，就能找到標記有RS的位置。

desc_to_clean_to = (uint16_t)(last_desc_cleaned + txq->tx_rs_thresh);

當網卡把隊列的數據包發送完成後，就會把DD位設置爲1，這個時候，先檢查標記RS位置的描述符DD位，若是已經設置爲1，則能夠進行清理回收，不然，就不能清理。

接下來確認要清理的描述符個數

if (last_desc_cleaned > desc_to_clean_to)
nb_tx_to_clean = (uint16_t)((nb_tx_desc - last_desc_cleaned) +
desc_to_clean_to);
else
nb_tx_to_clean = (uint16_t)(desc_to_clean_to -
last_desc_cleaned);

而後，就把標記有RS位的描述符中的RS位清掉，確切的說，DD位等都清空了。調整上次清理的位置和空閒描述符大小。

txr[desc_to_clean_to].wb.status = 0;

/ Update the txq to reflect the last descriptor that was cleaned /
txq->last_desc_cleaned = desc_to_clean_to;
txq->nb_tx_free = (uint16_t)(txq->nb_tx_free + nb_tx_to_clean);

這樣，就算清理完畢了！

繼續看發送，依次處理每一個要發送的數據包：

取出數據包，取出其中的卸載標誌

ol_flags = tx_pkt->ol_flags;

/ If hardware offload required /
tx_ol_req = ol_flags & IXGBE_TX_OFFLOAD_MASK;
if (tx_ol_req) {
tx_offload.l2_len = tx_pkt->l2_len;
tx_offload.l3_len = tx_pkt->l3_len;
tx_offload.l4_len = tx_pkt->l4_len;
tx_offload.vlan_tci = tx_pkt->vlan_tci;
tx_offload.tso_segsz = tx_pkt->tso_segsz;
tx_offload.outer_l2_len = tx_pkt->outer_l2_len;
tx_offload.outer_l3_len = tx_pkt->outer_l3_len;

/* If new context need be built or reuse the exist ctx. */
ctx = what_advctx_update(txq, tx_ol_req,
    tx_offload);
/* Only allocate context descriptor if required*/
new_ctx = (ctx == IXGBE_CTX_NUM);
ctx = txq->ctx_curr;

}

這裏卸載還要使用一個描述符，暫時不明白。

計算了發送這個包須要的描述符數量，主要是有些大包會分紅幾個segment,每一個segment

nb_used = (uint16_t)(tx_pkt->nb_segs + new_ctx);

若是此次要用的數量加上設置RS以後積累的數量，又到達了tx_rs_thresh，那麼就設置RS標誌。

if (txp != NULL &&
nb_used + txq->nb_tx_used >= txq->tx_rs_thresh)
/ set RS on the previous packet in the burst /
txp->read.cmd_type_len |=
rte_cpu_to_le_32(IXGBE_TXD_CMD_RS);

接下來要確保用足夠可用的描述符

若是描述符不夠用了，就先進行清理回收，若是沒能清理出空間，則把最後一個打上RS標誌，更新隊列尾寄存器，返回已經發送的數量。

if (txp != NULL)
txp->read.cmd_type_len |= rte_cpu_to_le_32(IXGBE_TXD_CMD_RS);

rte_wmb();

/*
 * Set the Transmit Descriptor Tail (TDT)
 */
PMD_TX_LOG(DEBUG, "port_id=%u queue_id=%u tx_tail=%u nb_tx=%u",
       (unsigned) txq->port_id, (unsigned) txq->queue_id,
       (unsigned) tx_id, (unsigned) nb_tx);
IXGBE_PCI_REG_WRITE_RELAXED(txq->tdt_reg_addr, tx_id);
txq->tx_tail = tx_id;

接下來的判斷就頗有意思了，

unlikely(nb_used > txq->tx_rs_thresh)

爲何說它奇怪呢？其實他本身都標明瞭unlikely,一個數據包會分爲N多segment,多於txq->tx_rs_thresh（默承認是32啊），但即便出現了這種狀況，也沒作更多的處理，只是說會影響性能，而後開始清理描述符，其實這跟描述符還剩多少沒有半毛錢關係，只是一個包占的描述符就超過了tx_rs_thresh,然而，並不見得是沒有描述符了。因此，這時候清理描述符意義不明。

下面的處理應該都是已經有充足的描述符了，若是卸載有標誌，就填充對應的值。不詳細說了。

而後，就把數據包放到發送隊列的sw_ring,並填充信息

m_seg = tx_pkt;
do {
txd = &txr[tx_id];
txn = &sw_ring[txe->next_id];
rte_prefetch0(&txn->mbuf->pool);

if (txe->mbuf != NULL)
        rte_pktmbuf_free_seg(txe->mbuf);
    txe->mbuf = m_seg;

    /*
     * Set up Transmit Data Descriptor.
     */
    slen = m_seg->data_len;
    buf_dma_addr = rte_mbuf_data_dma_addr(m_seg);
    txd->read.buffer_addr =
        rte_cpu_to_le_64(buf_dma_addr);
    txd->read.cmd_type_len =
        rte_cpu_to_le_32(cmd_type_len | slen);
    txd->read.olinfo_status =
        rte_cpu_to_le_32(olinfo_status);
    txe->last_id = tx_last;
    tx_id = txe->next_id;
    txe = txn;
    m_seg = m_seg->next;
} while (m_seg != NULL);

這裏是把數據包的每一個segment都放到隊列sw_ring，很關鍵的是設置DMA地址，設置數據包長度和卸載參數。

一個數據包最後的segment的描述符須要一個EOP標誌來結束。再更新剩餘的描述符數：

cmd_type_len |= IXGBE_TXD_CMD_EOP;
txq->nb_tx_used = (uint16_t)(txq->nb_tx_used + nb_used);
txq->nb_tx_free = (uint16_t)(txq->nb_tx_free - nb_used);

而後再次檢查是否已經達到了tx_rs_thresh，並作處理

if (txq->nb_tx_used >= txq->tx_rs_thresh) {
PMD_TX_FREE_LOG(DEBUG,
"Setting RS bit on TXD id="
"%4u (port=%d queue=%d)",
tx_last, txq->port_id, txq->queue_id);

cmd_type_len |= IXGBE_TXD_CMD_RS;

/* Update txq RS bit counters */
txq->nb_tx_used = 0;
txp = NULL;

} else
txp = txd;

txd->read.cmd_type_len |= rte_cpu_to_le_32(cmd_type_len);

最後還是作一下末尾的處理，更新隊列尾指針。發送就結束啦！！

IXGBE_PCI_REG_WRITE_RELAXED(txq->tdt_reg_addr, tx_id);
txq->tx_tail = tx_id;

## 總結
能夠看出數據包的發送和接收過程與驅動緊密相關，也與咱們的配置有關，尤爲是對於收發隊列的參數配置，將直接影響性能，能夠根據實際進行調整。對於收發虛擬化的部分，此文並未涉及，待後續有機會補充完整。

參考dpdk17.08源碼閱讀

此文轉載自：https://www.cnblogs.com/yhp-smarthome/p/6705638.html