intel dpdk的Poll Model Driver機制簡介

        最近基於intell的dpdk框架開發高性能的Dns服務器，纔開始接觸dpdk。 

        Linux內核的網絡協議棧是基於中斷的，網卡收到數據包後發出中斷信號，CPU中止當前的處理工做，進入中斷環境，執行相應網卡驅動的中斷處理程序。 這種處理方式適合通用的處理平臺，使CPU能夠執行多樣化的任務。而對於一些專門處理網絡事務的設備，不少都配有1G, 10G, 甚至40G網卡，高負載時，網卡中斷很是頻繁，每次處理網卡中斷都要涉及進程上下文的切換，花費不少額外的時間，這時便有了輪詢的處理方法，由CPU主動收包，更適合大流量的處理。

        這裏使用的是dpdk-16.07,  intell 82599網卡，ixgbe驅動。

DMA環形緩衝區

        intell 82599網卡最多支持128的Rx隊列，設備初始化時能夠設置啓用的隊列數量，對每個隊列都會分配一個DMA緩衝區隊列，大小爲256。隊列的每個元素是一個結構體，存放一塊預分配的內存的地址等信息。 

        下面是intell 官方文檔中關於接收隊列的描述。Head指向下一個空閒的元素，Tail指向已使用的元素，Head和Tail之間藍色區域爲空閒的元素。收到一個數據包後Head逆時針移動一格， Head等於Tail時沒有空閒的元素。

網卡接收網絡包流程

        網口收到數據包 -> 經過設定的規則肯定發送到哪一個隊列 -> 從DMA環形緩衝區尋找空閒的預分配內存--> 經過DMA複製數據包到內存。

        若是已經設定好了DMA環形緩衝區，那麼從收到網絡包到複製到內存，不須要CPU的參與，網卡能夠自動完成。 數據包複製到內存後即可以被CPU訪問，若是採用中斷的方式，這時會由設備產生中斷信號，CPU執行中斷處理程序，進入內核的網絡協議棧處理。這裏採用PMD輪詢，不會產生中斷信號。

PMD機制

       進程中主動調用rte_eth_rx_burst函數收取數據包，最多收取nb_pkts個包，每一個數據包都放在ret_mbuf的結構體中，這時數據包已經在內存中了，這裏的收包只是把相應的rte_mbuf的指針放在rx_pkts中，返回收到包的個數。

static inline uint16_t
rte_eth_rx_burst(uint8_t port_id, uint16_t queue_id,
		 struct rte_mbuf **rx_pkts, const uint16_t nb_pkts)
{
	struct rte_eth_dev *dev = &rte_eth_devices[port_id];

    /*****此處省略****/

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

    /*****此處省略****/

	return nb_rx;
}

實際調用的收包函數是dev->rx_pkt_burst，每種網卡都不相同，即便同一種網卡也會由於intell針對CPU的優化而採用不一樣的處理。這裏爲了簡單以rx_recv_pkts爲例。

• 調用ixgbe_rx_scan_hw_ring， 把須要收取的包放在rxq的rx_stage中。
• 若是rxq->rx_tail > rxq_rx_free_trigger，代表收取了數據包，再分配nb_rx個rte_mbuf結構補充進DMA環形緩衝區中。
• 調用ixgbe_rx_fill_from_stage將rxq的rx_stage中保存的收取的包放在rx_pkts中返回。

static inline uint16_t
rx_recv_pkts(void *rx_queue, struct rte_mbuf **rx_pkts,
	     uint16_t nb_pkts)
{
	struct ixgbe_rx_queue *rxq = (struct ixgbe_rx_queue *)rx_queue;
	uint16_t nb_rx = 0;

	/* Any previously recv'd pkts will be returned from the Rx stage */
	if (rxq->rx_nb_avail)
		return ixgbe_rx_fill_from_stage(rxq, rx_pkts, nb_pkts);

	/* Scan the H/W ring for packets to receive */
	nb_rx = (uint16_t)ixgbe_rx_scan_hw_ring(rxq);

	/* update internal queue state */
	rxq->rx_next_avail = 0;
	rxq->rx_nb_avail = nb_rx;
	rxq->rx_tail = (uint16_t)(rxq->rx_tail + nb_rx);

	/* if required, allocate new buffers to replenish descriptors */
	if (rxq->rx_tail > rxq->rx_free_trigger) {
		uint16_t cur_free_trigger = rxq->rx_free_trigger;

		if (ixgbe_rx_alloc_bufs(rxq, true) != 0) {
			int i, j;
			PMD_RX_LOG(DEBUG, "RX mbuf alloc failed port_id=%u "
				   "queue_id=%u", (unsigned) rxq->port_id,
				   (unsigned) rxq->queue_id);

			rte_eth_devices[rxq->port_id].data->rx_mbuf_alloc_failed +=
				rxq->rx_free_thresh;

			/*
			 * Need to rewind any previous receives if we cannot
			 * allocate new buffers to replenish the old ones.
			 */
			rxq->rx_nb_avail = 0;
			rxq->rx_tail = (uint16_t)(rxq->rx_tail - nb_rx);
			for (i = 0, j = rxq->rx_tail; i < nb_rx; ++i, ++j)
				rxq->sw_ring[j].mbuf = rxq->rx_stage[i];

			return 0;
		}

		/* update tail pointer */
		rte_wmb();
		IXGBE_PCI_REG_WRITE(rxq->rdt_reg_addr, cur_free_trigger);
	}

	if (rxq->rx_tail >= rxq->nb_rx_desc)
		rxq->rx_tail = 0;

	/* received any packets this loop? */
	if (rxq->rx_nb_avail)
		return ixgbe_rx_fill_from_stage(rxq, rx_pkts, nb_pkts);

	return 0;
}

        這樣CPU與網卡分工合做， CPU不斷接收數據包而後處理， 網卡不斷接收網絡包而後複製到內存，構成一個生產者-消費者的模型，共同完成處理任務。