報文的接收是整個協議棧的入口，負責從網卡中把報文接收並送往內核協議棧相應協議處理模塊處理。
一種是網卡產生中斷，通知內核進行接收報文。一次中斷接收一個報文。在中斷處理程序中把報文從硬件緩存中拷貝到內存中，並把報文加入到協議棧中對應的入口隊列中，中斷退出時調用收包軟中斷來從相應隊列來讀取報文進行處理。這種方式優點是內核對報文響應較快，在網卡上 有少量報文時效果較好。這樣如果網卡有大量報文的話，會產生大量中斷。中斷會不斷打斷處理報文的軟中斷，這樣先前的報文來不及處理，總是被打斷，隊列裏的報文大量堆積導致不能被及時處理，導致報文性能下降。
另一種是中斷加輪詢的方式進行接收報文的。當一個報文到來時，產生中斷通知內核，註冊的中斷處理程序首先禁止網卡的中斷，內核開始輪詢的從網卡硬件緩存或網卡硬件管理的內存隊列中讀取報文並處理，直到報文讀完或超出了設置的允許讀取最大報文個數，輪詢函數停止處理報文。如果這次輪詢處理完了全部報文，並接著使能網卡中斷。如果輪詢沒有處理完全部報文，這時就重新調度軟中斷，等下次軟中斷被調度到後繼續輪詢處理報文。這樣是為了防止輪詢獨占CPU時間。以後重復這樣的接收報文步驟。這樣內核處理一次中斷可以處理多個報文，這樣就避免了產生大量中斷造成的系統開銷。但支持這樣的操作的網卡必須支持硬件能自動把報文存入內存中，或硬件緩存足夠大。否則當一次輪詢結束後並沒有處理完全部的報文時，到下次軟中斷被調度到開始輪詢的這段時間裏，當網卡硬件內部的緩存滿後會產生丟包。
Linux 提供對兩種報文接收方式的支持，供不同網卡驅動來用。在本文中把第一種稱為舊的方式（舊是相對於NAPI來說的），第二種方法linux內核中稱為NAPI方式。
NAPI的另一個優點是設備處理更為公平，因為軟中斷順序執行每個需要處理的設備的輪詢函數，每個輪詢函數執行時間是有限制的，超過最大的執行時間後就會執行下一個設備的輪詢函數。這樣就保證負載低的和負載高的設備被處理的機會是一樣。
Linux 內核協議棧中報文接收的設計思路分析如下：
NAPI接口和舊接口兩者有一下相同點：
（1）、對報文的處理都是放在軟中斷中處理。
（2）、兩者都有存儲報文的隊列，NAPI的隊列是由網卡驅動來管理的（一般是有網卡DMA到內存中，網卡驅動管理著存儲報文的隊列），而舊接口的隊列是由Linux內核管理的，由網卡驅動負責往隊列裏加報文。
每個NAPI設備都有一個輪詢函數來由收包軟中斷調用，來進行輪詢處理報文。我們可以建立一個虛擬的NAPI設備（backlog)，讓她的輪詢函數來輪詢的處理舊接口的報文隊列。收包軟中斷處理函數只是進行調用相應NAPI的輪詢函數進行處理，並不關心是虛擬NAPI還是非虛擬NAPI。具體輪詢細節由相關NAPI的輪詢函數自己實現。這樣舊接口和NAPI就可以很好的融合在一起來實現了。
如上所述，我們現在可以知道每個CPU上需要有如下幾個元素：
1、一個報文的接收隊列，由舊接口來使用。
2、一個虛擬的NAPI設備，來有Linux協議棧自己創建並處理舊接口使用的隊列中的報文。
3、一個NAPI的鏈表，上面掛著有報文需要處理的NAPI設備，由收包軟中斷來遍歷該鏈表，順序執行每個NAPI的輪詢函數。
Linux內核具體實現：
結構體定義如下
structsoftnet_data
{
struct sk_buff_head input_pkt_queue;//舊接口的輸入隊列
struct list_head poll_list;//有需要處理報文的NAPI設備鏈表
struct napi_struct backlog;//虛擬的NAPI設備 backlog
};
定義一個per_cpu變量 softnet_data
DECLARE_PER_CPU(struct softnet_data,softnet_data);
softnet_data 的初始化如下：
static int __init net_dev_init(void)
{
int i, rc = -ENOMEM;
/*
* Initialise the packet receive queues.
*/
for_each_possible_cpu(i)
{
struct softnet_data *queue;
queue = &per_cpu(softnet_data, i);
/*初始化輸入隊列*/
skb_queue_head_init(&queue->input_pkt_queue);
queue->completion_queue = NULL;
/*初始化pool 鏈表頭*/
INIT_LIST_HEAD(&queue->poll_list);
/*把backlog的輪詢函數初始化為 process_backlog
該函數來處理傳統接口使用的輸入隊列的報文*/
queue->backlog.poll = process_backlog;
/*backlog 輪詢函數一次可以處理的報文上限個數*/
queue->backlog.weight = weight_p;
queue->backlog.gro_list = NULL;
queue->backlog.gro_count = 0;
}
}
NAPI的數據結構：
每個NAPI需要如下幾個元素：
1、一個輪詢函數，來輪詢處理報文。
2、需要有一個狀態，來標識NAPI的調度狀態，是正在被調度還是沒有被調度。
3、每個NAPI的一次輪詢能處理的報文個數應該有一個上限，不能無限制的處理下去，防止獨占CPU資源導致其他進程被餓死。
4、每個NAPI應該和一個網絡設備進行關聯。
5、NAPI設計時考慮了多隊列的情況。一個網絡設備可以有多個報文隊列，這樣一個網絡設備可以關聯多個NAPI，每個NAPI只處理特定的隊列的報文。這樣在多核情況下，多個CPU核可以並行的進行報文的處理。一般專用的網絡多核處理器存在這種情況。
6、每個NAPI在有報文的情況下應該掛到softnet_data的pool_list上去。
如上所述，NAPI的結構體定義如下
struct napi_struct
{
struct list_head poll_list;//掛到softnet_data的pool_list上
unsigned long state;//NAPI的調度狀態
int weight;//一次輪詢的最大處理報文數
int (*poll)(struct napi_struct *, int);//輪詢函數
struct net_device *dev;//指向關聯的網絡設備
struct list_head dev_list;//對應的網絡設備上關聯的NAPI鏈表節點
/*其他字段是gso功能用，這裏先不討論*/
};
NAPI的調度狀態：
NAPI_STATE_SCHED設置時表示該NAPI有報文 需要接收。即把NAPI掛到softnet_data 時要設置該狀態，處理完從softnet_data 上摘除該NAPI時要清除該狀態。
一些NAPI的函數詳解：
1、netif_napi_add（）,把網絡設備net_device 和NAPI結構相綁定。
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->poll = poll;
napi->weight = weight;
/*把NAPI加入到網絡設備相關聯的NAPI鏈表上去。*/
list_add(&napi->dev_list, &dev->napi_list);
napi->dev = dev;
/*綁定時設置NAPI是已調度狀態，禁用該NAPI,以後手動的來清除該標識來使
能NAPI.*/
set_bit(NAPI_STATE_SCHED, &napi->state);
}
2、使能和禁用NAPI：
static inline void napi_disable(struct napi_struct *n)
{
/*先設置NAPI狀態為DISABLE*/
set_bit(NAPI_STATE_DISABLE, &n->state);
/*循環的等待NAPI被調度完成,變成可用的，設置成SCHED狀態*/
while (test_and_set_bit(NAPI_STATE_SCHED, &n->state))
msleep(1);
/*清除DISABLE狀態*/
clear_bit(NAPI_STATE_DISABLE, &n->state);
}
3、調度NAPI
void __napi_schedule(struct napi_struct *n)
{
unsigned long flags;
/*鏈表操作必須在關閉本地中斷的情況下操作，防止硬中斷搶占*/
local_irq_save(flags);
/*把NAPI加入到本地CPU的softnet_data 的pool_list 鏈表上*/
list_add_tail(&n->poll_list, &__get_cpu_var(softnet_data).poll_list);
/*調度收包軟中斷,詳見軟中斷的實現*/
__raise_softirq_irqoff(NET_RX_SOFTIRQ);
local_irq_restore(flags);
}
4、NAPI執行完畢，如果NAPI一次輪詢處理完隊列的所以報文，調用該函數。
void __napi_complete(struct napi_struct *n)
{
BUG_ON(!test_bit(NAPI_STATE_SCHED, &n->state));
BUG_ON(n->gro_list);
/*把NAPI從softnet_data的pool_list上摘除下來*/
list_del(&n->poll_list);
/*使能NAPI，允許它下次可以被再次調度*/
smp_mb__before_clear_bit();
clear_bit(NAPI_STATE_SCHED, &n->state);
}
一般網卡驅動的NAPI的步驟：
1、不同的網卡驅動都會為網絡設備定義自己的結構體。有的是自己的結構體中即包括net_device和napi,有的是把自己的結構體放到net_device的priv部分。
2、實現NAPI的輪詢函數 rx_pool(struct napi_struct *napi，int weigh);
int rx_pool_xxx(struct napi_struct *napi,int weigh)
{
int rx_cnt = 0;
struct sk_buff *skb;
while(rx_cnt < weigh)
{
skb = netdev_alloc_skb();
copy_skb_form_hw(skb);/*把報文從硬件緩存中讀到內存skb中*/
rx_cnt ++;
netif_receive_skb(skb);//送協議棧直接處理
}
if(rx_cnt < weigh)
{
/*處理報文小與允許處理最大個數，表示一次輪詢處理完全部的報文。使能收包中斷*/
napi_complete(napi)
enable_rx_irq();
}
else
{
/*否則輪詢超時後沒有處理完該處理的報文，就不用使能中斷，也不用把NAPI從鏈表上摘下來。直接返回即可*/
}
return rx_cnt;
}
3、把網絡設備跟NAPI相關聯，netif_napi_add(netdev,napi,rx_pool_xxx,weigh)；
4、註冊網卡的收包中斷：例：
request_irq(irq,&rx_irq_handle, 0, netdev->name, netdev);
在中斷處理函數中,把對應的NAPI加入到內核管理的鏈表中。
static irqreturn_t rx_irq_handle(int irq, void *data)
{
struct net_device *netdev = data
struct napi_struct *napi = netdev_priv(netdev)->napi;
disable_rx_irq();
__napi_schedule(napi);
}
網卡使用舊接口的方法：
在網卡收包中斷處理程序中，把報文從網卡硬件緩存中拷貝到內存後，初始化好skb結構體，直接調用
netif_rx(skb)把報文加入到內核管理的隊列中即可。
留言列表
