前言

上一篇博客我們分析了普通的中斷方式接收數據包的流程。并且在分析流程和代碼的過程中看到napi的入口函數(napi_schedule)。當時提到了該函數，大家應該還有印象。我們這里再次把代碼截取一份，來回顧一下調用的地方。

__IRAM_GEN static inline void rtl_rx_interrupt_process(unsigned int status, struct dev_priv *cp) {#if defined(REINIT_SWITCH_CORE)if(rtl865x_duringReInitSwtichCore==1) {return;}#endif#ifdef CONFIG_RTL_8197Fif (status & (RX_DONE_IP_ALL | PKTHDR_DESC_RUNOUT_IP_ALL)) #else//if (status & (RX_DONE_IP_ALL | PKTHDR_DESC_RUNOUT_IP_ALL)) #endif{ #if defined(CONFIG_RTL_ETH_NAPI_SUPPORT)if ((rtl_rx_tasklet_running==0)&&(cp->napi.poll)) {rtl_rx_tasklet_running=1;REG32(CPUIIMR) &= ~(RX_DONE_IE_ALL | PKTHDR_DESC_RUNOUT_IE_ALL);rtl_rxSetTxDone(FALSE);napi_schedule(&cp->napi);} #else#if defined(RX_TASKLET)if (rtl_rx_tasklet_running==0) {rtl_rx_tasklet_running=1;REG32(CPUIIMR) &= ~(RX_DONE_IE_ALL | PKTHDR_DESC_RUNOUT_IE_ALL);rtl_rxSetTxDone(FALSE);tasklet_hi_schedule(&cp->rx_dsr_tasklet);}#elseinterrupt_dsr_rx((unsigned long)cp);#endif #endif} }

數據包在中斷下半部即tasklet中有兩個分支。分別是普通接收方式和NAPI接收方式。我們來看看連一個分支，即napi_schedule函數的處理

根據上一篇博客，我們可以在這里來做一個總結。非NAPI的內核接口為netif_rx()，NAPI的內核接口為napi_schedule()。只不過在這里netif_rx被再次封裝了。

NAPI流程分析

napi主要涉及到3個函數：

• netif_napi_add(NAPI初始化函數，一般在網卡初始化的時候會調用)
• napi_schedule(該函數是napi的調度函數，一般在接受網卡數據包的時候調用-我們在上一篇博客中已經看到了該函數在接收中斷下半部被調用)
• napi_complete(該函數是一個狀態機。當數據包不是很多的時候，用來告訴內核，后續采用純中斷的方式進行接收)

下面我們就來結合實際的驅動程序和接收程序來分析這三個函數以及具體的流程。

這里涉及到了一個結構體，即struct napi_struct(該結構是NAPI的重要數據結構，對于理解NAPI的流程很有幫助)。

struct napi_struct {/* The poll_list must only be managed by the entity which* changes the state of the NAPI_STATE_SCHED bit. This means* whoever atomically sets that bit can add this napi_struct* to the per-cpu poll_list, and whoever clears that bit* can remove from the list right before clearing the bit.*/struct list_head poll_list;unsigned long state;int weight;unsigned int gro_count;int (*poll)(struct napi_struct *, int); #ifdef CONFIG_NETPOLLspinlock_t poll_lock;int poll_owner; #endifstruct net_device *dev;struct sk_buff *gro_list;struct sk_buff *skb;struct list_head dev_list; };

在這里我們來對該結構體一些重要的值進行說明。

poll_list： 這是設備列表，其中的設備就是在入口隊列中有新的幀等待被處理的設備。這些設備就是所謂的處于輪訓狀態。次列表的頭問softnet_data->poll_list。在此列表中的設備都處于中斷功能關閉的狀態，而內核當前正在輪訓。
2）weight：該字段代表每次調用poll函數的時候分配的最大的skb的數量。或者可以理解為從DMA中每次可以獲取的最大的skb的數量。對于NAPI程序而言，默認的值為64。但是也有16和32。它可以通過sysfs來調整。

poll：該函數指針代表的是輪訓的具體方法。

netif_napi_add函數

該函數是為了初始化struct napi_struct結構體。因為在后續的接收程序中會調用napi的接收程序，即napi_schedule。napi_schedule的參數就是struct napi_struct。

#if defined(CONFIG_RTL_ETH_NAPI_SUPPORT)netif_napi_add(dev, &cp->napi, rtl865x_poll, RTL865X_NAPI_WEIGHT);napi_enable(&cp->napi); #else #if defined(RX_TASKLET)tasklet_init(&cp->rx_dsr_tasklet, (void *)interrupt_dsr_rx, (unsigned long)cp); #endif

從上面的這段代碼中我們可以看到如果是支持NAPI的方式即調用netif_napi_add函數和napi_enable函數。說明:該代碼是網卡open函數中的代碼，正如上文所說，該函數一般在網卡初始化的時候調用。所以大家以后查找該函數就在網卡初始化代碼中查找即可

netif_napi_add(dev, &cp->napi, rtl865x_poll, RTL865X_NAPI_WEIGHT);

void netif_napi_add(struct net_device *dev, struct napi_struct *napi,int (*poll)(struct napi_struct *, int), int weight) {INIT_LIST_HEAD(&napi->poll_list);napi->gro_count = 0;napi->gro_list = NULL;napi->skb = NULL;napi->poll = poll;if (weight > NAPI_POLL_WEIGHT)pr_err_once("netif_napi_add() called with weight %d on device %s\n",weight, dev->name);napi->weight = weight;list_add(&napi->dev_list, &dev->napi_list);napi->dev = dev; #ifdef CONFIG_NETPOLLspin_lock_init(&napi->poll_lock);napi->poll_owner = -1; #endifset_bit(NAPI_STATE_SCHED, &napi->state); }

在調用該函數的時候傳遞了4個參數。分別是網卡對應的dev，和struct napi_struct，以及輪訓處理函數，和最大能夠處理的skb的個數。

說明:從上面的代碼我們可知，每一個網卡都有一個napi_struct結構。都需要注冊自己的處理函數(輪訓函數)，以及指定最大skb的個數(64)。#define RTL865X_NAPI_WEIGHT 64

napi_schedule函數

我們先來看看napi_schedule函數的源代碼

static inline void napi_schedule(struct napi_struct *n) {if (napi_schedule_prep(n))__napi_schedule(n); }

該函數調用napi_schedule_prep來判斷是否能夠運行napi。如果napi沒有被禁止且沒有被調度則調用__napi_schedule

void __napi_schedule(struct napi_struct *n) {unsigned long flags;local_irq_save(flags);____napi_schedule(&__get_cpu_var(softnet_data), n);local_irq_restore(flags); }

1）保存中斷
2）調用____napi_schedule函數，注意該函數的參數
3）恢復中斷

這里的____napi_sule函數是一個重點函數，可以看到napi方式接收程序底層最主要調用的就是該函數。

我們首先來分析一下__get_cpu_var(softnet_data)。該函數是從cup中獲取softnet_data結構體指針。獲取當前CPU上的待輪詢設備隊列。

softnet_data結構

每一個cpu都有隊列，用來接收進來的數據幀。因為每個cpu都有其數據結構來處理入口和出口的流量，因此，不同的cpu之間沒有必要使用上鎖機制。此隊列就是用sodtnet_data來表示

struct softnet_data {struct Qdisc *output_queue;struct Qdisc **output_queue_tailp;struct list_head poll_list;struct sk_buff *completion_queue;struct sk_buff_head process_queue;/* stats */unsigned int processed;unsigned int time_squeeze;unsigned int cpu_collision;unsigned int received_rps;#ifdef CONFIG_RPSstruct softnet_data *rps_ipi_list;/* Elements below can be accessed between CPUs for RPS */struct call_single_data csd ____cacheline_aligned_in_smp;struct softnet_data *rps_ipi_next;unsigned int cpu;unsigned int input_queue_head;unsigned int input_queue_tail; #endifunsigned int dropped;//下面兩個接口舊的接口，netif_rx 會把skb掛到input_packet_queue上struct sk_buff_head input_pkt_queue;struct napi_struct backlog; };

• output_queue和output_queue_tailp：代表的是隊列的策略，流控會使用到該數據結構（流控一般是在出口做流量控制）
• poll_list:是一個雙向列表，其中的設備都帶有輸入幀等著被處理
• input_pkt_queue：該隊列(在net_dev_init中初始化) 用來保存進來的數據幀(被驅動程序處理前)。非NAPI驅動程序也會使用此字段，那些還沒有更新為使用NAPI的則會使用其自己的私有隊列

那么這里衍生出另一個問題，該結構體在什么時候初始化的呢

static int __init net_dev_init(void) {int i, rc = -ENOMEM;BUG_ON(!dev_boot_phase);if (dev_proc_init())goto out;if (netdev_kobject_init())goto out;INIT_LIST_HEAD(&ptype_all);for (i = 0; i < PTYPE_HASH_SIZE; i++)INIT_LIST_HEAD(&ptype_base[i]);INIT_LIST_HEAD(&offload_base);if (register_pernet_subsys(&netdev_net_ops))goto out;/** Initialise the packet receive queues.*/for_each_possible_cpu(i) {struct softnet_data *sd = &per_cpu(softnet_data, i);memset(sd, 0, sizeof(*sd));skb_queue_head_init(&sd->input_pkt_queue);skb_queue_head_init(&sd->process_queue);sd->completion_queue = NULL;INIT_LIST_HEAD(&sd->poll_list);sd->output_queue = NULL;sd->output_queue_tailp = &sd->output_queue; #ifdef CONFIG_RPSsd->csd.func = rps_trigger_softirq;sd->csd.info = sd;sd->csd.flags = 0;sd->cpu = i; #endifsd->backlog.poll = process_backlog;sd->backlog.weight = weight_p;sd->backlog.gro_list = NULL;sd->backlog.gro_count = 0;}dev_boot_phase = 0;/* The loopback device is special if any other network devices* is present in a network namespace the loopback device must* be present. Since we now dynamically allocate and free the* loopback device ensure this invariant is maintained by* keeping the loopback device as the first device on the* list of network devices. Ensuring the loopback devices* is the first device that appears and the last network device* that disappears.*/if (register_pernet_device(&loopback_net_ops))goto out;if (register_pernet_device(&default_device_ops))goto out;open_softirq(NET_TX_SOFTIRQ, net_tx_action);open_softirq(NET_RX_SOFTIRQ, net_rx_action);hotcpu_notifier(dev_cpu_callback, 0);dst_init();rc = 0; out:return rc; }

在這里我們著重強調一下該函數。該函數是在內核加載的時候被調用的。所以不需要我們手動調用，驅動也不會去調用。在非NAPI的方式中，我們已經分析了一部分代碼。包括軟中斷的注冊。這里又進行了softnet_data結構的初始化。


接下來我們繼續分析____napi_schedule函數

static inline void ____napi_schedule(struct softnet_data *sd,struct napi_struct *napi) {list_add_tail(&napi->poll_list, &sd->poll_list);__raise_softirq_irqoff(NET_RX_SOFTIRQ); }

這里會觸發NET_RX_SOFTIRQ軟中斷。即調用到了net_rx_action函數。這個函數是不是很熟悉，沒錯該函數和我們上一篇博客講的非NAPI方式一樣了。

那么在上文注冊的rtl865x_poll函數什么時候調用呢.rtl865x_poll函數已經加入到了napi->poll = poll中.

我們再貼一次net_rx_action的代碼

static void net_rx_action(struct softirq_action *h) {struct softnet_data *sd = &__get_cpu_var(softnet_data);unsigned long time_limit = jiffies + 2;int budget = netdev_budget;void *have;local_irq_disable();while (!list_empty(&sd->poll_list)) {struct napi_struct *n;int work, weight;

可以看到從cpu從取出了softnet_data結構，然后調用了其中的poll_list函數。

我們按照上文繼續貼出流程圖

總結

非NAPI的方式和NAPI的方式在接收中斷函數中產生出分支，在net_rx_action又回歸到一條主線上。
歡迎大家加入qq群：610849576。個人知識有限，只有多討論，多思考，才可以進步，歡迎大家一起學習

• 下一講我們來講解連接跟蹤，然后基于連接跟蹤寫出自己的內核模塊，增加協議棧轉發速率(實測內網可以從200M提高的680M)。敬請期待

總結

以上是生活随笔為你收集整理的论网卡数据是如何从驱动到桥接/ip层(NAPI方式)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。