1. 程式人生 > >DPDK收發包全景分析

DPDK收發包全景分析

前言:DPDK收發包是基礎核心模組,從網絡卡收到包到驅動把包拷貝到系統記憶體中,再到系統對這塊資料包的記憶體管理,由於在處理過程中實現了零拷貝,資料包從接收到傳送始終只有一份,對這個報文的管理在前面的mempool記憶體池中有過介紹。這篇主要介紹收發包的過程。

一、收發包分解

收發包過程大致可以分為2個部分

  • 1.收發包的配置和初始化,主要是配置收發佇列等。
  • 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_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_threshtx_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);

傳送佇列就啟動完成了。

三、資料包的獲取和傳送

資料包的獲取是指驅動把資料包放入了記憶體中,上層應用從佇列中去取出這些資料包;傳送是指把要傳送的資料包放入到傳送佇列中,為實際傳送做準備。

資料包的獲取

業務層面獲取資料包是從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位,進行回寫。所以,從上次清理的位置跳過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;

總結:

可以看出資料包的傳送和接收過程與驅動緊密相關,也與我們的配置有關,尤其是對於收發佇列的引數配置,將直接影響效能,可以根據實際進行調整。對於收發虛擬化的部分,此文並未涉及,待後續有機會補充完整。