1. 引入問題
核心收包主要有兩種手段：輪詢和中斷。
通過輪詢，核心可以不斷持續的檢查裝置時候有包收上來，例如設定一個定時器，定期檢查裝置上的某個定時器。這種方法會輕易浪費掉很多系統資源。
如果采用中斷收包，當裝置收到包時，可以産生一個硬體中斷通知核心，核心将中斷其他活動，然後調用一個中斷處理程式以滿足裝置的需求，核心隻是将資料包放到某個隊列中并通知核心中的收包子產品。這種方式是非常常見的，在低流量負載下是很好的選擇，但是在高流量負載下就無法良好的運作，每接收一個幀就産生一個中斷，很快就會讓CPU為進行中斷而浪費所有的時間。
以太網驅動收包就是通過以太網裝置産生收包中斷通知核心來收包的，但是如上所述，不能每收一個幀都要産生一個中斷，下面的内容将介紹驅動中如何結合論需和中斷來收包。
2. 幾個關鍵函數
有必要先介紹幾個收包相關的函數，也許會對了解後面的内容有幫助。
2.1 netif_receive_skb()
該函數是核心收包的入口，驅動收到的資料包通過這個函數進入核心協定棧進行處理，我在這裡不會分析它的實作，隻要記住，接下來的幾種驅動收包方式最終都是為了将資料包送到這個函數。
2.2 net_rx_action()
收包軟中斷處理函數，即中斷下半部。中斷處理函數要求盡可能快的執行完成，核心為了快速響應中斷，在處理硬體中斷時，隻是将資料包放到CPU的某個隊列中去，并排程軟中斷。而實際的資料包處理過程則交給中斷下半部處理。
中斷下半部的處理可以通過軟中斷或tasklet來完成：
1. 軟中斷：核心中定義好了一個收包軟中斷處理函數net_rx_action()，後面會分析該函數。
open_softirq(NET_RX_SOFTIRQ,net_rx_action);
2. tasklet：使用tasklet_init(t, func, data)注冊你自己的下半部收包函數。
工作隊列也可以實作延期執行一個函數，但網絡代碼中主要使用的是軟中斷和tasklet，是以我們不考慮工作隊列。
2.3 dma_alloc_coherent()
函數原型為：
void *dma_alloc_coherent(struct device *dev, size_t size, dma_addr_t*dma_handle, gfp_t gfp);
該函數用于配置設定一個DMA一緻性緩沖區。大多數以太網裝置都支援DMA機制，裝置收到資料包後，DMA将其放入記憶體中，并産生一個收包中斷通知CPU從記憶體中拿走資料包。
DMA隻能識别實體位址，而OS是操作虛拟位址的，這就需要緩存資料包的區域可以讓DMA和OS都能操作。dma_alloc_coherent()函數就是為了達到這個目标，它配置設定size大小的一緻性記憶體，其實體起始位址存放在dma_handle中，函數傳回值為這段記憶體的虛拟起始位址，這樣，裝置向這塊位址放資料包，OS響應中斷後可以從這塊位址拿包。
而由于我們要配置設定一緻性記憶體（任何時候，cache中的内容和記憶體中的内容是相同的），是以傳回的虛拟位址盡量是非緩存的，例如在mips中這個虛拟位址就是KSEG1中的位址。
3. 舊的收包接口netif_rx
在裝置驅動在DMA中拿到一個資料包，做一些和裝置相關的處理後，初始化一個skb執行個體，就可以将資料包交給netif_rx()來處理了。
核心中定義了全局的per-cpu收包隊列softnet_data，其定義如下：
DEFINE_PER_CPU_ALIGNED(struct softnet_data, softnet_data);
EXPORT_PER_CPU_SYMBOL(softnet_data);
結構體struct softnet_data的定義：
/*
* Incoming packets are placed on per-cpuqueues
*/
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;
/* 統計資料 */
unsigned int processed;
unsigned int time_squeeze;
unsigned int cpu_collision;
unsigned int received_rps;
unsigned dropped; //被丢棄的包的數量
struct sk_buff_head input_pkt_queue; //收包隊列
struct napi_struct backlog; //處理積壓隊列的napi結構
};
對于收包來講，需要用到的成員已經給出注釋。struct softnet_data結構體中，poll_list是NAPI裝置清單。input_pkt_queue為per-cpu的收包隊列。backlog是預設的處理收包的napi裝置。
我們看到，NAPI的架構已經被整合到核心中，即使不使用NAPI機制，核心中也有其他地方使用napi_struct等相關結構。是以這裡需要先說明一下napi結構：
struct napi_struct {
/* 連結清單指針，用于挂在softnet_data上 */
struct list_head poll_list;
/* 此NAPI裝置目前的狀态 */
unsigned long state;
/* 一個權重值，每次排程NAPI可處理資料包個數的限制 */
int weight;
/* poll函數，用于實際來處理資料包 */
int (*poll)(structnapi_struct *, int);
……
};
netif_rx()函數中主要是調用enqueue_to_backlog()将skb放入per-cpu的收包隊列中去。
static intenqueue_to_backlog(struct sk_buff *skb, int cpu,
unsigned int *qtail)
{
struct softnet_data *sd;
unsigned long flags;
/* 獲得per-cpu的softnet_data結構。 */
sd = &per_cpu(softnet_data, cpu);
local_irq_save(flags);
rps_lock(sd);
/* 如果input_pkt_queue隊列的長度沒超出限制。 */
if(skb_queue_len(&sd->input_pkt_queue) <= netdev_max_backlog) {
/* 如果隊列中已經有資料包，則将資料包加入隊列。 */
if(skb_queue_len(&sd->input_pkt_queue)) {
enqueue:
__skb_queue_tail(&sd->input_pkt_queue,skb);
input_queue_tail_incr_save(sd, qtail);
rps_unlock(sd);
local_irq_restore(flags);
return NET_RX_SUCCESS;
}
/* Schedule NAPI for backlog device
*We can use non atomic operation since we own the queue lock
*/
/* 如果隊列中還沒有資料包，則先排程軟中斷，再将資料包加入隊列。 */
if (!__test_and_set_bit(NAPI_STATE_SCHED,&sd->backlog.state)) {
if (!rps_ipi_queued(sd))
/* 加入輪詢清單，排程NET_RX_SOFTIRQ */
____napi_schedule(sd,&sd->backlog);
}
goto enqueue;
}
/* 如果代碼走到這裡，說明input_pkt_queue隊列的長度超出限制。 */
sd->dropped++;
rps_unlock(sd);
local_irq_restore(flags);
atomic_long_inc(&skb->dev->rx_dropped);
kfree_skb(skb);
return NET_RX_DROP;
}
函數流程比較清晰：
1. 獲得per-cpu的softnet_data結構執行個體sd。
2. 如果sd->input_pkt_queue收包隊列長度超出限制，即收到的包積滿了，直接給sd->dropped++，并嘗試釋放skb，傳回NET_RX_DROP。這個限制的預設值為1000，可在/proc/sys/net/core/netdev_max_backlog中檢視和修改。
3. 如果sd->input_pkt_queue收包隊列長度未達到限制，則将skb放入sd->input_pkt_queue隊列中。這裡分兩種情況：
a) 如果sd->input_pkt_queue是空的，則将napi裝置sd->backlog添加到poll_list輪詢清單中去，并排程NET_RX_SOFTIRQ，同時将napi裝置sd->backlog的state設定為NAPI_STATE_SCHED。然後将skb加入到sd->input_pkt_queue的隊尾。
b) 如果sd->input_pkt_queue不是空的，則NET_RX_SOFTIRQ已經被排程過了，是以，直接将skb加入到sd->input_pkt_queue的隊尾即可。
這裡需要注意兩點：1. 把幀排入隊列是相當快的，因為不涉及任何記憶體拷貝，隻是指針操作而已。2. 在操作per-cpu變量softnet_data時，需要關閉本地中斷（netif_rx可能不是在中斷處理程式中被調用的，是以此時本地中斷可能是開啟的）。
____napi_schedule()函數用于将napi裝置添加到poll_list輪詢清單中，并排程NET_RX_SOFTIRQ。
static inline void ____napi_schedule(struct softnet_data *sd,
struct napi_struct *napi)
{
/* 添加到poll_list裝置輪詢清單 */
list_add_tail(&napi->poll_list,&sd->poll_list);
/* 排程NET_RX_SOFTIRQ */
__raise_softirq_irqoff(NET_RX_SOFTIRQ);
}
早在dev module初始化的時候，net_dev_init()中就定義了softnet_data中napi結構的poll函數以及軟中斷處理函數：
for_each_possible_cpu(i){
……
struct softnet_data *sd =&per_cpu(softnet_data, i);
sd->backlog.poll= process_backlog;
sd->backlog.weight = weight_p;
}
……
open_softirq(NET_RX_SOFTIRQ,net_rx_action);
排程了軟中斷，則後續會執行下半部函數net_rx_action()。
net_rx_action()是一個很重要的下半部收包函數，NAPI裝置和非NAPI裝置都可能會使用它來收包。該函數的主要工作就是操作收包隊列和執行poll函數。
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();
/* 如果sd->poll_list不為空則進行周遊，(每處理完裡面的一個napi_struct->poll_list，就将其删除)*/
while (!list_empty(&sd->poll_list)) {
struct napi_struct *n;
int work, weight;
/* If softirq window is exhuasted thenpunt.
*Allow this to run for 2 jiffies since which will allow
*an average latency of 1.5/HZ.
*/
if (unlikely(budget <= 0 ||time_after(jiffies, time_limit))) /* 強制本輪軟中斷收包結束 */
goto softnet_break;
local_irq_enable();
/* 軟中斷處理過程開中斷 */
/* Even though interrupts have beenre-enabled, this
*access is safe because interrupts can only add new
*entries to the tail of this list, and only ->poll()
*calls can remove this head entry from the list.
*/
n = list_first_entry(&sd->poll_list,struct napi_struct, poll_list); /* 獲得struct napi_struct執行個體*/
have = netpoll_poll_lock(n);
weight = n->weight;
/* This NAPI_STATE_SCHED test is foravoiding a race
*with netpoll's poll_napi(). Only theentity which
*obtains the lock and sees NAPI_STATE_SCHED set will
*actually make the ->poll() call. Therefore we avoid
*accidentally calling ->poll() when NAPI is not scheduled.
*/
work = 0;
/* 如果狀态為被排程，則調用poll函數進行實際的收包 */
if (test_bit(NAPI_STATE_SCHED,&n->state)) {
work= n->poll(n, weight);
trace_napi_poll(n);
}
WARN_ON_ONCE(work > weight);
/* 更新budget */
budget -= work;
local_irq_disable();
/* Drivers must not modify the NAPI stateif they
*consume the entire weight. In such casesthis code
*still "owns" the NAPI instance and therefore can
*move the instance around on the list at-will.
*/
if (unlikely(work == weight)) { /* 這時應該還有包沒有收完 */
if(unlikely(napi_disable_pending(n))) {
local_irq_enable();
napi_complete(n);
local_irq_disable();
} else
/* 将該NAPI執行個體後移到softnet_data的隊尾。*/
list_move_tail(&n->poll_list,&sd->poll_list);
}
netpoll_poll_unlock(have);
}
out:
net_rps_action_and_irq_enable(sd); /*local_irq_enable() */
return;
softnet_break:
sd->time_squeeze++; /* 用于proc */
/* 還有包沒收完，重新排程軟中斷來收包 */
__raise_softirq_irqoff(NET_RX_SOFTIRQ);
goto out;
}
該函數的流程如下：
1. 周遊softnet_data 的輪詢清單sd->poll_list，并取出其中的napi struct，獲得napi裝置的weight和poll函數。
2. 如果napi裝置的狀态為被排程（NAPI_STATE_SCHED），則調用poll函數進行實際的收包，poll函數傳回實際收包個數，根據這個傳回值，會有不同的動作：
a) 如果收包個數小于weight，說明收包已經完成，則将該napi裝置從輪詢清單中删除（在poll函數中完成，是以這裡代碼中看不到），然後繼續周遊softnet_data 的輪詢清單。
b) 如果收包個數等于weight，則可能還有資料包沒收完，則将該napi裝置移到輪詢清單的末尾，使之後續還能周遊到。然後繼續周遊softnet_data 的輪詢清單。
将napi裝置從輪詢清單中删除是在函數napi_complete()中完成的，它除了從softnet_data的輪詢清單sd->poll_list中删除napi裝置，還将該裝置的state的NAPI_STATE_SCHED位清除。
napi的state有三種，
1. NAPI_STATE_SCHED：napi裝置是否被排程了，1：被排程了，0：沒有被排程。
2. NAPI_STATE_DISABLE：napi裝置是否被屏蔽了，如果被屏蔽，則不能被排程。
3. NAPI_STATE_NPSVC：netpoll機制中使用，我們不關注。
在net_rx_action()函數中還對每一次軟中斷處理的時間做了限制，這是由兩個變量來控制的：
1. time_limit = jiffies + 2，如果目前時間超過了time_limit，就強制終止此次軟中斷處理。即時間不能超過2個jiffies。
2. budget = netdev_budget，每次poll函數傳回，budget就減去此次收包數，當budget減到0時，就強制終止此次軟中斷處理。netdev_budget設定的值為300。
強制終止此次軟中斷處理并不是不處理了，這是為了與其他任務公平運作，net_rx_action會主動釋放CPU，當然softnet_data中很可能還有沒輪詢到的napi裝置，是以，net_rx_action()重新排程NET_RX_SOFTIRQ軟中斷，讓核心後面有時間再進行處理。
另外需要注意的是，在操作softnet_data的時候需要關閉本地中斷，而在進行軟中斷處理時，是開中斷的。
poll函數用于實際的核心收包，在不使用NAPI機制時，softnet_data的poll函數固定為process_backlog()，他接受兩個參數：napi執行個體和weight（即下面函數參數中的quota）。
static int process_backlog(struct napi_struct *napi, int quota)
{
int work = 0;
/* 獲得softnet_data結構*/
struct softnet_data *sd = container_of(napi,struct softnet_data, backlog);
napi->weight = weight_p;
/* 操作softnet_data時關中斷*/
local_irq_disable();
while (1) {
struct sk_buff *skb;
/* 收取process_queue隊列上的包 */
while ((skb =__skb_dequeue(&sd->process_queue))) {
local_irq_enable(); /* 開中斷 */
__netif_receive_skb(skb); /* 收包入口 */
local_irq_disable(); /* 操作softnet_data時關中斷 */
input_queue_head_incr(sd);
/* 超過weight，則結束收包，傳回收包個數。 */
if (++work >= quota) {
local_irq_enable();
return work;
}
}
rps_lock(sd);
/* 如果隊列為空，從輪詢清單中删除該napi */
if(skb_queue_empty(&sd->input_pkt_queue)) {
list_del(&napi->poll_list);
napi->state = 0;
rps_unlock(sd);
break;
}
/* input_pkt_queue隊列上的包放到process_queue隊列上 */
skb_queue_splice_tail_init(&sd->input_pkt_queue,
&sd->process_queue);
rps_unlock(sd);
}
local_irq_enable();
return work;
}
該函數就是從input_pkt_queue隊列上拿包，然後交給__netif_receive_skb()處理，即我們最開始說的協定棧收包入口。當收包數量超過napi設定的weight，就結束該函數并傳回收包數。
在處理收包隊列時，實際上每次操作的都是process_queue隊列，input_pkt_queue隊列上的包也是放到process_queue隊列中再處理的。process_queue隊列不知道什麼時候加到核心中去的，好像是為了收取offline CPU的資料包。
至此，__netif_receive_skb()收到包了，我們講netif_rx函數就先告一段落。接下來看看NAPI機制下的收包流程有什麼不同。
4. NAPI機制
上面講到的netif_rx函數在收包過程中已經用到了napi_strcut結構，因為軟中斷處理使用了NAPI的架構，本章講述NAPI機制的工作流程，你會發現，軟中斷處理過程和上面講到的沒什麼差别。
對了，NAPI是New API的縮寫，即處理入口幀的一套新API，雖然這個名字沒什麼可擴充性，但是NAPI估計能撐很長時間，是以在更new的API出來之前，暫時不用考慮給它換名字。
NAPI機制采用中斷和輪詢結合的方式收包，防止收包中斷太多處理不過來。傳統的API是每收到一個包就産生一個中斷，在與高速網絡擴充卡協作時，就會遇到在處理一個中斷時另一個中斷已經來了，而進行中斷過程是關中斷的，那新的中斷就會被阻塞。
NAPI使用了IRQ和輪詢的組合。假設資料分組将以高頻率頻繁到達，NAPI的工作機制如下：
1）第一個分組将導緻網絡擴充卡發出IRQ，為防止進一步的分組導緻更多的IRQ，驅動程式會關閉該擴充卡的rx IRQ，并将該擴充卡放到一個輪詢表上。
2）隻要擴充卡上還有分組需要處理，核心就一直對輪詢表上的裝置進行輪詢，處理剩下的分組。
3）重新啟動rx IRQ。
如果在新分組到達時，舊的分組仍然處于處理過程中，工作也不會因額外的中斷而減速。
隻有裝置滿足如下兩個條件，才能實作NAPI方法：
1. 裝置必須能夠保留多個接收的分組，例如儲存到DMA環形緩沖區中。
2. 裝置必須能夠禁止用于接收分組的IRQ，而且發送分組或其他可能通過IRQ進行的操作，都仍然必須是啟用的。
幾乎所有的網卡都是支援DMA模式的，能夠自行将資料傳輸到實體記憶體并通知CPU處理。
初始化一個napi執行個體的函數為netif_napi_add()，就是給napi_struct做初始化工作：
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;
napi->weight = weight;
list_add(&napi->dev_list,&dev->napi_list);
napi->dev = dev;
set_bit(NAPI_STATE_SCHED,&napi->state);//設定NAPI_STATE_SCHED标記
}
EXPORT_SYMBOL(netif_napi_add);
該函數接受四個參數：
1. dev：napi執行個體所屬的裝置。
2. napi：将要做初始化的napi執行個體。
3. poll：napi的poll函數。支援NAPI必須提供一個poll函數。
4. weight：napi的權重值。可以取任何值，但不能超過該裝置可以在rx緩沖區中存儲的分組的數目，通常10/100Mbit網卡驅動指定為16，而1000/10000Mbit網卡驅動指定為64。
做了初始化的成員我們上面都講到過了，在此不贅述。netif_napi_add()函數的目的就是完成一個napi_struct結構執行個體的初始化，後續将被添加到輪詢清單中，通常在網卡驅動的xxx_probe()階段被調用。
下面我們從裝置驅動的收包開始談NAPI的工作機制：
在裝置收包一個包時，驅動中注冊的收包中斷處理函數被執行，中斷處理函數中不同做太多事情，實際上隻需要做兩件事：
1. 關閉裝置的收包中斷。
2. 調用napi_schedule(napi)函數将我們初始化好的napi對象注冊到輪詢清單中，并排程軟中斷。
關閉裝置中斷後，裝置收到包後不再産生中斷（或者核心不再響應中斷），而隻是将資料包放到DMA中。
napi_schedule()的實作如下：
static inline void napi_schedule(struct napi_struct *n)
{
if (napi_schedule_prep(n))
__napi_schedule(n);
}
napi_schedule_prep()先做一些檢查工作：如果napi對象的狀态為NAPI_STATE_DISABLE或已經是NAPI_STATE_SCHED，則不進行排程，如果沒有設定NAPI_STATE_SCHED标記，則置上NAPI_STATE_SCHED标記。
接下來__napi_schedule()就是添加到目前CPU的softnet_data結構的poll_list輪詢清單中（這裡由于操作了softnet_data，是以要關一下中斷），并排程NET_RX_SOFTIRQ軟中斷（對應前面，開中斷）。
排程NET_RX_SOFTIRQ軟中斷後，核心後續會去執行處理函數net_rx_action()。這個函數的流程我們已經講過，和前面唯一的不同就在于poll函數，在非NAPI收包過程中，poll函數是在netif_rx()中注冊的process_backlog()函數，而NAPI收包中的poll函數是我們在netif_napi_add()中注冊的，即自定義的一個函數。
接下來就看一下一個實際的NAPI裝置收包的poll函數都是怎麼實作的，我不帖特定的代碼，隻是大緻說一下流程：
1. 進行收包，這裡收包并不是從softnet_data的某個隊列收包，由于CPU已經不接受裝置的收包中斷了，是以在DMA中可能會積壓了一些包，是以直接從DMA的緩沖區中收包，并交給netif_receive_skb()進入協定棧。每次收包的數量由napi對象的weight權值限制。
2. 如果收包個數小于weight的值，說明全收完了。則調用napi_complete()将napi對象從輪詢清單中删除，并清除其NAPI_STATE_SCHED位，同時開啟裝置的收包中斷，傳回收包個數到net_rx_action。
3. 如果收包個數大于等于weight的值，說明可能還沒收完，則傳回收包個數到net_rx_action。
也就是說，在關閉收包中斷的情況下，napi的poll函數會去不停的從DMA中收包，直到收完才開中斷，開中斷後的下一個包就以中斷的方式通知CPU收包。當然中斷不能長時間關閉，前面講到在net_rx_action設定了每次軟中斷的時間限制。
講到這裡我們需要對比一下直接使用netif_rx收包和使用NAPI收包的差別：
netif_rx收包
NAPI收包
從圖中擷取資料包的方式就可以看出NAPI相對于單純的netif_rx的優勢。為什麼說單純的netif_rx呢。因為，目前還有很多不使用NAPI收包的裝置驅動，這些驅動可以采用其他類似NAPI的方法，來緩解高吞吐量下的中斷風暴。
5. 在中斷期間處理多幀
一些驅動雖然沒有使用NAPI收包機制，但在驅動中通過設定類似weight的權值，實作在一個中斷到來時嘗試處理多個資料包。
例如，有些驅動在中斷處理程式中添加了一個quota值，限定每次中斷可以處理資料包的個數，在每次中斷到來時關閉裝置自身的收包中斷，并嘗試從DMA中擷取不大于quota數量的資料包，每次擷取到資料包就交給netif_rx處理或直接交給netif_receive_skb()。當然，拿包并處理的過程可能比較長，那麼可以将這些動作放到tasklet任務中，中斷處理程式隻需排程tasklet任務即可。在處理完quota個資料包之後再開啟裝置的收包中斷。
這樣一來，使用quota結合netif_rx，就實作了在一次中斷中處理多個包。