Linux协议栈--NAPI机制分析_netif_napi_add

内核版本：2.6.37

1 linux协议栈收包概述

NAPI是linux新的网卡数据处理API，据说是由于找不到更好的名字，所以就叫NAPI(New API)，在2.5之后引入。

简单来说，NAPI是综合中断方式与轮询方式的技术。

中断的好处是响应及时，如果数据量较小，则不会占用太多的CPU事件；缺点是数据量大时，会产生过多中断，而每个中断都要消耗不少的CPU时间，从而导致效率反而不如轮询高。轮询方式与中断方式相反，它更适合处理大量数据，因为每次轮询不需要消耗过多的CPU时间；缺点是即使只接收很少数据或不接收数据时，也要占用CPU时间。

NAPI是两者的结合，数据量低时采用中断，数据量高时采用轮询。平时是中断方式，当有数据到达时，会触发中断处理函数执行，中断处理函数关闭中断开始处理。如果此时有数据到达，则没必要再触发中断了，因为中断处理函数中会轮询处理数据，直到没有新数据时才打开中断。

很明显，数据量很低与很高时，NAPI可以发挥中断与轮询方式的优点，性能较好。如果数据量不稳定，且说高不高说低不低，则NAPI则会在两种方式切换上消耗不少时间，效率反而较低一些。

1.1 非NAPI帧的接收

我们将讨论内核在接收一个数据帧后的大致处理流程，不会详细叙述所有细节。
我们认为有必要先了解一下传统的数据包处理流程以便更好的理解NAPI和传统收包方式的区别。
在传统的收包方式中（如下图）数据帧向网络协议栈中传递发生在 中断上下文（在接收数据帧时）中调用netif_rx的函数中。 这个函数还有一个变体netif_rx_ni，他被用于中断上下文之外。

netif_rx函数将网卡中收到的数据包（包装在一个socket buffer中）放到系统中的接收队列中（input_pkt_queue），前提是这个接收队列的长度没有大于netdev_max_backlog。这个参数和另外一些参数可以在/proc文件系统中看到（/proc/sys/net/core文件中，可以手动调整这个数值）。

softnet_data结构体的定义：

/*
 * Incoming packets are placed on per-cpu queues so that
 * no locking is needed.
 */
struct softnet_data
{
    struct net_device *output_queue;
    struct sk_buff_head input_pkt_queue;
    struct list_head poll_list;
    struct sk_buff *completion_queue;
    struct net_device backlog_dev;
#ifdef CONFIG_NET_DMA
    struct dma_chan *net_dma;
#endif
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

input_pkt_queue是softnet_data结构体中的一个成员，定义在netdevice.h文件中。
如果接收到的数据包没有因为input_pkt_queue队列已满而被丢弃，它会被netif_rx_schedule函数调度给软中断NET_RX_SOFTIRQ处理，netif_rx_schedule函数在netif_rx函数内部被调用。
软中断NET_RX_SOFTIRQ的处理逻辑在net_rx_action函数中实现。
此时，我们可以说此函数将数据包从input_pkt_queue队列中传递给了网络协议栈，现在数据包可以被处理了。

1.2 NAPI帧的接收

在NAPI架构中（如下图），当接收到数据包产生中断时，驱动程序会通知网络子系统有新的数据包到来（而不是立即处理数据包），这样就可以在ISR（Interrupt Service Routines - 中断服务程序）上下文之外使用轮询的方式来一次性接收多个数据包。

因此网卡支持NAPI必须满足几个条件：驱动程序不再使用数据包接收队列，网卡本身需要维护一个缓冲区来保存接收到数据包，并且可以禁止中断。
这种方法减少了中断的产生并且在突发情况下减少了丢包的可能性，避免了接收队列的饱和。
从NAPI实现的角度来看，与传统收包方式的不同地方在中断程序和轮询函数上（在net_device结构体中定义），定义如下：

int (*poll)(struct net_device *dev, int *budget);
1

除此之外，net_device结构体中还有另外两个属性quota和weight，他们用于在一个轮询周期中实现抢占机制（译者注：意思是通过这两个参数来控制一个轮询周期的运行时间）我们将在后面详细讨论。
NAPI模型中的中断函数将数据帧传送到协议栈的任务交给poll函数执行。 换句话说中断函数的工作被简化为禁用网络设备中断（再此期间设备可以继续接收数据帧），和确认中断然后调度（通过netif_rx_schedule函数调度）软中断NET_RX_SOFTIRQ关联的net_rx_action函数。
等待被轮询的设备通过netif_rx_schedule函数将net_device结构体实例的指针加入到poll_list链表中。 在调用net_rx_action函数执行软中断NET_RX_SOFTIRQ时会遍历poll_list链表，然后调用每个设备的poll()函数将数据帧存放在socket buffers中并通知上层协议栈。

net_rx_action函数的执行步骤如下：

1. 回收当前处理器的poll_list链表的引用。
2. 将jiffies的值保存在start_time变量中。
3. 设置轮询的budget（预算，可处理的数据包数量）为netdev_budget变量的初始值（这个值可以通过/proc/sys/net/core/netdev_budget来配置）
4. 轮询poll_list链表中的每个设备，直到你的budget用完，当你的运行时间还没有超过一个jiffies时:
   • 如果quantum（配额）为正值则调用设备的poll()函数，否则将weight的值加到quantum中，将设备放回poll_list链表；
   • 如果poll()函数返回一个非零值，将weight的值设置到quantum中然后将设备放回poll_list链表；
   • 如果poll()函数返回零值，说明设备已经被移除poll_list链表（不再处于轮询状态）。

budget的值和net_device结构体的指针会传递到poll()函数中。poll()函数应该根据数据帧的处理数量来减小budget的值。数据帧从网络设备的缓冲区中复制出来包装在socket buffers中，然后通过netif_receive_skb函数传递到协议栈中去。

抢占策略是依赖budget变量的配额机制实现的：poll()函数必须根据分配给设备的最大配额来决定可以传递多少个数据包给内核。 当配额使用完就不允许在传递数据包给内核了，应该轮询poll_list链表中的下一个设备了。因此poll()必须和减小budget的值一样根据数据帧的处理数量来减小quota的值。

如果驱动在用完了所有的quota之后还没有传递完队列中所有的数据包，poll()函数必须停止运行并返回一个非NULL值。
如果所有数据包都传递到了协议栈，驱动程序必须再次使能设备的中断并停止轮询，然后调用netif_rx_complete函数（它会将设备从poll_list链表去除），最后停止运行并返回零值给调用者（net_rx_action函数）。

net_device结构体中的另一个重要成员weight，它用于每次调用poll()函数时重置quota的值。 很明显weight的值必须被初始化为一个固定的正值。通常对于高速网卡这个值一般在16和32之间，对于千兆网卡来说这个值会大一点（通常时64）。
从net_rx_action函数的实现中我们可以看到当weight的值设置太大时，驱动使用的budget会超过quantum，此时会导致一个轮询周期的时间变长。

我们给出了设备驱动程序接收中断并执行轮询函数的伪代码：

static irqreturn_t sample_netdev_intr(int irq, void *dev)
{
    struct net_device *netdev = dev;
    struct nic *nic = netdev_priv(netdev);

    if (! nic->irq_pending())
        return IRQ_NONE;

    /* Ack interrupt(s) */
    nic->ack_irq();

    nic->disable_irq();  

    netif_rx_schedule(netdev);

    return IRQ_HANDLED;
}

 
static int sample_netdev_poll(struct net_device *netdev, int *budget)
{
    struct nic *nic = netdev_priv(netdev);

    unsigned int work_to_do = min(netdev->quota, *budget);
    unsigned int work_done = 0;

    nic->announce(&work_done, work_to_do);

    /* If no Rx announce was done, exit polling state. */

    if(work_done == 0) || !netif_running(netdev)) {

    netif_rx_complete(netdev);
    nic->enable_irq();  

    return 0;
    }

    *budget -= work_done;
    netdev->quota -= work_done;

    return 1;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

下图展示了非NAPI数据包接收处理过程的时序图：

下图展示了NAPI数据包接收处理过程的时序图：

2 实现

来看下NAPI和非NAPI的区别：

(1) 支持NAPI的网卡驱动必须提供轮询方法poll()。

(2) 非NAPI的内核接口为netif_rx()，NAPI的内核接口为napi_schedule()。

(3) 非NAPI使用共享的CPU队列softnet_data->input_pkt_queue，NAPI使用设备内存(或者设备驱动程序的接收环)。

2.1 NAPI设备结构

/* Structure for NAPI scheduling similar to tasklet but with weighting */
 
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; /* 每次处理的最大数量，非NAPI默认为64 */
    int (*poll) (struct napi_struct *, int); /* 此设备的轮询方法，非NAPI为process_backlog() */
 
#ifdef CONFIG_NETPOLL
    ...
#endif
 
    unsigned int gro_count;
    struct net_device *dev;
    struct list_head dev_list;
    struct sk_buff *gro_list;
    struct sk_buff *skb;
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

2.2 初始化

初始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; /* 设备的poll函数 */
    napi->weight = weight; /* 设备每次poll能处理的数据包个数上限 */
 
    list_add(&napi->dev_list, &dev->napi_list); /* 加入设备的napi_list */
    napi->dev = dev; /* 所属设备 */
 
#ifdef CONFIG_NETPOLL
    spin_lock_init(&napi->poll_lock);
    napi->poll_owner = -1;
#endif
    set_bit(NAPI_STATE_SCHED, &napi->state); /* 设置NAPI标志位 */
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

2.3 调度

在网卡驱动的中断处理函数中调用napi_schedule()来使用NAPI。

/**
 * napi_schedule - schedule NAPI poll
 * @n: napi context
 * Schedule NAPI poll routine to be called if it is not already running.
 */
 
static inline void napi_schedule(struct napi_struct *n)
{
    /* 判断是否可以调度NAPI */
    if (napi_schedule_prep(n))
        __napi_schedule(n);
}
1
2
3
4
5
6
7
8
9
10
11
12

判断NAPI是否可以调度。如果NAPI没有被禁止，且不存在已被调度的NAPI，

则允许调度NAPI，因为同一时刻只允许有一个NAPI poll instance。

/**
 * napi_schedule_prep - check if napi can be scheduled
 * @n: napi context
 * Test if NAPI routine is already running, and if not mark it as running.
 * This is used as a condition variable insure only one NAPI poll instance runs.
 * We also make sure there is no pending NAPI disable.
 */
 
static inline int napi_schedule_prep(struct napi_struct *n)
{
    return !napi_disable_pending(n) && !test_and_set_bit(NAPI_STATE_SCHED, &n->state);
}
 
static inline int napi_disable_pending(struct napi_struct *n)
{
    return test_bit(NAPI_STATE_DISABLE, &n->state);
} 
 
enum {
    NAPI_STATE_SCHED, /* Poll is scheduled */
    NAPI_STATE_DISABLE, /* Disable pending */
    NAPI_STATE_NPSVC, /* Netpoll - don't dequeue from poll_list */
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

NAPI的调度函数。把设备的napi_struct实例添加到当前CPU的softnet_data的poll_list中，

以便于接下来进行轮询。然后设置NET_RX_SOFTIRQ标志位来触发软中断。

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);
}
 
static inline void ____napi_schedule(struct softnet_data *sd, struct napi_struct *napi)
{
    /* 把napi_struct添加到softnet_data的poll_list中 */
    list_add_tail(&napi->poll_list, &sd->poll_list);
    __raise_softirq_irqoff(NET_RX_SOFTIRQ); /* 设置软中断标志位 */
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2.4 轮询方法

NAPI方式中的POLL方法由驱动程序提供，在通过netif_napi_add()加入napi_struct时指定。

在驱动的poll()中，从自身的队列中获取sk_buff后，如果网卡开启了GRO，则会调用

napi_gro_receive()处理skb，否则直接调用netif_receive_skb()。

POLL方法应该和process_backlog()大体一致，多了一些具体设备相关的部分。

2.5 非NAPI和NAPI处理流程对比

以下是非NAPI设备和NAPI设备的数据包接收流程对比图：

NAPI方式在上半部中sk_buff是存储在驱动自身的队列中的，软中断处理过程中驱动POLL方法调用

netif_receive_skb()直接处理skb并提交给上层。

/**
 * netif_receive_skb - process receive buffer from network
 * @skb: buffer to process
 * netif_receive_skb() is the main receive data processing function.
 * It always succeeds. The buffer may be dropped during processing
 * for congestion control or by the protocol layers.
 * This function may only be called from softirq context and interrupts
 * should be enabled.
 * Return values (usually ignored):
 * NET_RX_SUCCESS: no congestion
 * NET_RX_DROP: packet was dropped
 */
 
int netif_receive_skb(struct sk_buff *skb)
{
    /* 记录接收时间到skb->tstamp */
    if (netdev_tstamp_prequeue)
        net_timestamp_check(skb);
 
    if (skb_defer_rx_timestamp(skb))
        return NET_RX_SUCCESS;
 
#ifdef CONFIG_RPS
    ...
#else
    return __netif_receive_skb(skb);
#endif
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

__netif_receive_skb()在####中已分析过了，接下来就是网络层来处理接收到的数据包了。

3 简单使用

对NAPI 的使用，一般包括以下的几个步骤：

1、在中断处理函数中，先禁止接收中断，且告诉网络子系统，将以轮询方式快速收包，其中禁止接收中断完全由硬件功能决定，而告诉内核将以轮询方式处理包则是使用函数 netif_rx_schedule()：

void netif_rx_schedule(struct net_device *dev);
1

也可以使用下面的方式，其中的 netif_rx_schedule_prep 是为了判定现在是否已经进入了轮询模式:

if (netif_rx_schedule_prep(dev))
	__netif_rx_schedule(dev);
1
2

2、在驱动中创建轮询函数，它的工作是从网卡获取数据包并将其送入到网络子系统，其原型是：

NAPI 的轮询方法

int (*poll)(struct net_device *dev, int *budget);
1

这里的轮询函数用于在将网卡切换为轮询模式之后，用 poll() 方法处理接收队列中的数据包，如队列为空，则重新切换为中断模式。切换回中断模式需要先关闭轮询模式，使用的是函数 netif_rx_complete ()，接着开启网卡接收中断:

退出轮询模式

void netif_rx_complete(struct net_device *dev);
1

3、在驱动中创建轮询函数，需要和实际的网络设备 struct net_device 关联起来，这一般在网卡的初始化时候完成，示例代码如下：

设置网卡支持轮询模式

dev->poll = my_poll;
dev->weight = 64; // 权重 (weight)，该值并没有一个非常严格的要求，实际上是个经验数据，一般 10Mb 的网卡，我们设置为 16，而更快的网卡，我们则设置为 64。
1
2

https://blog.csdn.net/zhangskd/article/details/21627963
http://abcdxyzk.github.io/blog/2015/08/27/kernel-net-napi/
http://cxd2014.github.io/2017/10/15/linux-napi/