Linux内核源码分析--详谈NAPI原理机制

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，就实现了在一次中断中处理多个包。